在Kafka中发布订阅模型

这是第四个柱中的一系列关于同步客户端集成与异步系统（ 1， 2， 3 ）。在这里，我们将尝试了解Kafka的工作方式，以便正确利用其发布-订阅实现。

卡夫卡概念

根据官方文件：

Kafka是一种分布式的，分区的，复制的提交日志服务。它提供消息传递系统的功能，但具有独特的设计。

Kafka作为集群运行，这些节点称为代理。代理可以是领导者或副本，以提供高可用性和容错能力。代理负责分区，分区是存储消息的分发单元。这些消息是有序的，可以通过名为offset的索引进行访问。一组分区构成一个主题，是消息的提要。一个分区可以有不同的使用者，它们使用自己的偏移量访问消息。生产者将消息发布到Kafka主题中。 Kafka文档中的以下图表可以帮助您理解以下内容：

排队与发布-订阅

消费者群体是另一个关键概念，有助于解释为什么Kafka比RabbitMQ等其他消息传递解决方案更灵活，功能更强大。消费者与消费者群体相关联。如果每个使用者都属于同一个使用者组，则主题的消息将在各个使用者之间平均负载均衡；这就是所谓的“排队模型”。相反，如果每个使用者都属于不同的使用者组，则所有消息都将在每个客户端中使用。这就是所谓的“发布-订阅”模型。

您可以混合使用这两种方法，分别针对不同的需求使用不同的逻辑使用者组，并在每个组中有多个使用者以通过并行提高吞吐量。同样， Kafka文档中的另一个图表：

了解我们的需求

正如我们在以前的文章（见1， 2， 3 ）该项目服务发布消息到卡夫卡的话题叫item_deleted 。此消息将位于该主题的一个分区中。为了定义消息将驻留在哪个分区中，Kafka提供了三种选择：

如果记录中指定了分区，请使用它
如果未指定分区但存在键，则根据键的哈希值选择一个分区
如果没有分区或密钥，则以循环方式选择一个分区

我们将使用item_id作为密钥。执法服务的不同实例中包含的消费者仅对特定分区感兴趣，因为他们保留某些商品的内部状态。让我们检查不同的Kafka使用者实现，以了解哪种使用最方便。

卡夫卡消费者

卡夫卡共有三个消费者：高级消费者，简单消费者和新消费者

在这三个消费者中， 简单消费者在最低级别上运行。它满足我们的要求，因为它允许消费者“在流程中仅使用主题中分区的子集”。但是，正如文档所述：

SimpleConsumer确实需要使用者组中不需要的大量工作：

您必须跟踪应用程序中的偏移量，才能知道从何处停止消费
您必须确定哪个Broker是主题和分区的主要Broker。
您必须处理经纪人负责人变更

如果您阅读了建议的用于处理这些问题的代码，则将不鼓励您使用此使用者。

新使用者提供正确的抽象级别，并允许我们订阅特定的分区。他们在文档中建议以下用例：

第一种情况是，如果进程正在维护与该分区关联的某种本地状态（例如本地磁盘上的键值存储），因此该进程应仅获取其在磁盘上维护的分区的记录。

不幸的是，我们的系统使用的是Kafka 0.8，而该使用者仅从0.9开始可用。我们没有足够的资源来迁移到该版本，因此我们需要坚持使用高级消费者 。

该使用者提供了一个不错的API，但不允许我们订阅特定的分区。这意味着，执法服务的每个实例都将使用每条消息，即使那些无关的消息也是如此。我们可以通过为每个实例定义不同的消费者组来实现。

利用Akka Event Bus

在上一篇文章中，我们定义了一些等待ItemDeleted消息的有限状态机ItemDeleted 。

when(Active) {case Event(ItemDeleted(item), currentItemsToBeDeleted@ItemsToBeDeleted(items)) =>val newItemsToBeDeleted = items.filterNot(_ == item)newItemsToBeDeleted.size match {case 0 => finishWorkWith(CensorResult(Right()))case _ => stay using currentItemsToBeDeleted.copy(items = newItemsToBeDeleted)}}

我们的卡夫卡消费者可以将所有消息转发给那些演员，并让他们丢弃/过滤不相关的物品。但是，我们不想让参与者浪费大量的冗余和低效的工作，因此我们将添加一层抽象，使他们能够以一种非常有效的方式丢弃适当的消息。

final case class MsgEnvelope(partitionKey: String, payload: ItemDeleted)class ItemDeletedBus extends EventBus with LookupClassification {override type Event = MsgEnvelopeoverride type Classifier = Stringoverride type Subscriber = ActorRefoverride protected def mapSize(): Int = 128override protected def publish(event: Event, subscriber: Subscriber): Unit = subscriber ! event.payloadoverride protected def classify(event: Event): Classifier = event.partitionKeyoverride protected def compareSubscribers(a: Subscriber, b: Subscriber): Int = a.compareTo(b)
}

Akka Event Bus按分区为我们提供订阅，而我们的Kafka高级消费者中缺少该分区。我们将从卡夫卡消费者处发布每条消息到公交车上：

itemDeletedBus.publish(MsgEnvelope(item.partitionKey, ItemDeleted(item)))

在上一篇文章中，我们展示了如何使用该分区键订阅消息：

itemDeletedBus.subscribe(self, item.partitionKey)

LookupClassification将过滤不需要的消息，因此我们的参与者不会过载。

摘要

由于Kafka提供的灵活性，我们能够设计我们的系统以了解不同的折衷方案。在接下来的文章中，我们将看到如何协调这些FSM的结果以向客户端提供同步响应。

第一部分 | 第2部分 | 第三部分

翻译自: https://www.javacodegeeks.com/2016/05/publish-subscribe-model-kafka.html