Pulsar与Kafka消费模型对比

2024-09-06 21:58

本文主要是介绍Pulsar与Kafka消费模型对比,希望对大家解决编程问题提供一定的参考价值,需要的开发者们随着小编来一起学习吧!

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 这两种消费模型,具体如下图所示:

640?wx_fmt=other

其中,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 的消费模型。

640?wx_fmt=other


假设目前有一个 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<byte[]> consumer1 = pulsarClient.newConsumer().topic("topic-1").subscriptionName("my-subscriber-name")	.subscriptionType(SubscriptionType.Failover)	.subscribe();	Consumer<byte[]> consumer2 = pulsarClient.newConsumer().topic("topic-2").subscriptionName("my-subscriber-name")	.subscriptionType(SubscriptionType.Failover)	.subscribe();


如上图示例所示,在同一个订阅组下,启动三个 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 \	-p 6650:6650 \	-p 8080:8080 \	-v $PWD/pulsardata:/pulsar/data \	apachepulsar/pulsar:2.3.0 \	bin/pulsar standalone

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"

结果如下所示:

640?wx_fmt=other

可以看到,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

640?wx_fmt=other


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


640?wx_fmt=gif

640?wx_fmt=jpeg

这篇关于Pulsar与Kafka消费模型对比的文章就介绍到这儿,希望我们推荐的文章对编程师们有所帮助!



http://www.chinasem.cn/article/1143228

相关文章

Java的IO模型、Netty原理解析

《Java的IO模型、Netty原理解析》Java的I/O是以流的方式进行数据输入输出的,Java的类库涉及很多领域的IO内容:标准的输入输出,文件的操作、网络上的数据传输流、字符串流、对象流等,这篇... 目录1.什么是IO2.同步与异步、阻塞与非阻塞3.三种IO模型BIO(blocking I/O)NI

基于Flask框架添加多个AI模型的API并进行交互

《基于Flask框架添加多个AI模型的API并进行交互》:本文主要介绍如何基于Flask框架开发AI模型API管理系统,允许用户添加、删除不同AI模型的API密钥,感兴趣的可以了解下... 目录1. 概述2. 后端代码说明2.1 依赖库导入2.2 应用初始化2.3 API 存储字典2.4 路由函数2.5 应

Python实现Microsoft Office自动化的几种方式及对比详解

《Python实现MicrosoftOffice自动化的几种方式及对比详解》办公自动化是指利用现代化设备和技术,代替办公人员的部分手动或重复性业务活动,优质而高效地处理办公事务,实现对信息的高效利用... 目录一、基于COM接口的自动化(pywin32)二、独立文件操作库1. Word处理(python-d

Java常用注解扩展对比举例详解

《Java常用注解扩展对比举例详解》:本文主要介绍Java常用注解扩展对比的相关资料,提供了丰富的代码示例,并总结了最佳实践建议,帮助开发者更好地理解和应用这些注解,需要的朋友可以参考下... 目录一、@Controller 与 @RestController 对比二、使用 @Data 与 不使用 @Dat

python中字符串拼接的几种方法及优缺点对比详解

《python中字符串拼接的几种方法及优缺点对比详解》在Python中,字符串拼接是常见的操作,Python提供了多种方法来拼接字符串,每种方法有其优缺点和适用场景,以下是几种常见的字符串拼接方法,需... 目录1. 使用 + 运算符示例:优缺点:2. 使用&nbsjsp;join() 方法示例:优缺点:3

C++ 各种map特点对比分析

《C++各种map特点对比分析》文章比较了C++中不同类型的map(如std::map,std::unordered_map,std::multimap,std::unordered_multima... 目录特点比较C++ 示例代码 ​​​​​​代码解释特点比较1. std::map底层实现:基于红黑

Golang中拼接字符串的6种方式性能对比

《Golang中拼接字符串的6种方式性能对比》golang的string类型是不可修改的,对于拼接字符串来说,本质上还是创建一个新的对象将数据放进去,主要有6种拼接方式,下面小编就来为大家详细讲讲吧... 目录拼接方式介绍性能对比测试代码测试结果源码分析golang的string类型是不可修改的,对于拼接字

一文详解kafka开启kerberos认证的完整步骤

《一文详解kafka开启kerberos认证的完整步骤》这篇文章主要为大家详细介绍了kafka开启kerberos认证的完整步骤,文中的示例代码讲解详细,感兴趣的小伙伴可以跟随小编一起学习一下... 目录一、kerberos安装部署二、准备机器三、Kerberos Server 安装1、配置krb5.con

C#集成DeepSeek模型实现AI私有化的流程步骤(本地部署与API调用教程)

《C#集成DeepSeek模型实现AI私有化的流程步骤(本地部署与API调用教程)》本文主要介绍了C#集成DeepSeek模型实现AI私有化的方法,包括搭建基础环境,如安装Ollama和下载DeepS... 目录前言搭建基础环境1、安装 Ollama2、下载 DeepSeek R1 模型客户端 ChatBo

MySQL表锁、页面锁和行锁的作用及其优缺点对比分析

《MySQL表锁、页面锁和行锁的作用及其优缺点对比分析》MySQL中的表锁、页面锁和行锁各有特点,适用于不同的场景,表锁锁定整个表,适用于批量操作和MyISAM存储引擎,页面锁锁定数据页,适用于旧版本... 目录1. 表锁(Table Lock)2. 页面锁(Page Lock)3. 行锁(Row Lock