kafka生产者向集群发送消息超时后的重试机制如何?怎样保证只有所有消费者都不能访问时才丢弃消息?

单调 发表于: 2018-05-29   最后更新时间: 2018-05-29 14:37:42   8,239 游览

说明示例如下:

C作为生产者需要给集群生产某个topic的消息,集群内有S1\S2\S3三个服务。

正常情况下,C只要给S1\S2\S3三个的任何一个发送成功了就完成了一次生产。

现在S1的防火墙因为一些原因,把C给拒了。现象就是C给S1发送是会超时,其他两个则正常。造成消费报告现在是偶尔会丢失一个C的产品。


上面是我现在遇到的问题,不知道为什么会造成这种现象。

希望达到的目的是,只有S1\S2\S3同时不能不能访问了(挂掉了或者防火墙拦截了等所有原因),才考虑丢弃。其他情况都逐个尝试。

发表于 2018-05-29
添加评论

1、消费者和生产者之间毫无关系,互不影响,所谓解耦正是如此,我直管发,不管谁来消费。
2、消息传递保证,可以看这篇文章。https://www.orchome.com/21

kafka生产者发送是根据一个主题的分区进行发送的。假设s1、s2、s3三个节点的kafka集群,创建一个3个分区的主题,则这3个分区会平均分配到3个节点上,当生产者发消息的时候,会轮询这3个分区,将消息写进去,正好有个分区在s1上,由于你防火墙的原因,就失败了,消息丢失。

单调 -> 半兽人 5年前

“acks=1,这意味着leader写入消息到本地日志就立即响应,而不等待所有follower应答。在这种情况下,如果响应消息之后但follower还未复制之前leader立即故障,那么消息将会丢失。   ”, 我使用的默认配置就是这个。按照我这个示例得到的现象,S1的防火墙直接把C给拦截了,C是知道timeout了的。但是C却没有再去重试S2和S3,这是为什么呢?

半兽人 -> 单调 5年前

设置一下 retries 。默认是0呢。

单调 -> 半兽人 5年前

重试次数默认是3的。就是不知道这个重试,是对同一个broker重试,还是会对其他broker重试?

感谢 @半兽人 的帮助。通过这个问题我基本了解了kafka生产者发送消息的流程。这里总结下我的理解,有误请指正:

我这里的示例,实际就是某个broker不能访问了,但实际服务是还存在的。当某个消息发送失败时,协议层(kafka的发送逻辑)的重试逻辑是为这个消息对上一个分配的分区重试,而不是重新选择分区重试,因为协议层也不确定是不是用户又这个需求。

那我这里是希望重新选择分区进行发送的,所以应该在我的业务层,接收到消息发送失败后,重新对消息入栈发送。至于怎么保证不会再去尝试同一个分区,这个跟具体的分区选择算法相关。config.Producer.Partitioner 决定了用什么方式再去选择分区。最简单的是直接使用roundbin算法。

当业务层也进行了重试,重试若干次仍然失败后,那基本上能确定是所有的broker都访问不了了,由业务层进一步去确认该怎么处理。

单调 -> 单调 5年前

这里还会出现问题,就是roundbin分区算法,是全局的。当消息并发发送的时候,上一次发送失败的消息还是有可能被分配到那个连不上的broker上。我采取的解决方式是,消息发送失败的时候,将失败的分区号记在消息的meta结构中,然后自己实现一个分区选择器,这个分区选择器过滤掉消息的meta结构中记录的分区号,然后随机分配一个分区。

你的答案

查看kafka相关的其他问题或提一个您自己的问题