Pulsar简介及Pulsar部署、原理和使用介绍

Pulsar简介

诞生背景

Apache Pulsar 是一个企业级的分布式消息系统，最初由 Yahoo 开发，在 2016 年开源，并于2018年9月毕业成为 Apache 基金会的顶级项目。Pulsar 已经在 Yahoo 的生产环境使用了三年多，主要服务于Mail、Finance、Sports、 Flickr、 the Gemini Ads platform、 Sherpa (Yahoo 的 KV 存储)。

Pulsar是一种用于服务器到服务器消息传递的多租户，高性能解决方案。Pulsar最初由雅虎公司开发，由Apache Software Foundation负责管理。

Pulsar特点

• 对Pulsar实例中的多个群集的本机支持，跨群集的消息的无缝地理复制。
• 极低的发布和端到端延迟。
• 无缝扩展到超过一百万个主题。
• 一个简单的客户端API，包含Java，Go，Python和C ++的绑定。
• 主题的多种订阅模式（独占，共享和故障转移）。
• Apache BookKeeper提供的持久消息存储保证消息传递。
• 无服务器轻量级计算框架Pulsar Functions提供流本地数据处理功能。
• 基于Pulsar函数构建的无服务器连接器框架Pulsar IO可以更轻松地将数据移入和移出Apache Pulsar。
• 当数据老化时，分层存储将数据从热/暖存储卸载到冷/长期存储（例如S3和GCS）。

消息传递

Pulsar基于publish-subscribe(pub-sub)生产订阅模式，生产者将消息发布到Topic，消费者可以订阅这些主题来处理消息，并在处理完成后发送确认消息

关键词	词解
value	Pulsar中的数据存储为byte
key	消息可以被Key打标签。这对诸如topic压缩之类的事情有作用
properties	一个可以选择的map，用于用户配置参数
Sequence ID	每条消息都放在对应Topic的有序序列中，这个字段记录的是消息所在的序列顺序
publish time	消息的发布时间，消息发布的时间戳(producer自动附上)
event time	应用程序可以附加到消息上的时间戳，用户业务需求的时间标注，可选的时间戳，应用可以附在消息上，代表某个事件发生的时间，例如，消息被处理时间。如果没有明确的设置，那么事件时间为0。

消息生产者

发送模式

生产者将消息发布到Topic上，发送消息分可为同步发送和异步发送两种模式：

Mode	说明：
同步发送	产者每次发送完消息后需要等到broker的ack确认，如果没有接收到确认信息生产者就认为发送消息失败
异步发送	生产者将消息放入到阻塞队列中就直接返回。Pulsar的客户端通过后台线程将消息发送给broker，如果队列满了，生产者再放入消息时会被告知推送失败

消息压缩

生产者发布消息在传输过程中会对数据进行压缩，目前Pulsar支持的压缩方式有LZ4，ZLIB, ZSTD, SNAPPY。如果启用了批处理，生产者将在单个请求中累积一批消息进行发送，批处理大小可以由最大消息数和最大发布延迟定义

批处理发送（Batching）

如果批处理开启，producer将会累积一批消息，然后通过一次请求发送出去。批处理的大小取决于最大的消息数量及最大的发布延迟。

消息消费者

消费模式

消费者从Topic上接收消息进行数据处理，同样，消息接收也分为同步接收和异步接收两种模式：

Mode	说明：
同步接收	同步接收一直处于阻塞状态，直到有消息传入
异步接收	异步接收立即返回一个未来future值，一旦有新消息，它就直接完成，例如java中的 CompletableFuture

消费确认（ack）

1. 消费者成功接收到消息时：

   当消费者成功处理完一条消息后，会发送一个确认请求给broker，告诉broker可以删除这条消息了,否则broker会一直存储这条消息。消息可以逐个确认也可以累积确认，消费者只需要确认收到的最后一条消息，这个流中所涉及到的所有消息都不会再重新传递给这个消费者。

2. 消费者不成功消费时

   当消费者处理消息失败时，会给broker发送一个失败确认，这个时候broker就会给消费者重新发送这条消息，失败确认可以逐条发送，也可以累积发送，这取决于消费订阅模式。在exclusive和failover订阅模式中，消费者只会对收到的最后一条消息进行失败确认。在Pulsar客户端可以通过设置timeout的方式触发broker自动重新传递消息，如果在timeout范围内消费者都没有发送确认请求，那么broker就会自动重新发送这条消息给消费者。

3. 确认超时

   如果某条消息一直处理失败就会触发broker一直重发这条消息给消费者，使消费者无法处理其他消息，Dead letter topic机制可以让消费者在无法成功消费某些消息时接收新的消息进行消费，在这种机制下，无法消费的消息存储在一个单独的topic中（Dead letter topic），用户可以决定如何处理这个topic中的消息。

消息的持久化

消息的持久化是通过BookKeeper实现的，一旦创建了订阅关系，Pulsar将保留所有的消息（即使消费者断开了链接），只有当消费者确认已经成功处理保留的消息时，才会将这些消息丢弃消息。

消息的保留分为两种：

​ 1.在保留策略内的消息即使消费者已发送了确认也可以持久地存储在Pulsar中，保留策略未涵盖的已确认消息将被删除，如果没有保留策略所有已确认的消息都将被删除；

​ 2.设置消息到期时间，会根据应用于namespace的TTL过期时间，如果到期了，即使消息没有被确认也会被删除

当有某条消息被重复发送时，可以选择两种持久化策略：

​ 1.是将重复的消息也持久化到BookKeeper中

​ 2.是判断如果是重复消息，则不再进行持久化操作

租户（tenant）

Pulsar 从一开始就支持多租户，topic 的名称是层级化的，最上层是租户（tenant）

命名空间（namespace）

命名空间是租户内部逻辑上的命名术语。一个租户可以通过admin API创建多个命名空间。例如，一个对接多个应用的租户，可以为每个应用创建不同的namespace。

Topic

与其他pub-sub系统一样，Pulsar中的topic被命名为从生产者向消费者传输消息的通道：

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

关键词	词解
persistent|non-persistent	标识topic类型： 持久：所有消息都持久保存在磁盘上(BookKeeper节点） 非持久：数据只存在内存中，当broker重启后会造成消息丢失
tenant	租户；实例中的topic的租户。租户是pulsar对多租户支持的重要组成，可以分散在集群中
namespace	用作topic的分组机制，大多数topic配置是在命名空间级别执行的，每个tenant租户可以有多个namespace
topic	可以用户自定义，topic的名字是自由格式的，在Pulsar实例中没有特殊含义

producer写入不存在的主题时 了会在提供的命名空间下自动创建该主题

用户不需要在Pulsar中明确地创建主题，如果客户端尝试往不存在的主题中写入/接收信息，Pulsar将在topic提供的namespace下自动创建该主题
消费者可以订阅多个topic：
通过名称配置：persistent://public/default/finance-.*
配置topic订阅列表
常规topic只能由单个broker提供，这限制了topic的最大吞吐量，分区topic是由多个broker处理的一种特殊类型的topic，它允许更高的吞吐量。分区topic和普通topic在订阅模式的工作方式上没有区别，在创建主题时可以指定分区数。

消息路由模式

发布到分布分区topic主题时，必须指定路由模式。默认三个路由模式，默认轮询-和Kafka类似。

模式	描述
RoundRobinPartition	如果未提供key，则以轮询方式往各分区上面发布消息，以实现最大吞吐量 --默认模式
SinglePartition	如果未提供key，则生产者随机选择一个分区并将所有消息发布到此分区。如果指定了key，将对key进行散列（默认javaStringHash=多客户端推荐Murmur3_32Hash）并将消息分配给特定的分区
CustomPartition	使用将调用的自定义消息路由器实现来特定消息的分区。用户在在java clent端实现MessageRouter接口来实现自定义路由模式。

消息订阅模式（subscription）

Pulsar具有exclusive，shared，failover三种订阅模式

独占（exclusive）

exclusive模式：一个topic只允许一个消费者订阅，否者会报错

在 exclusive 模式下，一个 subscription 只允许被一个 consumer 用于订阅 topic ，如果多个 consumer 使用相同的 subscription 去订阅同一个 topic，则会发生错误。exclusive 是默认的订阅模式。如下图所示，Consumer A-0 和 Consumer A-1 都使用了相同的 subscription（相同的消费组），只有 Consumer A-0 被允许消费消息。

故障转移|灾备（failover）

failover模式：多个消费者订阅同一个topic，按照消费者名称进行排序，第一个消费者时唯一接收到消息的消费者（主消费者），当主消费者断开连接时，所有的后续消息都将发给下一个消费者

在 failover 模式下，多个 consumer 允许使用同一个 subscription 去订阅 topic。但是对于给定的 topic，broker 将选择⼀个 consumer 作为该 topic 的主 consumer ，其他 consumer 将被指定为故障转移 consumer 。当主 consumer 失去连接时，topic 将被重新分配给其中⼀个故障转移 consumer ，⽽新分配的 consumer 将成为新的主 consumer 。发⽣这种情况时，所有未确认的消息都将传递给新的主 consumer ，这个过程类似于 Kafka 中的 consumer 组重平衡（rebalance）。

如下图所示，Consumer B-0 是 topic 的主 consumer ，当 Consumer B-0 失去连接时，Consumer B-1 才能成为新的主 consumer 去消费 topic。

共享（shared）

shared模式：多个消费者订阅同一个topic，消息在消费者之间以循环的方式发送，并且给定的某条消息只能发送给一个消费者，当消费者断开连接时，所有发送给它但没有确认的消息将重新安排发送给其他消费者

在 shared 模式下，多个 consumer 可以使用同一个 subscription 去订阅 topic。消息以轮询的方式分发给 consumer ，并且每条消费仅发送给一个 consumer 。当有 consumer 失去连接时，所有发送给该 consumer 但未被确认的消息将被重新安排，以便发送给该 subscription 上剩余的 consumer 。

但是消息不能保证有序以及不支持批量ack

如下图所示，Consumer C-1，Consumer C-2，Consumer C-3 以轮询的方式接受消息。

共享键（key_shared）

key_shared模式：多个消费者订阅同一个topic，消息以分布方式在消费者之间传递（<key, value>），具有相同key的消息传递给同一个消费者，当这个消费者断开连接时，将导致key对应的消费者更改

在 shared 模式下，多个 consumer 可以使用同一个 subscription 去订阅 topic。消息按照 key 分发给 consumer ，含有相同 key 的消息只被发送给同一个 consumer 。

如下图所示，不同的 consumer 只接受到对应 key 的消息。

Pulsar原理架构

体系结构

​ 在最高级别中，一个Pulsar实例有一个或多个Pulsar集群组成，实例中的集群可以彼此复制数据。在Pulsar集群中，一个或多个broker处理和加载来自生产者传入的消息，将消息发送给消费者，与Pulsar配置存储通信以处理各种协调任务，Pulsar集群架构如下所示，包括一个或多个broker，用于集群级配置和协调的Zookeeper，用于持久存储消息的BookKeeper，集群可以使用地理复制在集群间进行复制

Pulsar组件

Broker

Pulsar 的 broker 是一个无状态组件，本身不存储数据。主要负责处理 producer 和 consumer 的请求，消息的复制与分发，数据的计算。可以理解成Broker 是 Pulsar 的自身实例

主要有2部分组成：

1. HTTP服务器，向生产者和消费者公开，用于管理任务和topic查找端的REST API；

2. 调度程序，异步TCP服务器，通过用于所有数据传输的自定义二进制协议；

每个集群都有自己的本地Zookeeper用于存储集群特定的配置和协调，如所有权元数据、代理加载报告、簿记员分类帐元数据等等。
Pulsar使用BookKeeper进行持久消息存储，BookKeeper是一个分布式预写日志（WAL）系统，它的优势为：
• 使Pulsar利用多个独立日志，成为ledgers，随着时间推移，可以为topic创建多个ledger
• 为处复制的顺序数据提供了非常有效的存储
• 保证在出现各种系统故障时ledger的读取一致性
• 提供多个Bookies的I/O分布
• 在容量和吞吐量方面都是水平扩展的，可以通过向集群中添加更多的bookies来增加容量
• Bookies用于处理数千个同事读写的ledger，通过使用多个磁盘设备（一个用于日志，一个用于存储），Bookies能够将读写操作的延迟隔离开
• 除了消息数据外，消费者的订阅位置cursor也可以持久地存储在BookKeeper中

每个 topic 的 partition 都会分配到某一个 borker 上，producer 和 consumer 则会连接到这个 broker，从而向该 topic 的 partition 发送和消费消息。broker 主要负责消息的复制与分发，数据的计算。

zookeeper

主要用于存储元数据、集群配置，任务的协调（例如哪个 broker 负责哪个 topic），服务的发现（例如 broker 发现 bookie 的地址）。

bookkeeper

主要用于数据的持久化存储。除了消息数据，cursors（游标） 也会被持久化到 Bookeeper，cursors 是消费端订阅消费的位移。Bookeeper 中每一个存储节点叫做 bookie。

BookKeeper 是一种优化实时工作负载的存储服务，具有可扩展、高容错、低延迟的特点。企业级的实时存储平台应符合以下几项要求：

• 以极低的延迟（小于 5 毫秒）读写 entry 流
• 能够持久、一致、容错地存储数据
• 在写数据时，能够进行流式传输或追尾传输
• 有效地存储、访问历史数据与实时数据

数据存储

数据分区

写入主题的数据可能只有几个MB，也有可能是几个TB。所以，在某些情况下主题的吞吐量很低，有时候又很高，完全取决于消费者的数量。那么碰到有些主题吞吐量很高而有些又很低的情况该怎么处理？为了解决这个问题，Pulsar将一个主题的数据分布到多台机器上，也就是所谓的分区。

在处理海量数据时，为了保证高吞吐量，分区是一种很常见的手段。默认情况下，Pulsar的主题是不进行分区的，但通过命令行工具或API可以很容易地创建分区主题，并指定分区的数量。

在创建好分区主题之后，Pulsar可以自动对数据进行分区，不会影响到生产者和消费者。也就是说，一个应用程序向一个主题写入数据，对主题分区之后，不需要修改应用程序的代码。分区只是一个运维操作，应用程序不需要关心分区是如何进行的。

主题的分区操作由一个叫作broker的进程来处理，Pulsar集群里的每个节点都会运行自己的broker。

数据持久性

Pulsar broker在收到消息并进行确认之后，就必须确保消息在任何情况下都不会丢失。与其他消息系统不同的是，Pulsar使用Apache BookKeeper来保证持久性。BookKeeper提供了低延迟的持久化存储。Pulsar在收到消息之后，将消息发送给多个BookKeeper节点（具体由复制系数来定），节点将数据写入预写式日志（write ahead log），同时在内存里也保存一份。节点在对消息进行确认之前，强制将日志写入到持久化的存储上，因此即使出现电力故障，数据也不会丢失。因为Pulsar broker将数据发给了多个节点，所以只会在大多数节点（quorum）确认写入成功之后它才会将确认消息发给生产者。Pulsar就是通过这种方式来保证即使在出现了硬件故障、网络故障或其他故障的情况下仍然能够保证数据不丢失。在后续的文章中，我们将深入探讨这方面的细节。

Pulsar安装部署

环境准备

Pulsar项目官网： http://pulsar.apache.org/

Pulsar官方下载地址： http://archive.apache.org/dist/pulsar/

本次部署采用最新版： http://archive.apache.org/dist/pulsar/pulsar-2.7.2/apache-pulsar-2.7.2-bin.tar.gz

JDK官网下载地址： https://www.oracle.com/cn/java/technologies/javase/javase-jdk8-downloads.html

机器规划

搭建 Pulsar 集群至少需要 3 个组件：ZooKeeper 集群、BookKeeper 集群和 broker 集群（Broker 是 Pulsar 的自身实例)。这三个组件如下：

ZooKeeper 集群（3 个 ZooKeeper 节点组成）
BookKeeper (bookie) 集群（3 个 BookKeeper 节点组成）
broker 集群（3 个 Pulsar 节点组成）

从组件上看，重复利用机器来部署组件最少需要3台机器，官方建议6台机器（zk单独分离出3台+3台部署bookie和broker），如果组件互无干扰则需要9台机器。这里做实验部署介绍，用3台即可。机器多仅仅是服务启动的机器不同而已，不影响部署理解和流程梳理。

【注意:】 Pulsar 的安装包已包含了搭建集群所需的各个组件库。无需单独下载 ZooKeeper 安装包和 BookKeeper 安装包。

机器申请

准备3台系统干净裸机，本次实验用阿里云申请机器(可以使用按量付费，使用完毕后回收释放)，如果是测试生产环境，操作流程一样，生产环境建议采用多机器，分开部署组件。

1. 准备3台环境干净的机器；或申请阿里云主机 * 3 台 (4C8G CentOS7 )
   172.23.118.214 pulsar01
   172.23.118.215 pulsar02
   172.23.118.216 pulsar03

准备工作

【注意：】本次搭建采用root用户，非root用户可使用sudo命令还执行一些无权限操作

1、更改服务器主机名

172.23.118.214 -> hostnamectl set-hostname pulsar01
172.23.118.215 -> hostnamectl set-hostname pulsar02
172.23.118.216 -> hostnamectl set-hostname pulsar03

2、3台机器增加hosts文件主机解析，简化操作

[root@pulsar01 ~]# vim /etc/hosts

::1     localhost       localhost.localdomain   localhost6      localhost6.localdomain6
127.0.0.1       localhost       localhost.localdomain   localhost4      localhost4.localdomain4

# pulsar
172.23.118.214   pulsar01
172.23.118.215   pulsar02
172.23.118.216   pulsar03

3、配置免密，便于传输目录文件和安装包

[root@pulsar01 ~]# ssh-keygen -t rsa
Generating public/private rsa key pair.
Enter file in which to save the key (/root/.ssh/id_rsa): 
Enter passphrase (empty for no passphrase): 
Enter same passphrase again: 
Your identification has been saved in /root/.ssh/id_rsa.
Your public key has been saved in /root/.ssh/id_rsa.pub.
The key fingerprint is:
SHA256:56uK0ddMedu48pTC639dlwE7bIJLOyqkSQttJEhPrNc root@pulsar01
The key's randomart image is:
+---[RSA 2048]----+
|  .              |
| . o         .   |
|o + .     . . o  |
|.o + E   o o = . |
|  =     S * + . o|
| . + o   @ . = .o|
|  + * . o B = ..o|
|   + + o  .= .. .|
|    . o..o+++.   |
+----[SHA256]-----+
[root@pulsar01 ~]# 
[root@pulsar01 ~]# ssh-copy-id -i .ssh/id_rsa.pub root@172.23.118.214
[root@pulsar01 ~]# ssh-copy-id -i .ssh/id_rsa.pub root@172.23.118.215
[root@pulsar01 ~]# ssh-copy-id -i .ssh/id_rsa.pub root@172.23.118.216
# 测试
[root@pulsar01 ~]# for i in pulsar01 pulsar02 pulsar03;do echo "=== $i ===" && ssh $i hostname;done
=== pulsar01 ===
pulsar01
=== pulsar02 ===
pulsar02
=== pulsar03 ===
pulsar03

4、安装包准备

# 创建工作目录 module用来安装应用，software用来存放安装包
[root@pulsar01 ~]# for i in pulsar01 pulsar02 pulsar03;do ssh $i mkdir -p /opt/{module,software};done
[root@pulsar01 ~]# ls /opt/
module  software
# 把安装包上传或者wget至/opt/software/目录
[root@pulsar01 ~]# cd /opt/software/
[root@pulsar01 software]# ll
total 490480
-rw-r--r-- 1 root root 307228973 May 29 09:04 apache-pulsar-2.7.2-bin.tar.gz
-rw-r--r-- 1 root root 195013152 May 29 09:05 jdk-8u212-linux-x64.tar.gz
[root@pulsar01 software]# scp * pulsar02:/opt/software/ 
[root@pulsar01 software]# scp * pulsar03:/opt/software/

安装部署

1、安装JDK8

​ 在3台服务器上安装JDK（要求版本不低于 JDK 8）

​ 本次部署使用了：jdk-8u212-linux-x64.tar.gz

# 集群机器都需要安装配置jdk
[root@pulsar01 ~]# cd /opt/software/
[root@pulsar01 software]# tar -xf jdk-8u212-linux-x64.tar.gz -C /opt/module/
[root@pulsar01 software]# cd /opt/module/jdk1.8.0_212/
[root@pulsar01 jdk1.8.0_171]# pwd
/opt/module/jdk1.8.0_212
# 增加java环境变量
[root@pulsar01 jdk1.8.0_212]# vim /etc/profile
# JAVA_HOME
export JAVA_HOME=/opt/module/jdk1.8.0_212
export PATH=$PATH:$JAVA_HOME/bin
CLASSPATH=:$JAVA_HOME/lib/
export PATH JAVA_HOME CLASSPATH
[root@pulsar01 jdk1.8.0_212]#
# source引用环境变量
[root@pulsar01 jdk1.8.0_212]# source /etc/profile
# 验证是否生效
[root@pulsar01 jdk1.8.0_212]# java -version
java version "1.8.0_212"
Java(TM) SE Runtime Environment (build 1.8.0_212-b11)
Java HotSpot(TM) 64-Bit Server VM (build 25.212-b11, mixed mode)
[root@pulsar01 jdk1.8.0_212]# echo $JAVA_HOME
/opt/module/jdk1.8.0_212

# pulsar02 | pulsar03 操作相同

2、解压pulsar，配置环境变量

# 解压到/opt/module目录
[root@pulsar01 software]# tar -xf apache-pulsar-2.7.2-bin.tar.gz -C /opt/module/
[root@pulsar01 software]# cd /opt/module/
[root@pulsar01 module]# ll
total 8
drwxr-xr-x 8 root root 4096 May 29 09:18 apache-pulsar-2.7.2
drwxr-xr-x 7   10  143 4096 Apr  2  2019 jdk1.8.0_212
# 更改应用目录名，便于管理
[root@pulsar01 module]# mv apache-pulsar-2.7.2 pulsar
[root@pulsar01 module]# cd pulsar/
# 目录结构
[root@pulsar01 pulsar]# ll
total 84
drwxr-xr-x 3  501 games  4096 May  3 20:07 bin		# Pulsar 命令行工具，比如 pulsar 和 pulsar-admin
drwxr-xr-x 5  501 games  4096 May  3 20:07 conf		# 配置文件，包含ZooKeeper，Bookeeper，Pulsar 等等
drwxr-xr-x 3 root root   4096 May 27 16:12 examples
drwxr-xr-x 4 root root   4096 May 27 16:12 instances
drwxr-xr-x 3 root root  20480 May 27 16:13 lib		# Pulsar 使用的 JAR 文件
-rw-r--r-- 1  501 games 31556 May  3 20:07 LICENSE
drwxr-xr-x 2  501 games  4096 May  3 20:07 licenses
-rw-r--r-- 1  501 games  6599 May  3 20:07 NOTICE
-rw-r--r-- 1  501 games  1269 May  3 20:03 README
[root@pulsar01 pulsar]# 
# 增加pulsar环境变量
[root@pulsar01 pulsar]# vim /etc/profile
#PULSAR_HOME
export PULSAR_HOME=/opt/module/pulsar
export PATH=$PATH:$PULSAR_HOME/bin
# 引用生效
[root@pulsar01 pulsar]# source /etc/profile

# pulsar02 | pulsar03 操作相同

3、安装zookeeper

# 更改zookeeper配置文件，配置必要内容
[root@pulsar01 pulsar]# cd conf/
# 配置文件最后增加如下配置，IP需对应修改
[root@pulsar01 conf]# vim zookeeper.conf

server.1=172.23.118.214:2888:3888
server.2=172.23.118.215:2888:3888
server.3=172.23.118.216:2888:3888

# 完成配置文件内容
[root@pulsar01 conf]# cat zookeeper.conf |grep -vE "^$|^#"
tickTime=2000
initLimit=10
syncLimit=5
dataDir=data/zookeeper
clientPort=2181
admin.enableServer=true
admin.serverPort=9990
autopurge.snapRetainCount=3
autopurge.purgeInterval=1
forceSync=yes
server.1=172.23.118.214:2888:3888
server.2=172.23.118.215:2888:3888
server.3=172.23.118.216:2888:3888

# 根据配置文件dataDir=data/zookeeper配置，建立对应目录(目录是相对路径，相对于pulsar应用目录下)
[root@pulsar01 conf]# cd /opt/module/pulsar/
[root@pulsar01 pulsar]# mkdir -p data/zookeeper
[root@pulsar01 pulsar]# cd data/zookeeper/
# 每个Zookeeper节点的ID号不能重复，并且和server.N的编号对应，N依次为1，2，3(pulsar01->1 / pulsar02->2 / pulsar03->3)
[root@pulsar01 zookeeper]# echo 1 > myid
[root@pulsar01 zookeeper]# pwd
/opt/module/pulsar/data/zookeeper
[root@pulsar01 zookeeper]# ll
total 4
-rw-r--r-- 1 root root 2 May 27 16:37 myid
[root@pulsar01 zookeeper]# cat myid 
1
[root@pulsar01 zookeeper]# 
############################
# 【pulsar02机器】
[root@pulsar02 conf]# cd /opt/module/pulsar/
[root@pulsar02 pulsar]# mkdir -p data/zookeeper
[root@pulsar02 ~]# cd /opt/module/pulsar/data/zookeeper/
[root@pulsar02 zookeeper]# ls
myid
# 把myid文件中内容改为2
[root@pulsar02 zookeeper]# vim myid 
2
[root@pulsar02 pulsar]# 

############################
# 【pulsar03机器】
[root@pulsar03 conf]# cd /opt/module/pulsar/
[root@pulsar03 pulsar]# mkdir -p data/zookeeper
[root@pulsar03 software]# cd /opt/module/pulsar/data/zookeeper/
[root@pulsar02 zookeeper]# ls
myid
# 把myid文件中内容改为3
[root@pulsar03 zookeeper]# vim myid 
3
[root@pulsar03 pulsar]# 

4、启动zookeeper

# pulsar01
[root@pulsar01 zookeeper]# pulsar-daemon start zookeeper
[root@pulsar01 zookeeper]# jps
20237 ZooKeeperStarter
20415 Jps
[root@pulsar01 zookeeper]# netstat -tnlpu|grep 20237
tcp        0      0 172.23.118.214:3888      0.0.0.0:*               LISTEN      20237/java          
tcp        0      0 0.0.0.0:8000            0.0.0.0:*               LISTEN      20237/java          
tcp        0      0 0.0.0.0:2181            0.0.0.0:*               LISTEN      20237/java          
tcp        0      0 0.0.0.0:9990            0.0.0.0:*               LISTEN      20237/java  

# pulsar02
[root@pulsar02 zookeeper]# pulsar-daemon start zookeeper
[root@pulsar02 zookeeper]# jps
20257 Jps
20071 ZooKeeperStartercd .
[root@pulsar02 zookeeper]# netstat -tnlpu|grep 20071
tcp        0      0 172.23.118.215:3888      0.0.0.0:*               LISTEN      20071/java          
tcp        0      0 0.0.0.0:8000            0.0.0.0:*               LISTEN      20071/java          
tcp        0      0 0.0.0.0:2181            0.0.0.0:*               LISTEN      20071/java          
tcp        0      0 0.0.0.0:9990            0.0.0.0:*               LISTEN      20071/java          
tcp        0      0 172.23.118.215:2888      0.0.0.0:*               LISTEN      20071/java  

# pulsar03
[root@pulsar03 zookeeper]# pulsar-daemon start zookeeper
[root@pulsar03 zookeeper]# jps
10870 ZooKeeperStarter
20250 Jps
[root@pulsar03 zookeeper]# netstat -tnlpu|grep 10870
tcp        0      0 172.23.118.216:3888      0.0.0.0:*               LISTEN      10870/java          
tcp        0      0 0.0.0.0:8000            0.0.0.0:*               LISTEN      10870/java          
tcp        0      0 0.0.0.0:2181            0.0.0.0:*               LISTEN      10870/java          
tcp        0      0 0.0.0.0:9990            0.0.0.0:*               LISTEN      10870/java 

5、初始化集群元数据

Zookeeper 集群启动成功后，需要将一些 Pulsar 集群的元信息写入 ZooKeeper 集群的每个节点，由于数据在 ZooKeeper 集群内部会互相同步，因此只需要将元信息写入 ZooKeeper 的一个节点即可：

# 注意挑一台机器执行即可
[root@pulsar01 zookeeper]# pulsar initialize-cluster-metadata \
  --cluster pulsar-cluster-1 \
  --zookeeper 172.23.118.214:2181 \
  --configuration-store 172.23.118.214:2181 \
  --web-service-url http://172.23.118.214:8080,172.23.118.215:8080,172.23.118.216:8080 \
  --broker-service-url pulsar://172.23.118.214:6650,172.23.118.215:6650,172.23.118.216:6650
  
  ...
  ....
09:35:18.340 [main] INFO  org.apache.bookkeeper.stream.storage.impl.cluster.ZkClusterInitializer - Successfully initialized the stream cluster : 
num_storage_containers: 16
09:35:18.341 [Curator-Framework-0] INFO  org.apache.curator.framework.imps.CuratorFrameworkImpl - backgroundOperationsLoop exiting
09:35:18.447 [main] INFO  org.apache.zookeeper.ZooKeeper - Session: 0x100002585920003 closed
09:35:18.447 [main-EventThread] INFO  org.apache.zookeeper.ClientCnxn - EventThread shut down for session: 0x100002585920003
09:35:18.674 [main] INFO  org.apache.zookeeper.ZooKeeper - Session: 0x100002585920000 closed
09:35:18.674 [main-EventThread] WARN  org.apache.pulsar.zookeeper.ZookeeperClientFactoryImpl - Unexpected ZK event received: WatchedEvent state:Closed type:None path:null
09:35:18.674 [main-EventThread] INFO  org.apache.zookeeper.ClientCnxn - EventThread shut down for session: 0x100002585920000
09:35:18.776 [main] INFO  org.apache.zookeeper.ZooKeeper - Session: 0x100002585920001 closed
09:35:18.776 [main-EventThread] WARN  org.apache.pulsar.zookeeper.ZookeeperClientFactoryImpl - Unexpected ZK event received: WatchedEvent state:Closed type:None path:null
09:35:18.776 [main] INFO  org.apache.pulsar.PulsarClusterMetadataSetup - Cluster metadata for 'pulsar-cluster-1' setup correctly
09:35:18.776 [main-EventThread] INFO  org.apache.zookeeper.ClientCnxn - EventThread shut down for session: 0x100002585920001

参数说明如下：

参数	说明
—cluster	pulsar 集群名字
–zookeeper	zookeeper 地址，只需要包含 zookeeer 集群中的任意一台机器即可
–configuration-store	配置存储地址，只需要包含 zookeeer 集群中的任意一台机器即可
–web-service-url	pulsar 集群 web 服务的 URL 以及端口，默认的端口是8080
–broker-service-url	broker 服务的URL，用于与 pulsar 集群中的 brokers 进行交互，默认端口是 6650

6、验证初始化元数据

执行 zookeeper 客户端连接命令，验证初始化情况

[root@pulsar01 zookeeper]# pulsar zookeeper-shell
Connecting to localhost:2181
WATCHER::
WatchedEvent state:SyncConnected type:None path:null
ls /
[admin, bookies, ledgers, managed-ledgers, namespace, stream, zookeeper]
ls /namespace
[]
ls /admin
[clusters, partitioned-topics, policies]
ls /admin/clusters
[global, pulsar-cluster-1]
quit

WATCHER::
WatchedEvent state:Closed type:None path:null
11:13:48.699 [main-EventThread] INFO  org.apache.zookeeper.ClientCnxn - EventThread shut down for session: 0x1000010dffe0004
11:13:48.699 [main] INFO  org.apache.zookeeper.ZooKeeper - Session: 0x1000010dffe0004 closed

【注意1:】 Enter 键进入命令行界面后，行首没标识符，可使用 ZooKeeper 的各种命令，如ls、get等命令。使用quit命令退出

【注意2:】 如果初始话没成功，操作失败的情况，可以在zookeeper中删除两个路径，然后排查问题再次执行初始化

# zk
/namespace
/admin/clusters/pulsar-cluster-1

7、部署 BookKeeper 集群

修改 Bookeeper 配置文件

# 配置zkServers参数 【3台机器修改】
[root@pulsar01 conf]# vim bookkeeper.conf 
zkServers=172.23.118.214:2181,172.23.118.215:2181,172.23.118.216:2181

创建bookie所需要目录

# pulsar01、02、03
[root@pulsar01 conf]# cd /opt/module/pulsar/data/
[root@pulsar01 data]# mkdir bookkeeper

执行初始化元数据命令；若出现提示，输入 Y，继续（只需在一个bookie节点执行一次）

[root@pulsar01 data]# bookkeeper shell metaformat
JMX enabled by default
...
...
11:19:50.778 [main-EventThread] INFO  org.apache.bookkeeper.zookeeper.ZooKeeperWatcherBase - ZooKeeper client is connected now.
Ledger root already exists. Are you sure to format bookkeeper metadata? This may cause data loss. (Y or N) Y
11:19:56.614 [main] INFO  org.apache.bookkeeper.discover.ZKRegistrationManager - Successfully formatted BookKeeper metadata
11:19:56.718 [main] INFO  org.apache.zookeeper.ZooKeeper - Session: 0x3000010a3350000 closed
11:19:56.718 [main-EventThread] INFO  org.apache.zookeeper.ClientCnxn - EventThread shut down for session: 0x3000010a3350000
[root@pulsar01 data]# 

启动BookKeeper

[root@pulsar01 conf]# pulsar-daemon start bookie
doing start bookie ...
starting bookie, logging to /opt/module/pulsar/logs/pulsar-bookie-pulsar01.log
Note: Set immediateFlush to true in conf/log4j2.yaml will guarantee the logging event is flushing to disk immediately. The default behavior is switched off due to performance considerations.

############################################
[root@pulsar02 conf]# pulsar-daemon start bookie
doing start bookie ...
starting bookie, logging to /opt/module/pulsar/logs/pulsar-bookie-pulsar02.log
Note: Set immediateFlush to true in conf/log4j2.yaml will guarantee the logging event is flushing to disk immediately. The default behavior is switched off due to performance considerations.

############################################
[root@pulsar03 zookeeper]# pulsar-daemon start bookie
doing start bookie ...
starting bookie, logging to /opt/module/pulsar/logs/pulsar-bookie-pulsar03.log
Note: Set immediateFlush to true in conf/log4j2.yaml will guarantee the logging event is flushing to disk immediately. The default behavior is switched off due to performance considerations.

节点启动信息储存在一个VERSION文件中

[root@pulsar01 pulsar]# cd /opt/module/pulsar/data/bookkeeper/ledgers/current
[root@pulsar01 current]# cat VERSION 
4
bookieHost: "172.23.118.214:3181"
journalDir: "data/bookkeeper/journal"
ledgerDirs: "1\tdata/bookkeeper/ledgers"
instanceId: "f3d45b7f-f73a-4ded-acdc-3c2bad9e8311"

验证集群状态

在任意一台 Bookeeper 节点上使用 Bookeeper shell 的 simpletest 命令，去校验集群内所有的 bookie 是否都已经启动，3 为 Bookeeper 节点数量。

[root@pulsar01 pulsar]# bookkeeper shell simpletest --ensemble 3 --writeQuorum 3 --ackQuorum 3 --numEntries 3

参数含义如下：

-a,–ackQuorum Ack quorum size (default 2) 当指定数量的 bookie ack 响应时，认为消息写入成功
-e,–ensemble Ensemble size (default 3) 写入数据的 bookie 节点数量
-n,–numEntries Entries to write (default 1000) 一批消息的消息数量
-w,–writeQuorum Write quorum size (default 2) 每条消息副本数量

这个命令会在集群上创建和 bookie 同等数量的 ledger，并往里面写一些条目，然后读取它，最后删除这个 ledger。

8、部署pulsar集群

修改配置文件vim broker.conf

# pulsar01 pulsar02 pulsar03
[root@pulsar01 conf]# vim broker.conf 
# 配置pulsar broker连接的zookeeper集群地址
zookeeperServers=172.23.118.214:2181,172.23.118.215:2181,172.23.118.216:2181
configurationStoreServers=172.23.118.214:2181,172.23.118.215:2181,172.23.118.216:2181
clusterName=pulsar-cluster-1

启动Pulsar集群

[root@pulsar01 software]# pulsar-daemon start broker
[root@pulsar02 software]# pulsar-daemon start broker
[root@pulsar03 software]# pulsar-daemon start broker

查看集群 brokers 节点情况

[root@pulsar03 conf]# pulsar-admin brokers list pulsar-cluster-1
"pulsar01:8080"
"pulsar02:8080"
"pulsar03:8080"

9、配置客户端连接 Pulsar 集群

# pulsar01 pulsar02 pulsar03
[root@pulsar01 conf]# vim client.conf

webServiceUrl=http://172.23.118.214:8080,172.23.118.215:8080,172.23.118.216:8080
# URL for Pulsar Binary Protocol (for produce and consume operations)
# For TLS:
# brokerServiceUrl=pulsar+ssl://localhost:6651/
brokerServiceUrl=pulsar://172.23.118.214:6650,172.23.118.215:6650,172.23.118.216:6650

10、命令行验证生产消费消息

消费：

[root@pulsar01 conf]# pulsar-client consume \
  persistent://public/default/pulsar-test \
  -n 100 \
  -s "consumer-test" \
  -t "Exclusive"

换一个窗口生产：

[root@pulsar01 conf]# pulsar-client produce \
  persistent://public/default/pulsar-test \
  -n 1 \
  -m "Hello Pulsar"

观察消费端控制台输出，如打输出内容content:Hello Pulsar，则流程全部完成；

11、pulsar-dashboard

采用docker 安装 apachepulsar/pulsar-dashboard

# 安装docker
[root@pulsar01 conf]# yum install -y docker
# 启动docker 并开机自启动
[root@pulsar01 conf]# systemctl start docker && systemctl enable docker
# 查看80端口是否被占用
[root@pulsar01 conf]# netstat -tnlpu|grep 80
tcp        0      0 0.0.0.0:8000            0.0.0.0:*               LISTEN      1882/java           
tcp        0      0 0.0.0.0:8080            0.0.0.0:*               LISTEN      13468/java          
udp        0      0 0.0.0.0:68              0.0.0.0:*                           809/dhclient        
# 运行apachepulsar/pulsar-dashboard
[root@pulsar01 conf]# docker run --name pulsar-dashboard -dit -p 80:80 -e SERVICE_URL=http://PULSARSEVERIP:8080 apachepulsar/pulsar-dashboard
# 查看镜像服务是否正常运行
[root@pulsar01 conf]# docker ps
CONTAINER ID        IMAGE                           COMMAND              CREATED             STATUS              PORTS                NAMES
ed6e1ec3da05        apachepulsar/pulsar-dashboard   "/pulsar/start.sh"   47 seconds ago      Up 44 seconds       0.0.0.0:80->80/tcp   pulsar-dashboard
[root@pulsar01 conf]# netstat -tnlpu|grep 80
tcp        0      0 0.0.0.0:8000            0.0.0.0:*               LISTEN      1882/java           
tcp        0      0 0.0.0.0:8080            0.0.0.0:*               LISTEN      13468/java          
tcp6       0      0 :::80                   :::*                    LISTEN      14877/docker-proxy- 
udp        0      0 0.0.0.0:68              0.0.0.0:*                           809/dhclient