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

Posted 2021-08-29 民工哥

tags:

篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了都 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/