[大数据] 都 2021 年了，你还在用 Kafka？快试试这个全新平台吧

点击关注公众号，回复“1024”获取2TB学习资源！

最近这个 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

#下载官方二进制包
[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安装（重点介绍）

[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/

[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

设置管理员用户与密码

[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 消费者配置示例：

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 消费者配置示例：

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++ 消费者配置示例：

#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/

总结

Plusar 作为下一代分布式消息队列，拥有非常多吸引人的特性，也弥补了一些其他竞品的短板，例如地域复制、多租户、扩展性、读写隔离等等。

我的新书：《 Linux系统运维指南》已出版