作者：高天赐

编辑：Irene

Pulsar 是一个支持多租户的、高性能的消息中间件，本文详细记录了如何在 MacOS 上，利用一台服务器搭建一个简单测试集群的过程。后续我们会介绍如何在多台 Linux 服务器搭建集群。

一、准备资源

• 一台主机（本文以 MacOS 为例）

• JDK 8 运行环境（默认已准备好）

二、集群的组成

• ZooKeeper 集群（3 个 ZooKeeper 节点组成）

• bookie 集群（3 个 BookKeeper 节点组成）

• broker 集群（3 个 Pulsar 节点组成）

三、搭建 ZooKeeper 集群

本小节记录了如何在一台机器上运行多个 ZooKeeper 服务进程，搭建 ZooKeeper 集群的过程。

3.1 下载 ZooKeeper 安装包

从 ZooKeeper 官网下载最新版本的 ZooKeeper。本文以 ZooKeeper 3.4.14 版本为例。

3.2 创建 ZooKeeper 集群

创建 zookeepers 文件夹，把解压好的 zookeeper-3.4.14 文件夹复制到 zookeepers 文件夹中，重名为 server1。用 server1 复制出 server2 和 server3。

3.3 配置 ZooKeeper 集群一个节点：

server1

1.  在 server1 目录下，创建 data 和 dataLog 两个文件夹。

2. 在 server1/conf 目录中，复制 zoo_sample.cfg 文件，重命名为 zoo.cfg 文件。

3. 修改 zoo.cfg 文件中的 dataDir, dataLogDir, clientPort, admin.enableServer 和 admin.serverPort 这 5 个配置参数，并添加如下集群节点信息。修改端口号是为了避免在一台服务器上造成端口号冲突。

# 添加如下集群节点信息server.1=127.0.0.1:2888:3888
server.2=127.0.0.1:2889:3889
server.3=127.0.0.1:2890:3890

zoo.cfg 文件配置说明

在一台服务器上，部署多个实例，需要指定不同的端口号。

clientPort: 3 个 ZooKeeper 节点中分别配置为:2181，2182，2183。

admin.enableServer: 3 个 ZooKeeper 节点中都配置为 true。

admin.serverPort: 3 个 ZooKeeper 节点中分别配置为:9181，9182，9183。

集群节点信息：3 个 ZooKeeper 节点中都配置为:
server.1=127.0.0.1:2888:3888
server.2=127.0.0.1:2889:3889
server.3=127.0.0.1:2890:3890

oo.cfg 配置文件示例

如下是一个 zoo.cfg 文件示例。

# The number of milliseconds of each tick
tickTime=2000

# The number of ticks that the initial

# synchronization phase can take
initLimit=10

# The number of ticks that can pass between

# sending a request and getting an acknowledgement
syncLimit=10

# the directory where the snapshot is stored.

# do not use /tmp for storage, /tmp here is just

# example sakes.
dataDir=/Users/bilahepan/Softwares/zookeepers/server1/data
dataLogDir=/Users/bilahepan/Softwares/zookeepers/server1/dataLog

# the port at which the clients will connect

clientPort=2181

# the maximum number of client connections.

# increase this if you need to handle more clients

#maxClientCnxns=60
admin.enableServer=true
admin.serverPort=9181

#
server.1=127.0.0.1:2888:3888
server.2=127.0.0.1:2889:3889
server.3=127.0.0.1:2890:3890

# Be sure to read the maintenance p of the

# administrator guide before turning on autopurge.
#

#http://zookeeper.apache.org/doc/current/zookeeperAdmin.html#sc_maintenance
#

# The number of snapshots to retain in dataDir

#autopurge.snapRetainCount=3

# Purge task interval in hours

# Set to "0" to disable auto purge feature

#autopurge.purgeInterval=1

zoo.cfg 配置参数解释

tickTime：ZooKeeper 中使用的基本时间单位, 毫秒值，默认是 2000ms。

initLimit：用来配置 ZooKeeper 服务器集群中 Follower 服务器初始化连接 Leader 服务器时最长能忍受多少个 tickTime。这里设置为 10 表示最长容忍时间为 10s。
syncLimit：用来配置 Leader 与 Follower 之间发送消息、请求和应答时间最长能忍受多少个 tickTime。这里设置为 10 表示最长容忍时间为 10s。
dataDir：数据文件目录。
dataLogDir：日志文件目录。
clientPort：监听client连接的端口号。
server.{myid}={ip}:{leader服务器交换信息的端口}:{当leader服务器挂了后, 选举leader的端口}
maxClientCnxns：对于一个客户端的连接数限制，默认是 60。
admin.enableServer：是否启用 ZooKeeper 管理后台。
admin.serverPort：管理后台端口号。

4. 在 server1/data/ 目录下创建名字为 myid 的文件，并写入内容 1。每个 ZooKeeper 节点的 myid 文件内容不能一样，它是不同节点的唯一标识。3 个 ZooKeeper 节点中 myid 文件内容分别为1，2，3。

3.4 配置 ZooKeeper 另外两个节点：

server2，server3

对 ZooKeeper 的另外两个节点 server2，server3 做相应配置。具体操作，参考 server1 配置。

启动 ZooKeeper

启动 ZooKeeper 集群，在终端分别用命令启动 ZooKeeper 节点。

# 启动命令(zkServer.sh start)

./bin/zkServer.sh start

查看节点状态。

# 状态查看命令(zkServer.sh status)

./bin/zkServer.sh status

ZooKeeper 的其他操作命令。

# 连接 zookeeper 客户端

./bin/zkCli.sh -timeout 5000 -server 127.0.0.1:2181

# 查看 zookeeper 节点内容

ls /

四、搭建 bookie 集群

Bookie 是 BookKeeper 的别称。本小节记录了如何在一台机器上运行多个 bookie 服务进程，搭建 bookie 集群的过程。

4.1 下载 BookKeeper 安装包

从 BookKeeper 官网下载 BookKeeper 安装包。本文以 BookKeeper 4.8.2 版本为例。

4.2 创建 BookKeeper 集群

创建 bookkeepers 文件夹，把解压好的 bookkeeper-server-4.8.2 文件夹复制到 bookkeepers 文件夹中，重名为 bookie1。用 bookie1 复制出bookie2 和 bookie3。

4.3 配置 BookKeeper 集群一个节点：

bookie1

在 bookie1/conf 目录下，修改 bk_server.conf 文件内容，主要修改 bookiePort，httpServerPort，storageserver.grpc.port 这 3 个端口号，避免在一台主机上造成端口号冲突。添加 ZooKeeper 集群节点信息，并指定两个文件目录地址。

# 添加如下ZooKeeper 集群节点信息zkServers=localhost:2181,localhost:2182,localhost:2183

# 指定两个文件目录地址journalDirectories=/Users/<username>/Softwares/bookkeepers/bookie1/tmp/bk-txn
ledgerDirectories=/Users/<username>/Softwares/bookkeepers/bookie1/tmp/bk-data

bk_server.conf 文件配置说明

在一台服务器上，部署多个实例，需要指定不同的端口号。

bookiePort: 3个 bookie 节点中分别配置为:3181，3182，3183。

httpServerPort: 3个 bookie 节点中都配置为:8050，8060，8070。

storageserver.grpc.port: 3个 bookie 节点中都配置为: 4181，4182，4183。

zk集群节点信息: 3个 bookie 中都配置为: zkServers=localhost:2181,localhost:2182,localhost:2183

4.4 配置 BookKeeper 另外两个节点：

bookie2，bookie3

对 BookKeeper 另外两个节点 bookie2，bookie3 做对应配置。具体操作参考 bookie1 配置。

4.5 初始化集群

在任一个 bookie 中运行以下命令，初始化集群元数据。

./bin/bookkeeper shell metaformat

4.6 启动 BookKeeper 集群

在终端分别用如下命令启动三个 bookie 节点，启动 BookKeeper 集群。

./bin/bookkeeper bookie

4.7 检查 BookKeeper 启动状态

在任一个 bookie 实例中运行以下命令，检查 BookKeeper 启动状态。
集群启动成功会有 "Bookie sanity test succeeded" 日志输出。

./bin/bookkeeper shell bookiesanity

五、搭建 broker 集群

broker 是 Pulsar 实例别称。本小节记录了如何在一台机器上运行多个 broker 服务进程，搭建 broker 集群的过程。

5.1 下载 Pulsar 安装包

从 Pulsar官网下载最新版本的 Pulsar 安装包。本文以 Pulsar 2.3.1 版本为例。

5.2 创建 broker 集群

创建 brokers 文件夹，把解压好的 apache-pulsar-2.3.1 复制到 brokers 文件夹中，重名为 broker1。用 broker1 复制出 broker2 和 broker3

5.3 配置 broker 集群一个节点：

broker1

在 broker1/conf 目录中，修改 broker.conf 文件内容，主要修改 brokerServicePort，brokerServicePortTls，webServicePort，webServicePortTls 这 4 个端口号，避免在一台主机上造成端口号冲突。添加 ZooKeeper 集群节点信息，指定集群名。

# 添加如下 ZooKeeper 集群节点信息
zookeeperServers=localhost:2181,localhost:2182,localhost:2183
configurationStoreServers=localhost:2181,localhost:2182,localhost:2183

# 指定集群名
clusterName=pulsar-cluster

broker.conf 文件配置说明

在一台服务器上，部署多个实例，需要指定不同的端口号。

brokerServicePort: 3 个 broker 节点中分别配置为:6650，6660，6670。

brokerServicePortTls: 3 个 broker 节点中分别配置为:6651，6661，6671。

webServicePort: 3 个 broker 节点中分别配置为:8080，8081，8082。

webServicePortTls: 3 个 broker 节点中分别配置为:8443，8444，8445。

zk集群节点信息: 3 个 broker 中都配置为:
zookeeperServers=localhost:2181,localhost:2182,localhost:2183 configurationStoreServers=localhost:2181,localhost:2182,localhost:2183

5.4 配置 broker 集群另外两个节点：

broker2，broker3

对 broker 集群另外两个节点 broker2，broker3 做相应配置。具体操作，参考 broker1 配置。

5.5 初始化集群

在 broker1 中运行以下命令，初始化集群元数据。

bin/pulsar initialize-cluster-metadata \
 --cluster pulsar-cluster \
 --zookeeper 127.0.0.1:2181 \
 --configuration-store 127.0.0.1:2181 \
 --web-service-url http://pulsar.cluster.com:8080 \
 --web-service-url-tls https://pulsar.cluster.com:8443 \
 --broker-service-url pulsar://pulsar.cluster.com:6650 \
 --broker-service-url-tls pulsar+ssl://pulsar.cluster.com:6651

集群元数据解释

--cluster
集群名称
--zookeeper
ZooKeeper集群连接参数，仅需要包含集群中的一个节点即可
--configuration-store
Pulsar实例的配置存储集群（ZooKeeper），和--zookeeper参数一样只需要包含集群中的一个节点即可
--web-service-url
集群Web服务的URL+端口，URL必须是一个标准的DNS名称，默认端口8080，不建议修改。
--web-service-url-tls
集群Web提供TLS服务的URL+端口，端口默认8443，不建议修改。
--broker-service-url
集群brokers服务URL，URL中DNS的名称和Web服务保持一致，URL使用pulsar替代http/http，端口默认6650，不建议修改。
--broker-service-url-tls
集群brokers提供TLS服务的URL，默认端口6551，不建议修改。

5.6 启动 broker 集群

在终端分别用如下命令启动三个 broker 节点，启动 broker 集群。启动成功后会有日志 “PulsarService started” 输出。

./bin/pulsar broker

六、测试 demo

成功启动 Pulsar 集群后，用如下命令依次创建集群、租户、命名空间、分区topic，并为命名空间指定集群。

# 创建集群(集群名:pulsar-cluster)
./bin/pulsar-admin clusters create --url http://pulsar.cluster.com:8080  pulsar-cluster

# 创建租户(租户名:my-tenant)
./bin/pulsar-admin tenants create my-tenant

# 创建命名空间(命名空间名,指定了租户my-tenant:my-tenant/my-namespace)
./bin/pulsar-admin namespaces create my-tenant/my-namespace

# 创建持久性分区topic(topic全名:persistent://my-tenant/my-namespace/my-topic)
./bin/pulsar-admin topics create-partitioned-topic persistent://my-tenant/my-namespace/my-topic -p 3

# 更新命名空间为其指定集群
./bin/pulsar-admin namespaces set-clusters my-tenant/my-namespace --clusters pulsar-cluster

设置 maven 依赖

<dependency>
           <groupId>org.apache.pulsar</groupId>
           <artifactId>pulsar-client</artifactId>

<!-- 指定你的版本 -->

<version>2.2.0</version>
</dependency>

创建生产者

public class PulsarProducerDemo {    private static String localClusterUrl = "pulsar://localhost:6650";    public static void main(String[] args) {        try {            Producer<byte[]> producer = getProducer();            String msg = "hello world pulsar!";            Long start = System.currentTimeMillis();            MessageId msgId = producer.send(msg.getBytes());            System.out.println("spend=" + (System.currentTimeMillis() - start) + ";send a message msgId = " + msgId.toString());
       } catch (Exception e) {            System.err.println(e);
       }
   }    public static Producer<byte[]> getProducer() throws Exception {        PulsarClient client;
       client = PulsarClient.builder().serviceUrl(localClusterUrl).build();        Producer<byte[]> producer = client.newProducer().topic("persistent://my-tenant/my-namespace/my-topic").producerName("producerName").create();        return producer;
   }
}

创建消费者

public class PulsarConsumerDemo {    private static String localClusterUrl = "pulsar://localhost:6650";    public static void main(String[] args) {        try {            //将订阅消费者指定的主题和订阅
           Consumer<byte[]> consumer = getClient().newConsumer()
                   .topic("persistent://my-tenant/my-namespace/my-topic")
                   .subscriptionName("my-subscription")
                   .subscribe();            while (true) {                Message msg = consumer.receive();                System.out.printf("consumer-Message received: %s. \n", new String(msg.getData()));                // 确认消息，以便broker删除消息
               consumer.acknowledge(msg);
           }
       } catch (Exception e) {            System.out.println(e);
       }
   }    public static PulsarClient getClient() throws Exception {        PulsarClient client;
       client = PulsarClient.builder().serviceUrl(localClusterUrl).build();        return client;
   }
}

更多关于 Pulsar 的干货和动态，请关注公众号 ApachePulsar。

点击“阅读原文”，查看原作者 blog。