Kafka

kafka consumer 按照 Kafka 默认的消费逻辑设定，一个分区只能被同一个消费组（ConsumerGroup）内的一个消费者消费。 即同一个partition内的消息只能被同一个组中的一个consumer消费。当消费者数量多于partition的数量时，多余的消费者空闲 assignment, 分配策略 Kafka提供了消费者客户端参数partition.assignment.strategy用来设置消费者与订阅主题之间的分区分配策略。 默认情况下，此参数的值为：org.apache.kafka.clients.consumer.RangeAssignor，即采用RangeAssignor分配策略。除此之外，Kafka中还提供了另外两种分配策略： RoundRobinAssignor和StickyAssignor。消费者客户端参数partition.asssignment.strategy可以配置多个分配策略，彼此之间以逗号分隔。 RangeAssignor RangeAssignor 策略的原理是按照消费者总数和分区总数进行整除运算来获得一个跨度，然后将分区按照跨度进行平均分配，以保证分区尽可能均匀地分配给所有的消费者。对于每一个topic，RangeAssignor策略会将消费组内所有订阅这个topic的消费者按照名称的字典序排序，然后为每个消费者划分固定的分区范围，如果不够平均分配，那么字典序靠前的消费者会被多分配一个分区。 range assignor 在某些情况下会分配不均匀, 有可能会出现部分消费者过载的情况. RoundRobinAssignor 把 topic 和 consumer 排序, 依次给 topic 分配 consumer, 同一个 topic 中的不同 partition 分被均匀分配给不同的 consumer. RoundRobinAssignor 策略的原理是将消费组内所有消费者以及消费者所订阅的所有topic的partition按照字典序排序，然后通过轮询消费者方式逐个将分区分配给每个消费者。RoundRobinAssignor策略对应的partition.assignment.strategy参数值为：org.apache.kafka.clients.consumer.RoundRobinAssignor。 如果同一个消费组内所有的消费者的订阅信息都是相同的，那么RoundRobinAssignor策略的分区分配会是均匀的。 如果同一个消费组内的消费者所订阅的Topic 是不相同的，那么在执行分区分配的时候就不是完全的轮询分配，有可能会导致分区分配的不均匀。如果某个消费者没有订阅消费组内的某个topic，那么在分配分区的时候此消费者将分配不到这个topic的任何分区。 StickyAssignor “sticky”这个单词可以翻译为“粘性的”，Kafka从0.11.x版本开始引入这种分配策略，它主要有两个目的： ① 分区的分配要尽可能的均匀； ② 分区的分配尽可能的与上次分配的保持相同。 当两者发生冲突时，第一个目标优先于第二个目标。鉴于这两个目标，StickyAssignor策略的具体实现要比RangeAssignor和RoundRobinAssignor这两种分配策略要复杂很多。 从结果上看StickyAssignor策略比另外两者分配策略而言显得更加的优异，这个策略的代码实现也是异常复杂，如果大家在一个 group 里面，不同的 Consumer 订阅不同的 topic, 那么设置Sticky 分配策略还是很有必要的。 properties.put("enable.auto.commit", "true"); properties.put("auto.commit.interval.ms", "1000"); properties.put("auto.offset.reset", "latest"); properties.put("session.timeout.ms", "30000"); properties.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer"); properties.put("value.deserializer", "org.apache.kafka.common.serialization.ByteArrayDeserializer"); Consumer Group 主要用于实现高伸缩性，高容错性的 Consumer 机制。因此，消息的接收是基于 Consumer Group 的。组内多个 Consumer 实例可以同时读取 Kafka 消息，同一时刻一条消息只能被一个消费者消费，而且一旦某一个 consumer “挂了”， Consumer Group 会立即将已经崩溃的 Consumer 负责的分区转交给其他 Consumer 来负责。从而保证 Consumer Group 能够正常工作。 ...

背景 2020 年前后，我在大连 R 公司负责车联网平台协议解析层的开发和维护。R 平台是一套部署在公有云上的 IoT 平台，接入大量在线车载设备，实时处理车辆上报的行车数据。 那时平台有一项新的合作需求：一家做共享汽车业务的公司（以下简称 M 公司）采购了 R 公司的车载设备，安装在共享汽车上，希望通过接口从 R 平台获取这些车辆的实时行车数据，再通过自己的 App 提供给用户查询车辆状态和行驶记录。 我负责这次集成的开发工作，包括改造行车数据服务的分流逻辑，以及独立开发一个面向 M 公司的数据推送服务。这也不是我第一次做类似的事，R 平台之前也有过给其他第三方提供数据接口的项目，但 M 公司是其中规模比较大的一家。 技术方案 行车数据服务本身改动不大。原来的逻辑是收到设备上报的数据后直接写入平台内部各个下游。这次只是加了一段分流逻辑：根据车辆底盘号（VIN）查询 Redis 中的客户映射关系，如果这辆车属于 M 公司，就把这条数据额外发送到一个专用的 Kafka topic。 Kafka topic 的另一侧是我新写的 Go Kafka consumer，消费消息后做数据格式转换，处理鉴权，再通过 HTTP POST 批量推送给 M 公司的接口。推送时有一个打包策略：收到 N 条数据就打包发出，或者超过一定时间没有新数据也把当前这批发出，每次 HTTP 请求可能包含多条记录。 VIN 到客户的映射关系存储在 Redis，由业务侧维护，数据服务只做读取。 工程细节 可观测性 对接第三方的接口，出问题时的排查成本比内部服务高很多——双方都需要对齐日志，搞清楚是哪一侧的问题。 我在这个服务里做了几件事： 每一批次数据带一个唯一的 trace ID，日志里记录发送时间、trace ID、完整 payload、M 公司 API 的响应时间和状态码。在调试阶段和试运行阶段，日志级别调到 debug，把 payload 全量打印出来，方便排查数据是否正确、格式是否符合对方预期。 除了日志，我还在服务里埋了监控指标，记录推送并发量和调用对方 API 的响应时间，接入 R 平台已有的 Grafana 看板，可以直观看到推送量和延迟的变化趋势。 ...

背景 Committer 服务负责消费 Kafka 消息，将网络策略通过 NSO 提交到物理网络设备。原始设计使用单线程顺序消费：每次从 Kafka 拉取一条消息后提交，等待 NSO 返回结果再处理下一条。 单线程设计的瓶颈在于 NSO 提交设备的耗时差异很大，从几秒到几分钟不等。一条消息若因设备响应慢而阻塞，整个消费循环只能等待，整体吞吐严重受限。 改造目标是将 Committer 改为多线程并发消费，让不同设备的策略可以同时提交。 设备级并发约束 多线程并发带来一个新问题：物理网络设备在同一时刻只能处理一个配置提交。若并发向同一台设备提交多条策略，NSO 会返回冲突错误，设备状态也可能不一致。 因此，并发的粒度需要控制在设备维度：不同设备的提交可以同时进行，但同一台设备上的提交必须串行。 实现方案 多线程消费 Kafka 消费线程不再直接执行提交逻辑，而是将每条消息交给 ThreadPoolExecutor（默认 10 个 worker）异步处理。消费线程本身快速返回，继续 poll 下一批消息。 设备锁设计 引入一张设备锁表，记录当前正在进行中的提交和对应的设备，以及锁变更的时间戳。 提交任务在执行前先获取设备锁： SELECT ... FROM device_lock WHERE device_id = %s FOR UPDATE NOWAIT 加锁使用 NOWAIT：竞争失败时立即抛出异常而非阻塞等待，消费线程可以快速感知锁状态，不会因为等锁而阻塞整个提交流程。 提交完成后，在 finally 块中释放锁： SELECT ... FROM device_lock WHERE device_id = %s FOR UPDATE 解锁不使用 NOWAIT：此时当前任务本身就是锁的持有者，不存在竞争，阻塞等待语义更安全，确保成功取到行后再执行删除，不会因为瞬时异常而意外跳过清理步骤。无论 NSO 超时、解析异常还是其他未知错误，设备锁都会被清理，不会出现因为异常路径遗漏而导致的死锁。 锁的粒度是设备，不是请求。同一台设备的多条策略竞争同一把锁，不同设备的提交互不影响。 锁竞争与 Kafka 重投 竞争失败的消息不会被丢弃。Kafka 本身保证至少一次投递，消息会在下一个消费周期重新投递，再次参与锁竞争。NSO 侧做了幂等处理，重复提交不会导致设备状态错误。 ...

kafka kafka 大量使用页缓存, 经过良好调优的 kafka 集群, 磁盘读操作很少, 因为大部分消息读取操作会直接命中缓存 kafka 高吞吐量 大量使用操作系统页缓存，内存操作速度快且命中率高。 Kafka 不直接参与物理I/O操作，而是交由最擅长此事的操作系统来完成。 采用追加写入方式，摒弃了缓慢的磁盘随机读/写操作。 使用以 sendfile为代表的零拷贝技术加强网络间的数据传输效率。 摘自：《Apache Kafka实战》 — 胡夕 在豆瓣阅读书店查看：https://read.douban.com/ebook/59895902/ 本作品由电子工业出版社授权豆瓣阅读中国大陆范围内电子版制作与发行。 © 版权所有，侵权必究。 伸缩性，scalability 伸缩性表示向分布式系统中增加额外的计算资源（比如CPU、内存、存储或带宽）时吞吐量提升的能力。 partition Kafka的partition是不可修改的有序消息序列，也可以说是有序的消息日志。 replica 副本（replica） follower replica 是不能提供服务给客户端的，也就是说不负责响应客户端发来的消息写入和消息消费请求。它只是被动地向领导者副本（leader replica）获取数据 ISR的全称是in-sync replica，翻译过来就是与leader replica保持同步的replica集合 正常情况下，partition的所有replica（含leader replica）都应该与leader replica保持同步，即所有 replica都在 ISR中。因为各种各样的原因，一小部分 replica开始落后于 leader replica的进度。当滞后到一定程度时，Kafka会将这些 replica“踢”出 ISR。相反地，当这些 replica重新“追上”了 leader的进度时，那么 Kafka会将它们加回到 ISR中。这一切都是自动维护的，不需要用户进行人工干预，因而在保证了消息交付语义的同时还简化了用户的操作成本。 摘自：《Apache Kafka实战》 — 胡夕 在豆瓣阅读书店查看：https://read.douban.com/ebook/59895902/ 本作品由电子工业出版社授权豆瓣阅读中国大陆范围内电子版制作与发行。 © 版权所有，侵权必究。 Event Sourcing实际上是领域驱动设计（Domain-Driven Design，DDD）的名词，它使用事件序列来表示状态变更，这种思想和 Kafka 的设计特性不谋而合。 数据倾斜（skewed） 分布式消息队列是是大型分布式系统不可缺少的中间件,主要解决应用耦合、异步消息、流量削锋等问题。实现高性能、高可用、可伸缩和最终一致性架构。 为何使用消息系统 解耦 在项目启动之初来预测将来项目会碰到什么需求,是极其困难的。消息系统在处理过程中间插入了一个隐含的、基于数据的接口层,两边的处理过程都要实现这一接口。这允许你独立的扩展或修改两边的处理过程,只要确保它们遵守同样的接口约束。 冗余 有些情况下,处理数据的过程会失败。除非数据被持久化,否则将造成丢失。消息队列把数据进行持久化直到它们已经被完全处理,通过这一方式规避了数据丢失风险。许多消息队列所采用的"插入-获取-删除"范式中,在把一个消息从队列中删除之前,需要你的处理系统明确的指出该消息已经被处理完毕,从而确保你的数据被安全的保存直到你使用完毕。 ...