kafka

kafka 属于 Stream 的消费模型，为了支持多 partition 的消费关系，引入了 consumer group 的概念，同时支持在消费端动态的 reblance 操作，当多个 Consumer 订阅了同一个 Topic 时，会根据分区策略进行消费者订阅分区的重分配。只要 consumer-group 与 topic 之间的关系发生变更，就会动态触发 reblance 操作，诸如：

• 增加或减少 topic 中 partition 的数目

• consumer-group 中的 consumer 数减少

• consumer-group 与 topic 之间的订阅关系发生变更

• 等等

引入 reblance 的好处在于，当订阅关系发生变更时，用户无需重新启动系统，就可以实现订阅关系的变更，相当于 kafka 将这种分配的权利从服务端下放到客户端中的 consumer 来管理，这样用户就可以自定义自己的分配方案。

pulsar

类似 kafka 这样的 Stream MQ，更多时候适合做离线业务的处理与分析，很多线上业务会使用 Active MQ 这样 Queue 的 MQ。为了同时兼容这两种消费模型，pulsar 做了一层消费层的抽象，统一了 Queue 和 Stream 这两种消费模型，具体如下图所示：

其中，Exclusive 和 Failover 属于 Stream 的消费模型，Share 属于 Queue 的消费模型。在写此文章时，pulsar 最新版本为 2.3.1，Key_Shared 属于pulsar 新增加的一种订阅模型，在之后的文章中，我们会单独对 Key_shared 订阅模型做单独的分享，这里不在赘述。

对 Stream 支持的对比

由于 kafka 不支持 Queue 类型的消费模型，所以 Share 这种形式在这里不做对比。下面，和大家一起讨论以下在 Stream 下 pulsar 与 kafka 的消费模型。

如下图所示，左边为 pulsar 在 Failover 和 Exclusive 下的消费情况，右边为 kafka 的消费模型。

假设目前有一个 topic，topic name 为 topic1，有 5 个partition，分别为：topic1-p1，topic1-p2，topic1-p3，topic1-p4，topic1-p5，在 kafka 中，使用了 consumer-group 且该 group 下有三个 consumer，上文中提到，kafka 支持 reblance 机制，所以当 consumer-2 与 consumer-3 加入 consumer-group 的过程中，会动态分摊之前 consumer-1 的消费压力，表现为如上图右半部分所示，cousumer-1 消费 topic1-p1 和 ropic1-p2，consumer-2 消费 topic1-p3 和 topic1-p4，consumer-3 消费 topic1-p5 。所以当用户不断的往 consumer-group 中添加 consumer 时，利用 kafka 的 reblance 机制，是可以让用户动态指定具体哪一个 consumer 来消费 topic1 中的哪些 partition。

在 pulsar 中，你可以将 subscribe 理解为 kafka 中的 consumer-group，如果用户在启动 consumer 时，指定的 subscribe-name 是相同的，说明这两个 consumer 属于同一个订阅组，代码示例如下：

如上图示例所示，在同一个订阅组下，启动三个 consumer，在 pulsar 中，每一个 consumer 都会去订阅 topic1 中的 5 个 partition，所以每个 consumer 都会去启动 5 个 sub-consumer，在 failover 的订阅模型下，会使用 hashcode 的形式，将 5 个 partition 分配给三个 consumer 来消费，pulsar 将当前正在消费的 sub-consumer 看作是处于 leader 状态的 consumer，剩余未工作的 sub-consumer 作为从节点，当 leader 状态的 consumer 由于某些原因无法工作时，处于从状态的 sub-consumer 会去接替 leader 的 consumer，并继续工作。可以发现，kafka 加入 reblance 的机制，允许用户自己指定哪些 consumer 来消费 哪些 partition，在 pulsar 中，这个工作由 failover 的机制来完成，它通过 hash 的形式，将 consumer 分配到不同的 sub-consumer 中来执行。

现在，验证一下上述所描述的内容。

场景一

1. 以 standalone 的形式启 pulsar

$ docker run -it \

2. 创建一个 topic，partition 的数目为 4

$ ./bin/pulsar-admin topics mytopic1 create-partitioned-topic -p 4

以 failover 的订阅类型，启动 3 个 consumer，并指定他们为同一个订阅组，即-s sub-1

$ ./bin/pulsar-client consume mytopic1 -s sub-1 -n 0 -t Failover

3. 启动 producer，发送 10 条数据到 mytopic1

$ ./bin/pulsar-client produce mytopic1 -n 10 -m "hello-pulsar"

结果如下所示：

可以看到，consumer1 接收到 2 条消息，consumer2 接收到 5 条消息，consumer3 接收到 3 条消息。效果和我们所预期的是一致的。

上述情况是因为在 producer 发送之前，就已经启动好三个 consumer 来消费消息，所以 pulsar 会以 hash 的形式将消息分发到三个 consumer 中来消费。

场景二

以 Exclusive 的订阅形式启动两个 consumer，效果如下：

./bin/pulsar-client consume mytopic1 -s sub-1 -n 0 -t Exclusive

可以看到，当启动 consumer2 时，会报错 Exclusive consumer is already connected，这是因为，Failover 的订阅模式下，其它的 consumer 会以 “从” consumer 的形态存在，但是 Exclusive 只允许一个 consumer 订阅一个 topic。

本文分享自微信公众号 - 大数据技术与架构（import_bigdata）。
如有侵权，请联系 support@oschina.cn 删除。
本文参与“OSC源创计划”，欢迎正在阅读的你也加入，一起分享。