我和Kafka的第一英里有点问题。每个人都在处理分区等问题，但如何处理第一英里？
我的系统由许多产生分布在节点上的事件的应用程序组成。我需要以可靠/故障安全的方式将这些事件传递给一组充当使用者的应用程序。选择的消息传递系统是kafka（由于其日志特性），但它不是一成不变的。
事件应尽可能以解耦的fire-n-forget方式传播。这意味着生产者应该完全负责可靠地传递他们的消息。这意味着产生事件的应用程序根本不应该担心事件的传递。
生产商的可靠性方案必须考虑：
箱式连接中断-在中断期间，生产商根本无法访问网络；因此，Kafka集群是不可到达的
框式重启-生产者和事件产生应用程序重启（独立）；生产者应保留正在传输的消息（在重试、批处理等过程中）
内部Kafka例外-消息大小太大；序列化异常；等。
到目前为止，我查过的图书馆里没有一本涵盖这些案例。有没有解决这个问题的建议策略？
我知道在制作过程中有可重试和不可重试的错误 send() . 对于那些可检索的文件，库通常在内部处理所有内容。但是，不可重试在异步回调中以异常结束。。。
我是不是应该盲目地把这些重放到无穷远？对于网络中断，它应该工作，但如何Kafka内部错误-说消息太大。可能有一个死信队列，如机制+重播。但是，如何处理消息计数。。。
关于持久性-一个轻量级数据库后端应该可以解决这个问题。只是创建一个持久队列，然后删除那些已经发送/确认的队列。然而，我担心，如果它是这么简单，它早就在标准Kafka库中实现了。业绩可能会下滑。
看到Kafka-3686或Kafka-1955这样的东西，我有点担心。
提前谢谢。
1条答案
按热度按时间aiazj4mn1#
我们有一个生产系统，其主要用例是可靠的消息传递。我不能说太多细节，但是我可以分享一个如何实现这一点的高层次设计。然而，这个系统是保证“至少一次交付”的消息传递语义。
来源
首先我们设计了一个消息模式，所有发送到这个系统的消息都必须遵循它。
然后我们将消息写入mysql消息表，该表按日期分片，并带有标记为delivered或not的字段
我们有一个应用程序不断轮询数据库，其中的行标记为undelivered，提取一行，构造消息并将其发送到负载均衡器，这是一个阻塞调用并更新消息行以进行传递，只有在返回200时，如果是5xx，应用程序将在关闭sleep back的情况下重试消息。您还可以根据需要配置重试。
每个源系统都维护自己的轮询应用程序和数据库。
生产者阵列
这基本上是负载平衡器下的一组机器，它们等待传入的消息并将这些消息生成到kafka集群。
我们维护每个主题的3个副本，并且在producer配置中我们保持acks=-1，这对于您的fire-n-forget需求非常重要。根据文件
acks=所有这一切意味着领导者将等待全套同步副本确认记录。这保证了只要至少有一个同步副本保持活动状态，记录就不会丢失。这是最有力的保证。这相当于acks=-1设置
正如我所说的，生成是一个阻塞调用，如果在所有3个副本上成功生成消息，它将返回2xx。4xx，如果消息is不满足模式要求5xx，如果kafka代理抛出了一些异常。
用户阵列
这是一个普通的机器阵列，运行Kafka的高级消费者为主题的消费群体。
我们目前在运行这个设置时，很少为生产中的其他一些功能流添加其他组件，从源代码的Angular 来看，它基本上是fire-n-forget。
这个系统解决了你所有的问题。
盒连接中断：除非源轮询应用程序得到2xx，否则它将再次生成，这可能导致重复。
box restart：由于源代码的重试机制，这应该不是问题。
内部kafka异常：由polling app处理，因为producer array将以5xx无法生成的方式回复，并将进一步重试。
acks=-1，还确保所有副本都是同步的，并且有一个消息的副本，所以代理宕机也不会是一个问题。