Kafka 是目前非常主流的一款 MQ 产品，很多开发人员都使用过，实际场景中经常被用来做系统解耦、消息补偿、日志收集等。如果从零开始写一个生产者、消费者该怎么写呢？安哥写了个 Demo，先不管原理机制，跑通再说。

以下内容主要分为两大块：

1. producer 样例及主要参数详解
2. consumer 样例及主要参数详解

spring 提供了 kafka producer 和 consumer 工具，我们可以很方便的实现生产者和消费者。以下是 maven 依赖，注意 spring-kafka 版本一定要和 kafka-clients 版本兼容，不然可能会出现消费不到消息的情况。

<dependency>
  <groupId>org.springframework.boot</groupId>
  <artifactId>spring-boot-starter-web</artifactId>
  <version>2.1.0.RELEASE</version>
</dependency>
<dependency>
  <groupId>org.springframework.kafka</groupId>
  <artifactId>spring-kafka</artifactId>
  <version>2.2.0.RELEASE</version>
</dependency>
<dependency>
  <groupId>org.apache.kafka</groupId>
  <artifactId>kafka-clients</artifactId>
  <version>2.2.0</version>
</dependency>

01. Producer

我们可以通过 spring kafkaTemplate 向 Kafka server 发送消息，发送参数除了 data 之外还有 topic、partition、key 等，非特殊需求我们只需要关注 topic。

partition ：指定往哪个分区发送。
同一个partition的消息会顺序消费

key：key 的 hash 值会作为 partition，
并且此值会传递到 Consumer
一般顺序消费场景需要设置 key，例如订单号为 key

如果partition、key 都没有设置，则采用轮询的方式向 partition 发送消息。

由于 producer 是异步发送消息的，我们不能立刻知道发送结果，spring kafka 提供了 SuccessCallback、FailureCallback 两个接口分别处理成功回调和失败回调。

创建kafkatemplate 的时候需要指定一些参数，下面列举出我们比较关心参数的并详细介绍。

1. acks

这个参数用来指定分区中必须要有多少个副本收到这条消息，之后生产者才会认为这条消息是成功写入的，acks参数有3种类型的值（字符串）。

2. max.request.size

用来限制生产者客户端能发送的消息的最大值，默认值为 1048576B，即1MB。

3. retries 和retry.backoff.ms

retries参数用来配置生产者重试的次数，默认值为0，即在发生异常的时候不进行任何重试动作。消息在从生产者发出到成功写入服务器之前可能发生一些临时性的异常，比如网络抖动、leader副本的选举等，这种异常往往是可以自行恢复的，生产者可以通过配置retries大于0的值，以此通过内部重试来恢复而不是一味地将异常抛给生产者的应用程序。如果重试达到设定的次数，那么生产者就会放弃重试并返回异常。不过并不是所有的异常都是可以通过重试来解决的，比如消息太大，超过max.request.size参数配置的值时，这种方式就不可行了。

retry.backoff.ms 用来设定两次重试之间的时间间隔，单位毫秒，默认值为100。

如果发送失败会回调给 FailureCallback，可以进行业务处理。

4. compression.type

压缩方式，默认值为“none”，即默认情况下，消息不被压缩。该参数还可以配置为“gzip”“snappy”和“lz4”。

5. connections.max.idle.ms

指定在多久之后关闭限制的连接，默认值是540000 ms

6. linger.ms

单位：毫秒。用来指定生产者发送 ProducerBatch 之前等待更多消息（ProducerRecord）加入ProducerBatch 的时间，默认值为 0。生产者客户端会在 ProducerBatch 被填满或等待时间超过linger.ms 值时发送出去。增大这个参数的值会增加消息的延迟，但是能提升一定的吞吐量。

7. request.timeout.ms

单位：毫秒。用来配置Producer等待请求响应的最长时间，默认值为30000。请求超时之后可以选择进行重试。

8. buffer.memory

单位：byte。producer 发送消息，不是一条条的同步发送，而是经过缓冲区，producer 发送的消息先写JVM本地缓存，然后线程异步发送到Broker上去的。缓冲区的大小就是 buffer.memory 来控制的，默认值32MB。

如果buffer.memory设置的太小，可能导致：消息快速的写入内存缓冲里，但Sender线程来不及发送到Kafka服务器，会造成内存缓冲很快就被写满，这样就会阻塞用户线程，不让继续往Kafka写消息了。

所以即使业务代码发送了消息，也不一定会成功，需要依赖 SuccessCallback 回调，此回调也是异步的。

9. batch.size

单位：byte。每个Batch要存放batch.size大小的数据后，才可以发送出去。比如说batch.size默认值是16KB，那么里面凑够16KB的数据才会发送。

10. key.serializer 和 value.serializer

序列化方式，要与消费者的反序列化方式对应。key.serializer 作用于上面提到的发送参数 key，value.serializer 作用于发送消息体即 data 或叫 value。

11. bootstrap.servers

kafka broker host:port 列表，多个以逗号隔开

02. Consumer

sping-kafka 提供了 @KafkaListener 很优雅的代替了我们自己起线程循环 poll 消息的工作。

@KafkaListener 消费消息的核心逻辑在
KafkaMessageListenerContainer，它也是起了一个线程，循环调用 Consumer.poll(Duration timeout);接口拉取消息，timeout参数的解释是："The max time to block in the consumer waiting for records."，Consumer 一次 poll 消息最长的阻塞时间，默认 5000ms。

Consumer 负责订阅 Kafka 中的 Topic，并且从订阅的Topic上拉取消息。在Kafka的消费理念中还有一层消费组（Consumer Group）的概念，每个消费者都有一个对应的消费组。当消息投递到Kafka后，只会被投递给订阅它的每个消费组中的一个消费者。

比如支付中心 PaymentCenter 生成的消息 topic 是 paySuccess，同时有两个服务订阅了（订单中心和用户中心），那么这两个消费必须使用不同的 group。

1. group.id

消费组id，名字最好与服务名相关，例如 paymentCenterGroup、userCenterGroup，group.id 既可以指定在 @KafkaListener上也可以设置在ConsumerFactory。

2. enable.auto.commit

是否自动提交 offset，kafka 消费者提交offset 想要做好是一个比较麻烦的事情。

TURE - 自动提交。此时 Consumer 将定时提交 offset，提交周期是 auto.commit.interval.ms 配置的毫秒数。自动提交替我们做了部分工作，但是可能存在重复消费和消息丢失的情况。

FALSE - 不自动提交。这种情况需要开发人员配置提交策略或者手动提交。spring-kafka AckMode 提供了以下几种提交策略，可以在ConcurrentKafkaListenerContainerFactory.ContainerProperties.AckMode 里设置。

当使用 MANUAL 或 MANUAL_IMMEDIATE 时需要在 Consumer 手动调用
Acknowledgment.acknowledge 来提交 ACK。

手动提交也存在提交 offset 失败，重复消费的可能，这就需要业务上做到幂等。幂等常见有两种方式

1. 存储已经消费过的消息id、或业务id。例如 redis 保存已经消费的订单id，
重复消费时，先判断是否已经消费过。
2. DB 插入幂等或更新幂等。插入幂等依赖唯一键，
更新幂等使用乐观锁：update table set status = to where id = 1 and status = from。
这种需要有事务控制。

3. auto.offset.reset

支持配置：earliest、latest。

当各分区下有已提交的offset时，从提交的offset开始消费；无提交的offset时，从头开始消费

当各分区下有已提交的offset时，从提交的offset开始消费；无提交的offset时，消费新产生的该分区下的数据

4. pollTimeout

在
ConcurrentKafkaListenerContainerFactory.containerProperties.pollTimeout设置。作用于Consumer.poll(timeout)

5. value.deserializer 和 key.deserializer

与生产者的key.serializer、value.serializer对应

6. session.timeout.ms

用于消费者的故障检测，消费者会定期向 kafka server 发送心跳，这个值必须设置在broker configuration中的
group.min.session.timeout.ms 与
group.max.session.timeout.ms之间。但是 后面这两个参数没有看到配置的地方。

7.heartbeat.interval.ms

消费者心跳间隔时间，必须小于session.timeout.ms，通常不应该大于其 1/3。

公众号：看起来很美（kanqilaihenmei_）