MacOS 单机搭建 Pulsar 集群

作者：高天赐

编辑：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。

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 配置参数解释

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

对 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。

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

对 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

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

对 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，并为命名空间指定集群。

设置 maven 依赖

创建生产者

public class PulsarProducerDemo {    private static String localClusterUrl = "pulsar://localhost:6650";    public static void main(String[] args) {        try {            Producer 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 getProducer() throws Exception {        PulsarClient client;

client = PulsarClient.builder().serviceUrl(localClusterUrl).build();        Producer 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 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。