首页技术文章spring kafka 写生产者消费者及参数详解

spring kafka 写生产者消费者及参数详解

发布时间：2025-02-02 00:12:15
·
44 阅读
·
0 评论
文章标签： bootstrap table参数

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

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

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

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


  org.springframework.boot
  spring-boot-starter-web
  2.1.0.RELEASE


  org.springframework.kafka
  spring-kafka
  2.2.0.RELEASE


  org.apache.kafka
  kafka-clients
  2.2.0

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种类型的值（字符串）。

  • acks=1。默认值即为1。生产者发送消息之后，只要分区的leader副本成功写入消息，那么它就会收到来自服务端的成功响应。如果消息无法写入leader副本，比如在leader 副本崩溃、重新选举新的 leader 副本的过程中，那么生产者就会收到一个错误的响应，为了避免消息丢失，生产者可以选择重发消息。如果消息写入leader副本并返回成功响应给生产者，且在被其他follower副本拉取之前leader副本崩溃，那么此时消息还是会丢失，因为新选举的leader副本中并没有这条对应的消息。acks设置为1，是消息可靠性和吞吐量之间的折中方案。
  • acks=0。生产者发送消息之后不需要等待任何服务端的响应。如果在消息从发送到写入Kafka的过程中出现某些异常，导致Kafka并没有收到这条消息，那么生产者也无从得知，消息也就丢失了。在其他配置环境相同的情况下，acks 设置为 0可以达到最大的吞吐量。
  • acks=-1或acks=all。生产者在消息发送之后，需要等待ISR中的所有副本都成功写入消息之后才能够收到来自服务端的成功响应。在其他配置环境相同的情况下，acks 设置为-1（all）可以达到最强的可靠性。但这并不意味着消息就一定可靠，因为ISR中可能只有leader副本，这样就退化成了acks=1的情况。

    • 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 配置的毫秒数。自动提交替我们做了部分工作，但是可能存在重复消费和消息丢失的情况。

  • 重复消费。当业务逻辑已经处理完，但是 offset 未来得及提交到 kafka，服务就挂了，重启之后就会出现重复消费。
  • 消息丢失。假如在业务处理流程中使用了线程池，当 offset 提交之后，线程执行报错，就不会再次消费此条消息。

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

  • RECORD。每消费一条就提交一次。
  • BATCH。每次poll的时候批量提交一次，频率取决于每次poll的调用频率。
  • TIME。每次间隔设置的 ackTime 的时间提交。
  • COUNT。累计多少个 ACK 提交到 Kafka。
  • COUNT_TIME。count 或 time 达到一个就提交。
  • MANUAL。手动 ACK，异步提交。
  • MANUAL_IMMEDIATE。手动 ACK，立即提交。

    • 当使用 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。

  • earliest

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

  • latest

    • 当各分区下有已提交的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_）