0 缘起
Pulsar是一个支持多租户的、高性能的消息中间件。
2018年11月中旬开始初步在线上生产环境使用。
pulsar官网，部署教程文档还不是很详细，网络上的教程基本都是官网的翻译版，对于一个没有丰富经验的开发者还是会踩一些坑，本文记录一下相对详细测试集群搭建。
1 准备资源
一台主机(本文以macOS为例)
java8运行环境
2 搭建集群的组成
zk集群（3个ZooKeeper节点组成）
bookie集群（3个BookKeeper节点组成）
broker集群（3个Pulsar节点组成）
3 zk集群-搭建
ZooKeeper 版本 3.4.12。
使用一台机器，在该台机器上运行多个ZooKeeper 服务进程，搭建zk集群。
(3.1)下载zk，解压。
Zookeeper官网下载地址（zookeeper-3.4.12.tar.gz）
https://archive.apache.org/dist/zookeeper/zookeeper-3.4.12/
(3.2)将解压好的zookeeper-3.4.12复制到新建文件zookeepers目录下，重名为server1。用server1复制出server2和server3，目录结构如图所示。
(3.3)对server1进行配置(先对zk的一个节点进行配置)。
(3.3.1)在server1目录下，新建data和dataLog两个文件夹。目录结构如下。
(3.3.2)将conf目录下的zoo_sample.cfg文件重命名为zoo.cfg。如图所示。
(3.3.3)修改 zoo.cfg 文件内容,主要修改以下5个配置参数，以及添加集群节点信息。
*修改如下5个参数
dataDir
dataLogDir
clientPort
admin.enableServer
admin.serverPort
*添加如下集群节点信息
server.1=127.0.0.1:2888:3888
server.2=127.0.0.1:2889:3889
server.3=127.0.0.1:2890:3890
我的server1的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/gaotianci/Softwares/zookeepers/server1/data
dataLogDir=/Users/gaotianci/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 section 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。这里设置为5表示最长容忍时间为10秒。
syncLimit：用来配置Leader与Follower之间发送消息、请求和应答时间最长能忍受多少个tickTime。这里设置为2表示最长容忍时间为4秒。
dataDir：数据文件目录。
dataLogDir：日志文件目录。
clientPort：监听client连接的端口号。
server.{myid}={ip}:{leader服务器交换信息的端口}:{当leader服务器挂了后, 选举leader的端口}
maxClientCnxns：对于一个客户端的连接数限制，默认是60。
admin.enableServer：是否启用zk管理后台。
admin.serverPort：管理后台端口号。
在一台服务器上，部署多个实例，需要指定不同的端口号。
[1]clientPort: 3个zk节点中分别配置为:2181,2182,2183
[2]admin.enableServer:3个zk节点中都配置为 true
[3]admin.serverPort:3个zk节点中分别配置为:9181,9182,9183
[4]集群节点信息:3个zk节点中都配置为:
server.1=127.0.0.1:2888:3888
server.2=127.0.0.1:2889:3889
server.3=127.0.0.1:2890:3890
(3.3.4)在server1/data/目录下创建名字为myid文件。
向myid文件中写入内容 1
对myid文件的解释:
每一个zk节点的的myid内容都不能一样，它是不同节点的唯一标识。
myid文件内容,3个zk节点中分别为1,2,3
(3.3.5)server1配置完毕，对server2,server3做类似配置。具体不再赘述。
(3.3.6)启动zk集群,在终端分别用命令启动zk节点。
./bin/zkServer.sh start zk启动命令(zkServer.sh start)
./bin/zkServer.sh status zk状态查看命令(zkServer.sh status)
./bin/zkServer.sh stop zk关闭命令(zkServer.sh stop)
(3.3.7)./bin/zkServer.sh status 命令查看节点状态。
(3.3.8)用zk客户端命令(./bin/zkCli.sh -timeout 5000 -server 127.0.0.1:2181)连接zk节点,在这里连接server1。
(3.3.9)查看(ls /)zk集群当前内容。
(3.3.10)向zk集群写入元数据内容。用如下命令写入，后面解释写入内容。
create /--cluster pulsar-cluster
create /--zookeeper 127.0.0.1:2181
create /--configuration-store 127.0.0.1:2181
create /--web-service-url http://pulsar.cluster.com:8080
create /--web-service-url-tls https://pulsar.cluster.com:8443
create /--broker-service-url pulsar://pulsar.cluster.com:6650
create /--broker-service-url-tls pulsar+ssl://pulsar.cluster.com:6651
zk集群写入元数据内容解释:
--cluster
集群名称
--zookeeper
ZooKeeper集群连接参数，仅需要包含集群中的一个节点即可
--configuration-store
Pulsar实例的配置存储集群（ZooKeeper），和-zookeeper参数一样只需要包含集群中的一个节点即可
--web-service-url
集群Web服务的URL+端口，URL必须是一个i标准的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，不建议修改。
4 bookie集群-搭建
bookie 是bookkeeper 的别称。版本 4.7.2。
使用一台机器，在该台机器上运行多个bookie 服务进程，搭建bookie集群。
(4.1)下载zk，解压。
bookkeeper官网下载地址（bookkeeper-server-4.7.2-bin.tar.gz）
https://bookkeeper.apache.org/releases/
(4.2)将解压好的bookkeeper-server-4.7.2复制到新建文件bookkeepers目录下，重名为bookie1。用bookie1复制出bookie2和bookie3，目录结构如图所示。
(4.3)对bookie1进行配置(先对bookkeeper的一个节点进行配置)。
(4.3.1)修改 bk_server.conf 文件内容,主要修改以下3个端口号,以防在一台主机上造成端口号冲突，以及添加zk集群节点信息。
*修改如下3个参数
bookiePort
httpServerPort
storageserver.grpc.port
*添加如下zk集群节点信息
zkServers=localhost:2181,localhost:2182,localhost:2183
在一台服务器上，部署多个实例，需要指定不同的端口号。
[1]bookiePort: 3个bookie节点中分别配置为:3181,3182,3183
[2]httpServerPort: 3个bookie节点中都配置为:8050,8060,8070
[3]storageserver.grpc.port: 3个bookie节点中都配置为: 4181,4182,4183
[4]zk集群节点信息:3个bookie中都配置为:
zkServers=localhost:2181,localhost:2182,localhost:2183
(4.3.2)bookie1配置完毕，对bookie2,bookie3做类似配置。具体不再赘述。
(4.3.3)执行初始化集群元数据命令(在一个bookie上执行即可),命令如下。
./bin/bookkeeper shell metaformat
(4.3.4)在终端，3个bookie下，分别执行命令启动bookie,命令如下。
./bin/bookkeeper bookie
(4.3.5)检查bookie启动状态,集群启动成功会有"Bookie sanity test succeeded"日志输出。在一个bookie 实例下执行如下命令。
./bin/bookkeeper shell bookiesanity
5 broker集群-搭建
broker 是pulsar实例别称。版本 2.2.0。
使用一台机器，在该台机器上运行多个broker 服务进程，搭建broker集群。
(5.1)下载pulsar，解压。
pulsar官网下载地址（apache-pulsar-2.2.0-bin-tar.gz）
http://pulsar.apache.org/zh-CN/download/
(5.2)将解压好的apache-pulsar-2.2.0复制到新建文件brokers目录下，重名为broker1。用broker1复制出broker2和broker3，目录结构如图所示。
(5.3)对broker1进行配置(先对broker的一个节点进行配置)。
(5.3.1)修改 broker.conf 文件内容,主要修改以下4个端口号,以防在一台主机上造成端口号冲突，以及添加zk集群节点信息。
*修改如下4个参数
brokerServicePort
brokerServicePortTls
webServicePort
webServicePortTls
*添加如下zk集群节点信息
zookeeperServers=localhost:2181,localhost:2182,localhost:2183
configurationStoreServers=localhost:2181,localhost:2182,localhost:2183
在一台服务器上，部署多个实例，需要指定不同的端口号。
[1]brokerServicePort: 3个broker节点中分别配置为:6650,6660,6670
[2]brokerServicePortTls: 3个broker节点中分别配置为:6651,6661,6671
[3]webServicePort: 3个broker节点中分别配置为:8080,8081,8082
[4]webServicePortTls: 3个broker节点中分别配置为:8443,8444,8445
[5]zk集群节点信息:3个broker中都配置为:
zookeeperServers=localhost:2181,localhost:2182,localhost:2183
configurationStoreServers=localhost:2181,localhost:2182,localhost:2183
(5.3.2)broker1配置完毕，对broker2,broker3做类似配置。具体不再赘述。
(5.3.3)在终端，3个broker下，分别执行命令启动broker,启动成功后会有日志“PulsarService started”输出,命令如下。
./bin/pulsar broker
6 pulsar集群启动完毕，现用命令创建集群名，租户名，命名空间，topic并行给出测试demo。
(6.1)依次创建集群，租户，命名空间，分区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
(6.2)生产者，消费者测试demo。
(6.2.1)maven依赖
<dependency>
<groupId>org.apache.pulsar</groupId>
<artifactId>pulsar-client</artifactId>
<version>2.2.0</version>
</dependency>
(6.2.2)生产者
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;
}
}
(6.2.3)消费者
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;
}
}
