其他分享
首页 > 其他分享> > kafka 如何保证消息不丢失

kafka 如何保证消息不丢失

作者:互联网

今天我们来分析一下这个问题。

先来回忆一下kafka 中消息传输的整个过程

 

1、kafka 在producer 端产生消息,调用kafka producer client send方法发送消息

2、kafka producer client 使用一个单独的线程,异步的将消息发送给kafka server

3、kafka server收到消息以后,保存数据,并同步至副本

4、消息保存完成以后,返回给kafka producer client 端 【消息发送成功】

5、kafka consumer client 调用poll 方法,循环 从kafka server 端获取消息列表

6、kafka consumer 端 从kafka server获取到消息以后,开始消费消息

7、kafka consumer 消费消息完毕以后,向kafka server(topic为 _offset_consumer的消息队列) 发送偏移量

在上述的整个流程中,消息丢失的情况分为以几种可能性:

1、producer 端 发送消息给kafka server 端,中间网络出现问题,消息无法送达

2、kafka server端 在收到消息以后,保存消息时发生异常,异常分为三种

(1)可重试错误,通过重试来解决

(2) 网络连接错误

(3)无主(no leader)错误

3、consumer 在消费消息时发生异常,导致consumer端消费失败

注:当然这里还可能发生另一种错误,就是在producer发送消息到kafka server端时,消息体过大,producer client 直接抛出异常,导致发送失败

如何解决
1、producer 端的发送方式优化
我们先来了解一下,producer端发送消息的方式:

1.1    简单发送,无需关心结果

  1.   ProducerRecord<String,String> record = new ProducerRecord<>(
  2.        "topicName","key","value"  
  3.   );
  4.   try{
  5.       //这里只是把消息放进了一个缓冲区中,然后使用单独的线程将消息发送到服务端
  6.       producer.send(record);
  7.   }
  8.   catch(Exception){
  9.       e.printStackTrace();
  10.   }

1.2    同步发送

  1.   ProducerRecord<String,String> record = new ProducerRecord<>(
  2.        "topicName","key","value"  
  3.   );
  4.   try{
  5.       //send方法返回的是Future<RecordMetaData> 对象,然后我们可以调用get()方法等待响应
  6.       Future<RecordMetaData> future = producer.send(record); 
  7.       future.get();
  8.   }
  9.   catch(Exception){
  10.       e.printStackTrace();
  11.   }

1.3    异步发送

  1.   private class DemoProducerCallback implements Callback{
  2.       @override
  3.       public void onCompletion(RecordMetadata recordMetadata,Exception e){
  4.           //发生错误的回调方法,可以写入日志,或写入DB通过其它线程重重试,保证最终的数据送达
  5.       }
  6.   }
  7.   ProducerRecord<String,String> record = new ProducerRecord<>(
  8.        "topicName","key","value"  
  9.   );
  10.   producer.send(record,new DemoProducerCallback()))


总结:从以上的三种发送方式中,我们可以知道,采用第一种方式发送时,消息丢失时我们的应用程序是无感知的,如果需要保证消息的不丢失,那么必须要选择第二种或者第三种(需要配合下一节中讲到的acks 参数),当然这里更推荐第三方种方式。

2、producer端的配置优化
在producer 端的配置项中有很多的配置项,我们摘出几种比较重要的来一一解读:

acks:该参数指定了,kafka server的多少个副本收到消息以后才算真的正消息发送成功。取值范围:

acks = 0 表示producer 在将消息成功写入到 kafka server 之前不会收任消息
acks = 1 表示只要kafka server 集群中的leader节点收到消息,producer 端就会收到kafka server的成功响应
acks = all 表示只有当消息到leader节点,并且这条数据也同步到了所有副本中,producer 才会收到kafka server的成功响应。
buffer.memory:生产端 缓冲区的大小设置

compression type:生产端采用的数据压缩方式,取值 snappy,gzip,lz4,默认不会压缩。(启用压缩意味着,需要producer 和kafka server要占用更多的cpu资源)

retries:生产端发送消息到kafka server时,发生临时性错误以后,生产者发送消息到kafka server端重试的次数。如果重试超过该次数,则发生异常

batch.size: 当多个消息被发送至同一分区时,生产者会把它们发送到同一批。该参数指定了同一批次可以使用的内存大小,按字节数计算(而不是消息条数)。

linger.ms:该参数指定了生产者在发送批次之前等待更多消息加入批次的时间,producer client 会在批次填满(batch.size) 或linger.ms 到上限时,将消息发送至kafka server.

max.in.flight.requests.per.connection:该参数指定了生产者在收到kafka server 的成功响应之前,可以发送多少消息。(可以利用该配置让kafka server中的消息变得有序)

max.request.size:该参数用来控制生产者发送单个请求的数据大小。对于消费端也有相同的配置(message.max.bytes),建议两边设置相同。

总结:我们的问题,可以通过设置配置项 acks 、retries 来保证数据的不丢失。acks=1时,lead节点只要收到消息就会告诉producer消息接收成功,假如此时lead 挂掉了开始重新选主,选主成功后之前lead收到的那条消息就会丢失,如果需要保证消息的绝对不丢失,建议设置 acks =all

3、kafka server
这里需要补充一个知识点,kafka的server端同一个topic下有多个分区,单个分区会有不同的副本。如果producer 发送消息么kafka server端,leader收到了消息以后,告诉producer 发送成功,此时再同步消息到多个副本,但由于某一个副本同步较慢,此时leader挂了,需要选主,选主的过程中,一旦那个较慢的副本成为新的leader,那么新的leader中就不包含了原leader收到的那条最新数据,导致消息丢失。

broker中的配置项,unclean.leader.election.enable = false,表示不允许非ISR中的副本被选举为首领,以免数据丢失。

ISR:是指与leader保持一定程度(这种范围是可通过参数进行配置的)同步的副本和 leader 共同被称为ISR

OSR:与leader同步时,滞后很多的副本(不包括leader)被称为OSR

AR,分区中所有的副本统称为AR。AR = ISR + OSR

4、kafka consumer端的优化
kafka consumer的配置中,默认的enable.auto.commit = true,表示在kafka consumer 通过poll方法 获取到消息以后,每过5秒(通过配置项可修改)会自动获取poll中得到的最大的offset, 提交给kafka server 中的_offset_consumer(存储 offset 的特定topic )

如果enable.auto.commit = false时,则关闭了自动提交,你可以手动的通过应用程序代码进行提交,这里我来梳理一下,consumer 消费消息的整个流程

 consumer端循环向kafka server请求获取信息
如果kafka server中的分区中没有消息,则阻塞指定秒数(consumer端配置)后,返回给consumer端
 如果 kafka server中有消息 或是在阻塞等待的过程中有消息写入,则立即返回给consumer端
consumer开始消费消息
consumer消费消息完毕以后,提交偏移量到topic为 _offset_consumer(kafka server端) 的消息队列
我们来看一下,enale.auto.commit = false时,如何手动提交的

  1.   public void consumerMsg(){
  2.       while(true){
  3.           //这里的poll(100)指的是kafka server端没有消息时,连接等待的时间,超过该时间立即返回空给consumer
  4.           ConsumerRecords<String,String> records = consomer.poll(100);
  5.           for(ConsumerRecord<String,String> record : records){
  6.               // 这里是消费消息的逻辑(简单逻辑输入到控制台)
  7.               System.out.printIn(record.value));
  8.               //提交偏移量
  9.               try{
  10.                   consumer.commitSync();  //同步提交 如果异步的话,可以使用 consumer.commitAsync();
  11.               }
  12.               catch(CommitFailedException ex){
  13.                   log.error("commit fail");
  14.               }
  15.           }
  16.       }
  17.   }

consumer端消息丢失的情况分为两种:

consumer 端启用了 enable.auto.commit= true,在消费消息时发生了异常
consumer 端 enable.auto.commit= false,但是在消息消费之前,提交了offset
针对这两种丢失的情况,我们做以下处理:

1、设置 enable.auto.commit = false

2、在consumer端消费消息操作完成以后 再提交 offset,类似于上文中的代码示例

写在最后:
以上的我们从producer 、 kafka server、 consumer 端出发,通过相关的优化手段保证消息的不丢失,当然业界还有一些其它的办法,比如在三种 send 的 调用方式中,callback时将消息写入到mysql 或日志中,当consumer 消费消息成功以后,我们从mysql 或 日志中删除消息,未成功消费的消息,可以启动一个线程,将消息重新入队让consumer收到消息以后重新消费(rabbit mq中 可以 利用死信队列和备用交换机来完成)

标签:producer,server,发送,保证,消息,kafka,consumer,丢失
来源: https://www.cnblogs.com/cfas/p/16593290.html