最近这个 Apache Pulsar 消息中间件非常的火,号称下一代消息中件,今天,就一起来看看它到底有多牛逼?
概述
Apache Pulsar 是一个使用 Apache Bookkeeper 提供持久化的 pub/sub 消息平台,是一个用于服务端到服务端的消息中间件,最初由Yahoo开发并在2016年开源,目前正在Apache基金会下孵化。它可以提供如下特性:
- 跨地域复制
- 多租户
- 零数据丢失
- 零Rebalancing时间
- 统一的队列和流模型
- 高可扩展性
- 高吞吐量
- Pulsar Proxy
- 函数
架构
Pulsar 使用分层结构,将存储机制与 broker 隔离开来。此体系结构为 Pulsar 提供以下好处:
- 独立扩展broker
- 独立扩展存储(Bookies)
- 更容易容器化Zookeeper, Broker and Bookies
- ZooKeeper提供集群的配置和状态存储
在 Pulsar 集群中,一个或多个代理处理和负载均衡来自生产者的传入消息,将消息分派给消费者,与 Pulsar 配置存储通信以处理各种协调任务,将消息存储在 BookKeeper 实例(又名 bookies)中,依赖特定于集群的 ZooKeeper 集群任务等等。
- 由一个或多个 bookie 组成的 BookKeeper 集群处理消息的持久存储。
- 特定于该集群的 ZooKeeper 集群处理 Pulsar 集群之间的协调任务。
更多关于 Pulsar 的架构介绍请参阅:https://pulsar.apache.org/docs/en/concepts-architecture-overview/
四种订阅模式
Pulsar 中有四种订阅模式:exclusive、shared、failover和key_shared。这些模式如下图所示。
详细介绍参阅:https://pulsar.apache.org/docs/en/concepts-messaging/
性能优于 Kafka
Pulsar 表现最出色的就是性能,Pulsar 的速度比 Kafka 快得多,与 Kafka 相比,Pulsar 的速度提升了 2.5 倍,延迟降低了 40%。
数据来源:https://streaml.io/pdf/Gigaom-Benchmarking-Streaming-Platforms.pdf
注:对比是针对 1 个分区的 1 个主题,其中包含 100 字节消息,Pulsar 每秒可发送 220,000 条消息。
安装
二进制版本安装 Pulsar
代码语言:javascript复制#下载官方二进制包
[root@centos7 ~]# wget https://archive.apache.org/dist/pulsar/pulsar-2.8.0/apache-pulsar-2.8.0-bin.tar.gz
#解压
[root@centos7 ~]# tar zxf apache-pulsar-2.8.0-bin.tar.gz
[root@centos7 ~]# cd apache-pulsar-2.8.0
[root@centos7 apache-pulsar-2.8.0]# ll
total 72
drwxr-xr-x 3 root root 225 Jan 22 2020 bin
drwxr-xr-x 5 root root 4096 Jan 22 2020 conf
drwxr-xr-x 3 root root 132 Jul 6 11:47 examples
drwxr-xr-x 4 root root 66 Jul 6 11:47 instances
drwxr-xr-x 3 root root 16384 Jul 6 11:47 lib
-rw-r--r-- 1 root root 31639 Jan 22 2020 LICENSE
drwxr-xr-x 2 root root 4096 Jan 22 2020 licenses
-rw-r--r-- 1 root root 6612 Jan 22 2020 NOTICE
-rw-r--r-- 1 root root 1269 Jan 22 2020 README
#bin目录下就有直接启动的命令
Docker安装(重点介绍)
代码语言:javascript复制[root@centos7 ~]# docker run -it
-p 6650:6650
-p 8080:8080
--mount source=pulsardata,target=/pulsar/data
--mount source=pulsarconf,target=/pulsar/conf
apachepulsar/pulsar:2.8.0
bin/pulsar standalone
http协议访问使用8080端口,pulsar协议(Java、Python等客户端)访问使用6650端口。
官方提供的可视化工具 Pulsar Manager,可以对多个Pulsar进行可视化管理。https://pulsar.apache.org/docs/en/administration-pulsar-manager/
代码语言:javascript复制[root@centos7 ~]# docker pull apachepulsar/pulsar-manager:v0.2.0
[root@centos7 ~]# docker run -it
-p 9527:9527 -p 7750:7750
-e SPRING_CONFIGURATION_FILE=/pulsar-manager/pulsar-manager/application.properties
apachepulsar/pulsar-manager:v0.2.0
设置管理员用户与密码
代码语言:javascript复制[root@centos7 ~]# CSRF_TOKEN=$(curl http://localhost:7750/pulsar-manager/csrf-token)
curl
-H 'X-XSRF-TOKEN: $CSRF_TOKEN'
-H 'Cookie: XSRF-TOKEN=$CSRF_TOKEN;'
-H "Content-Type: application/json"
-X PUT http://localhost:7750/pulsar-manager/users/superuser
-d '{"name": "admin", "password": "admin123", "description": "test", "email": "mingongge@test.org"}'
{"message":"Add super user success, please login"}
浏览器直接输入 http://server_ip:9527 登录如下
输入刚刚创建的用户与密码,配置管理的服务端
列表
Toptic列表
Toptic详细信息
客户端配置
Java客户端
下面是一个使用共享订阅的 Java 消费者配置示例:
代码语言:javascript复制import org.apache.pulsar.client.api.Consumer;
import org.apache.pulsar.client.api.PulsarClient;
import org.apache.pulsar.client.api.SubscriptionType;
String SERVICE_URL = "pulsar://localhost:6650";
String TOPIC = "persistent://public/default/mq-topic-1";
String subscription = "sub-1";
PulsarClient client = PulsarClient.builder()
.serviceUrl(SERVICE_URL)
.build();
Consumer consumer = client.newConsumer()
.topic(TOPIC)
.subscriptionName(subscription)
.subscriptionType(SubscriptionType.Shared)
// If you'd like to restrict the receiver queue size
.receiverQueueSize(10)
.subscribe();
Python客户端
下面是一个使用共享订阅的 Python 消费者配置示例:
代码语言:javascript复制from pulsar import Client, ConsumerType
SERVICE_URL = "pulsar://localhost:6650"
TOPIC = "persistent://public/default/mq-topic-1"
SUBSCRIPTION = "sub-1"
client = Client(SERVICE_URL)
consumer = client.subscribe(
TOPIC,
SUBSCRIPTION,
# If you'd like to restrict the receiver queue size
receiver_queue_size=10,
consumer_type=ConsumerType.Shared)
C 客户端
下面是一个使用共享订阅的 C 消费者配置示例:
代码语言:javascript复制#include <pulsar/Client.h>
std::string serviceUrl = "pulsar://localhost:6650";
std::string topic = "persistent://public/defaultmq-topic-1";
std::string subscription = "sub-1";
Client client(serviceUrl);
ConsumerConfiguration consumerConfig;
consumerConfig.setConsumerType(ConsumerType.ConsumerShared);
// If you'd like to restrict the receiver queue size
consumerConfig.setReceiverQueueSize(10);
Consumer consumer;
Result result = client.subscribe(topic, subscription, consumerConfig, consumer);
更多配置及操作指南,官方的文档写的都很清楚,官方文档:https://pulsar.apache.org/docs/