​

数新网络官网已全新上线，欢迎点击访问

www.datacyber.com 数新网络_让每个人享受数据的价值

01 Kafka 介绍

Kafka 是最初由 Linkedin 公司开发，是一个分布式、支持分区的（partition）、多副本的（replica），基于 zookeeper 协调的分布式消息系统，它的最大的特性就是可以实时的处理大量数据以满足各种需求场景：比如基于 hadoop 的批处理系统、低延迟的实时系统、Storm/Spark 流式处理引擎，web/nginx 日志、访问日志，消息服务等等，用 scala 语言编写，Linkedin 于 2010 年贡献给了 Apache 基金会并成为顶级开源项目。

02 Kafka 的使用场景

日志收集：一个公司可以用 Kafka 收集各种服务的 log，通过 kafka 以统一接口服务的方式开放给各种 consumer，例如 hadoop、Hbase、Solr 等。

消息系统：解耦和生产者和消费者、缓存消息等。

用户活动跟踪：Kafka 经常被用来记录 web 用户或者 app 用户的各种活动，如浏览网页、搜索、点击等活动，这些活动信息被各个服务器发布到 kafka 的 topic 中，然后订阅者通过订阅这些 topic 来做实时的监控分析，或者装载到 hadoop、数据仓库中做离线分析和挖掘。

运营指标：Kafka 也经常用来记录运营监控数据。包括收集各种分布式应用的数据，生产各种操作的集中反馈，比如报警和报告。

03 Kafka 优化

Kafka 作为一款高性能消息中间件被广泛应用于各大系统中，但是同其他中间件一样，也会存在一些问题。

消息丢失情况

消息在发送端和消费端都有可能会产生数据丢失的情况。

消息发送端：

• acks=0 时，表示 producer 不需要等待任何 broker 确认收到消息的回复，就可以继续发送下一条消息。如果此时 broker 宕机，就会导致消息丢失。

• acks=1 时，至少要等待 leader 已经成功将数据写入本地 log，但是不需要等待所有 follower 是否成功写入。作为集群使用时，如果 follower 还未成功写入，在 leader 写入后，follow 还没有来得及 fetch 到 leader 的最新消息，leader 宕机了，follower 拉取失败，并开始进行 leader 选举，新的 leader 因为没有同步最新的消息，导致该消息丢失。

• acks=-1 或 all 时，这意味着 leader 需要等待所有备份都成功写入日志。这种策略会保证只要有一个备份存活就不会丢失数据。但是需要 min.insync.replicas 配置的备份个数大于等于 2，当 leader 宕机之后，会重新进行 leader 选举，选举 lsr 列表中的第一个 broker 作为 leader，而 lsr 列表中的 broker 都是同步数据最多的，会保证数据不丢失。

消息消费端：

如果消息是自动提交，万一消费到数据还没处理完，就自动提交 offset 了，但是此时你 consumer 直接宕机了，未处理完的数据丢失了，下次也消费不到了。

消息重复消费

消息重复消费与消息丢失一样，在发送端和消费端都会产生数据重复消费的情况。

消息发送端：

发送消息如果配置了重试机制，服务端收到了消息，并进行 ack 的时候，由于网络问题导致发送端一直没有收到服务端发送的返回消息，就会启动重试机制再次发送。

消息消费端：

如果消费端配置了自动提交，刚拉取了一批数据处理了一部分，还没来得及提交，服务挂了，下次重启又会拉取相同的一批数据重复处理。

一般消费端都要进行消费幂等处理。

消息乱序

如果发送端配置了重试机制，Kafka 不会等之前那条消息完全发送才去发送下一条消息，这样可能会出现，发送时顺序为 1，2，3 的消息，第一条超时后重新发送，后面两条发送成功，最终消费端消费的顺序是 2，3，1。

Kafka 要保证全链路消息顺序消费，需要从发送端开始，将所有消息有序发送到同一个分区，然后用一个消费者去消费，但是这种性能比较低，可以在消费者端接收到消息后将需要保证顺序消费的几条消费发到内存队列，一个内存队列开启一个线程顺序处理消息。

如果需要将消息发送到不同分区并保证顺序消费。一般不建议这么做。发送端在发送消息时，在消息中添加一个排序号，消费者端在接收时定义一个 CountDownLatch，确保将需要顺序消费的消息收齐，根据排序号排序后再处理。

消息积压

线上有时因为发送端发送消息速度过快，或者消费端处理消息过慢，导致 broker 积压大量未消费消息。这种情况可以修改消费端程序，让其将收到的消息快速转发到其他 topic，然后再启动多个消费者同时消费新主题的不同分区。

由于消息数据格式变动或消费端程序有 bug，导致消费者一致消费不成功，也会导致 broker 积压大量未消费消息。这种情况可以配置一个 topic 作为死信队列，将消费不成功的的消息放入到死信队列，之后再慢慢分析死信队列里的消息处理问题。

延时队列

延时队列存储的对象是延时消息。指消息被发送以后，并不想让消费者立刻获取，而是等待特定的时间后，消费者才能获取这个消息进行消费，延时队列的使用场景有很多，比如：

• 在订单系统中，一个用户下单之后通常有 30 分钟的时间进行支付，如果 30 分钟之内没有支付成功，那么这个订单将进行异常处理，这时就可以使用延时队列来处理这些订单了。

• 订单完成 1 小时后通知用户进行评价。

  实现思路：发送延时消息时先把消息按照不同的延迟时间段发送到指定的队列中（topic_5s,topic30s…,topic_n，这个一般不能支持任意时间段的延时），然后通过定时器进行轮询消费这些 topic，查看消息是否到期，如果到期就把这个消息发送到具体业务处理的 topic 中，队列中消息越靠前的到期时间越早，具体来说就是定时器在一次消费过程中，对消息的发送时间做判断，看下是否延迟到对应时间了，如果到了就转发，如果还没到这一次定时任务就可以提前结束了。

消息回溯

如果某段时间对已消费消息计算的结果觉得有问题，可能是由于程序 bug 导致的计算错误，当程序 bug 修复后，需要对之前已消费的消息重新消费，可以指定从多久之前的消息回溯消息，这种可以用 consumer 的 offsetForTimes、seek 等方法制定从某个 offset 偏移的消息开始消费。

消息传递保障

• at most once（消费者最多收到一次消息）：acks=0 可以实现

• at least once（消费者至少收到一次消息）：acks=all 可以实现

• exactly once（消费者刚好收到一次消息）：at least once 加上消费者幂等性可以实现，还可以用 Kafka 生产者的幂等性来实现。

04 Kafka 事务

Kafka 的事务不同于 Rocketmq，Rocketmq 是保障本地事务与 mq 消息发送的事务一致性，Kafka 的事务主要是保障一次发送多条消息的事务一致性，一般在 Kafka 的流式计算场景用得多一点，比如，Kafka 需要对一个 topic 里的消息做不同的流式计算处理，处理完分别发到不同的 topic 里，这些 topic 分别被不同的下游系统消费，这种我们肯定希望系统发送到多个 topic 的数据保持事务一致性。Kafka 要实现类似 Rocketmq 类似的分布式事务需要额外开发功能。

本期分享就到这里，欢迎关注我们了解更多精彩内容~

​