简介

Apache Pulsar 是一个分布式消息队列，它主要由以下三部分组成。

组件作用
Broker负责 producer 和 consumer 的请求还有消息的复制与分发，Broker 无状态不存储数据
Zookeeper存储元数据、集群配置，负责任务协调还有服务发现等
Bookkeeper消息数据还有 cursors 数据的持久化存储，Bookkeeper 的每一个存储节点叫做 bookie

producer 往 Pulsar 发送数据，consumer 从 Pulsar 接受数据，consumer 接收数据的过程叫做 subscription（订阅）。Pulsar 有四种订阅模式

模式名模式
独占（exclusive）一个 subscription 只能有一个 consumer，如果多个 consumer 使用相同的 subscription 去订阅一个 topic 就会报错
故障转移（failover）一个 subscription 允许多个 comsumer，但是只有一个 consumer 会工作。当当前 consumer 失去连接时，其它的 consumer 才会上线消费
共享（shared）多个 consumer 可以使用同一个 subscription 去订阅一个 topic，消息通过轮询的方式发送给 consumer
共享键（key_shared）和上面类似，只是消息会根据 key 的不同发送给不同的 consumer

Broker 不存储数据，数据存储在 Bookkeeper 上。topic 会分为多个 partition，partition 则会分配到不同的 broker 上，producer 和 consumer 会与 broker 上的 partition 连接从而实现数据的发送和接收。Pulsar 支持多层级的 topic，可以设置是否持久化以及租户、命名空间还有 topic 的名称

{persistent|non-persistent}://tenant/namespace/topic

安装

我们有三个节点172.19.67.171，172.19.67.190，172.19.67.202。首先我们在这三个节点上下载 Pulsar 的安装包

useradd pulsar
su - pulsar
wget https://archive.apache.org/dist/pulsar/pulsar-2.10.0/apache-pulsar-2.10.0-bin.tar.gz
tar -zxvf apache-pulsar-2.10.0-bin.tar.gz
cd apache-pulsar-2.10.0

1. 安装 Zookeeper 集群，初始化 Pulsar 集群元数据

修改每个节点的conf/zookeeper.conf配置文件，添加如下配置

server.1=172.19.67.171:2888:3888
server.2=172.19.67.190:2888:3888
server.3=172.19.67.202:2888:3888

之后针对每个节点的序号，把对应的序号设置到该节点的dataDir目录下的myid文件中

mkdir -p data/zookeeper
echo 1 > data/zookeeper/myid

修改完配置文件之后，在每个节点上启动 Zookeeper 的服务

bin/pulsar-daemon start zookeeper

启动好了集群之后使用如下命令写入 Pulsar 的元数据

bin/pulsar initialize-cluster-metadata \
    --cluster pulsar-cluster-1 \
    --zookeeper 172.19.67.171:2181 \
    --configuration-store 172.19.67.171:2181 \
    --web-service-url http://172.19.67.171:8080,172.19.67.190:8080,172.19.67.202:8080 \
    --broker-service-url pulsar://172.19.67.171:6650,172.19.67.190:6650,172.19.67.202:6650

具体含义如下

参数含义
clusterPulsar 的集群名称
zookeeperZookeeper 的地址
configuration-store配置存储地址，使用 Zookeeper 的地址
web-service-urlPulsar 集群 web 服务的地址
broker-service-urlbroker 服务的地址

2. 安装 Bookkeeper 集群

设置所有节点的conf/bookkeeper.conf配置文件，添加 Zookeeper 连接信息

zkServers=172.19.67.171:2181,172.19.67.190:2181,172.19.67.202:2181

之后在每个节点启动 Bookkeeper 服务

bin/pulsar-daemon start bookie

之后使用如下命令验证 Bookkeeper 的集群状态

bin/bookkeeper shell simpletest --ensemble 3 --writeQuorum 3 --ackQuorum 3 --numEntries 3

3. 安装 Pulsar Brokers

修改所有节点的conf/broker.conf配置文件

# 配置 pulsar broker 连接的 zookeeper 集群地址
zookeeperServers=172.19.67.171:2181,172.19.67.190:2181,172.19.67.202:2181
configurationStoreServers=172.19.67.171:2181,172.19.67.190:2181,172.19.67.202:2181
# broker 数据端口
brokerServicePort=6650
# broker web 服务端口
webServicePort=8080
# pulsar 集群名字，和前面 zookeeper 初始化集群元数据时配置的一样
clusterName=pulsar-cluster-1
# 创建一个 ledger 时使用的 bookie 数量
managedLedgerDefaultEnsembleSize=2
# 每个消息的副本数量
managedLedgerDefaultWriteQuorum=2
# 完成写操作前等待副本 ack 的数量
managedLedgerDefaultAckQuorum=2

之后在每个节点启动 Pulsar 服务

bin/pulsar-daemon start broker

使用客户端发送和消费消息

修改conf/client.conf文件

webServiceUrl=http://172.19.67.171:8080,172.19.67.190:8080,172.19.67.202:8080
brokerServiceUrl=pulsar://172.19.67.171:6650,172.19.67.190:6650,172.19.67.202:6650

之后使用客户端进行消费

bin/pulsar-client consume \
    persistent://public/default/pulsar-test \
    -n 100 \
    -s "consumer-test" \
    -t "Exclusive"

新开一个窗口，使用生产者发送消息

bin/pulsar-client produce \
    persistent://public/default/pulsar-test \
    -n 1 \
    -m "hello, this is a test for Pulsar"

参考

Pulsar 介绍与部署