邯郸住房城乡建设厅网站,响应式网站推广,受欢迎的郑州网站建设,在线一键建站系统这是第四个柱中的一系列关于同步客户端集成与异步系统#xff08; 1#xff0c; 2#xff0c; 3 #xff09;。 在这里#xff0c;我们将尝试了解Kafka的工作方式#xff0c;以便正确利用其发布-订阅实现。 卡夫卡概念 根据官方文件 #xff1a; 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), currentItemsToBeDeletedItemsToBeDeleted(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
