·
编程学习 ·
177147
- kafka消费中的问题及解决方法：
- 情况1:
- 问题：脚本读取kafka 数据，写入到数据库，有时候出现MySQL server has gone away，导致脚本死掉。再次启动，这过程中的kafka数据丢失。
- 原因：MySQL server has gone away 出现可能是连接超时，可能超过每秒请求上限…这些异常是小概率事件，难以避免。git kafka 的demo脚本是实时监听的脚本, 简单明了，没有再去针对kafka偏移量研究；但是一旦断掉, 过程中的kafka数据即丢失。
- 解决思路MySQL server has gone away无法避免，try catch 即可,异常消息存入日志中。通过数据补偿脚本把日志中的数据存入数据库，做到正常 异常的数据最终都能录入数据库。且脚本也不会异常死掉。脚本其他原因死掉(如服务器宕机)，通过集群控制风险。
- 情况2:
- 问题：kafka消费时做数据检验，数据处理。代码看的很混乱, 且一旦业务调整或者有bug意味着要停止消费后重启。风险高，维护困难。
- 解决思路：解藕。kafka消费脚本只负责把数据写入数据库，标识初始状态。业务处理脚本集中处理数据。
BlockingConnection因单线程同步模型导致basic_consume阻塞于start_consuming()，消息串行处理，受PythonGIL限制无法并行dispatch回调，吞吐受限；须改用SelectConnection等异步适配器获取事件循环控制权，解耦消息分发与业务处理。 单...
RabbitMQ本身保证单条消息只投递一次给一个消费者，数据一致性问题源于业务逻辑并发执行导致的竞态，而非消息重复投递；basic_ack仅确认消息接收，不保障业务执行唯一性，故无法解决数据一致性问题。 多进程消费同一个RabbitMQ队列时，数据一致性问题不来自RabbitMQ本身（它天然保证单条...
数据库连接失效时，RabbitMQ消费者不应直接basic_nack(requeue=True)触发无限重试，而应先重建DB连接、本地带退避重试（如tenacity指数退避3次），失败后basic_nack(requeue=False)入死信队列，并确保队列已正确绑定DLX。 数据库连接失效时，Ra...
aiokafka是Python唯一成熟原生异步Kafka消费方案，基于kafka-python重写，全I/O非阻塞；confluent-kafka-python不适配asyncio，因其依赖后台线程poll且回调无法被事件循环感知。 aiokafka是目前Python生态中唯一成熟、可直接用于生产环...
当仅有一个订阅者连接到具有多个分区的pub/sublite主题时，该订阅者会自动从所有分区拉取消息，不会导致消息堆积；分区负载由客户端库自动均衡处理。 当仅有一个订阅者连接到具有多个分区的pub/sublite主题时，该订阅者会自动从所有分区拉取消息，不会导致消息堆积；分区负载由客户端库自动均衡处理...
本文介绍如何改造传统的单条数据生产/消费模式，让producer批量入队、consumer批量出队，以满足机器学习模型对批量样本（如2000行）的输入需求，显著提升i/o与计算协同效率。 本文介绍如何改造传统的单条数据生产/消费模式，让producer批量入队、consumer批量出队，以满足机器学...
消费端必须考虑幂等性，因为消息队列通常只保证“至少一次”投递，网络抖动、重启等会导致重复消费，若无幂等控制易引发重复扣款、超卖等问题；常见方案包括消息ID+数据库去重、RedisSETNX、业务状态机校验、唯一约束+乐观锁。 为什么消费端必须考虑幂等性 消息队列（如RabbitMQ、Kafka、Ro...
根本原因是消费者未正确确认消息，导致中间件重复投递：rabbitmq需显式basic_ack()，kafka需正确commit()且处理异常，redis无ack机制需自行实现去重与持久化。 为什么consuming多次但只收到一条消息？ 根本原因通常是消费者没确认（ack），或确认方式不对，导致消息...
