温馨提示：本文是「Pulsar 傻瓜手册」，手把手为零经验小白排坑避雷。只要按照指引操作，成功率 100% 喔！
本文是「Pulsar 隔离策略系列」的第 2 篇，该系列的第 1 篇博客 — 「深度解析如何在 Pulsar 中实现隔离」— 重点介绍了 Pulsar 隔离策略以及如何通过以下 3 种方式实现资源隔离：
多个 Pulsar 集群
共享 BookKeeper 集群
单个 Pulsar 集群
本文将详细讲解第 1 种方式，即如何在「多个 Pulsar 集群」的环境中，玩转 Pulsar 隔离策略，并为以下操作提供详细步骤：
准备环境
本文以在 macOS（版本 11.2.3， 内存 8G）上操作为示例。
如下图所示，本文部署 2 个 Pulsar 集群，每个集群提供以下服务：
1 个 ZooKeeper 节点
1 个 bookie 节点
1 个 broker 节点
软件要求
Java 8
环境详情
下文即将部署 2 个 Pulsar 集群，组件详情如下表所示。
部署准备
下载 Pulsar 并解压。本文以安装 Pulsar 2.7.0 为例。
在本地任意位置，按照以下目录结构，创建相应的空文件夹。
|-separate-clusters |-configuration-store |-zk1 |-cluster1 |-zk1 |-bk1 |-broker1 |-cluster2 |-zk1 |-bk1 |-broker1
复制代码
复制解压后 Pulsar 文件夹中的内容至上一步创建的每个文件夹中。
启动 configuration store。
Configuration store 为 Pulsar 实例提供跨集群配置管理和任务协调。Pulsar 集群 1 和 Pulsar 集群 2 共享 configuration store。
cd configuration-store/zk1bin/pulsar-daemon start configuration-store
复制代码
部署 Pulsar 集群 1
启动 local ZooKeeper。为每个 Pulsar 集群部署 1 个 local ZooKeeper，它负责为该集群管理配置和协调任务。
cd cluster1/zk1bin/pulsar-daemon start zookeeper
复制代码
初始化元数据。设置好 configuration store 和 local ZooKeeper 后，你需要向 ZooKeeper 写入元数据。
cd cluster1/zk1bin/pulsar initialize-cluster-metadata \ --cluster cluster1 \ --zookeeper localhost:2181 \ --configuration-store localhost:2184 \ --web-service-url http://localhost:8080/ \ --web-service-url-tls https://localhost:8443/ \ --broker-service-url pulsar://localhost:6650/ \ --broker-service-url-tls pulsar+ssl://localhost:6651/
复制代码
部署 BookKeeper。BookKeeper 为 Pulsar 提供消息持久化存储。每个 Pulsar broker 拥有 bookie。BookKeeper 集群和 Pulsar 集群共享 local ZooKeeper。
1).配置 bookie。
更改 cluster1/bk1/conf/bookkeeper.conf 文件中以下配置项的值。
allowLoopback=trueprometheusStatsHttpPort=8002httpServerPort=8002
复制代码
2).启动 bookie。
cd cluster1/bk1bin/pulsar-daemon start bookie
复制代码
验证是否成功启动 bookie。
bin/bookkeeper shell bookiesanity
复制代码
输出
Bookie sanity test succeeded
复制代码
部署 broker。
1)配置 broker。
更改 cluster1/broker1/conf/broker.conf 文件中以下配置项的值。
zookeeperServers=127.0.0.1:2181configurationStoreServers=127.0.0.1:2184clusterName=cluster1managedLedgerDefaultEnsembleSize=1managedLedgerDefaultWriteQuorum=1managedLedgerDefaultAckQuorum=1
复制代码
2)启动 broker。
cd cluster1/broker1bin/pulsar-daemon start broker
复制代码
部署 Pulsar 集群 2
部署 local ZooKeeper。
1)配置 local ZooKeeper。
clientPort=2186admin.serverPort=9992
复制代码
OPTS="-Dstats_server_port=8011"
复制代码
2)启动 local ZooKeeper。
cd cluster2/zk1bin/pulsar-daemon start zookeeper
复制代码
2.初始化元数据。
bin/pulsar initialize-cluster-metadata \ --cluster cluster2 \ --zookeeper localhost:2186 \ --configuration-store localhost:2184 \ --web-service-url http://localhost:8081/ \ --web-service-url-tls https://localhost:8444/ \ --broker-service-url pulsar://localhost:6660/ \ --broker-service-url-tls pulsar+ssl://localhost:6661/
复制代码
部署 BookKeeper。
1)配置 bookie。
更改 cluster2/bk1/conf/bookkeeper.conf 文件中以下配置项的值。
bookiePort=3182zkServers=localhost:2186allowLoopback=trueprometheusStatsHttpPort=8003httpServerPort=8003
复制代码
2)启动 bookie。
cd cluster2/bk1bin/pulsar-daemon start bookie
复制代码
验证是否成功启动 bookie。
bin/bookkeeper shell bookiesanity
复制代码
输出
Bookie sanity test succeeded
复制代码
5.部署 broker。
1)配置 broker。
clusterName=cluster2zookeeperServers=127.0.0.1:2186configurationStoreServers=127.0.0.1:2184brokerServicePort=6660webServicePort=8081managedLedgerDefaultEnsembleSize=1managedLedgerDefaultWriteQuorum=1managedLedgerDefaultAckQuorum=1
复制代码
•更改 cluster2/broker1/conf/client.conf 文件中以下配置项的值。
webServiceUrl=http://localhost:8081/brokerServiceUrl=pulsar://localhost:6660/
复制代码
2)启动 broker。
cd cluster2/broker1bin/pulsar-daemon start broker
复制代码
验证数据隔离
本章验证 2 个 Pulsar 集群中的数据是否隔离。
创建 namespace1，并将 namespace1 分配给 cluster1。
提示：namespace 的命名规则是 /。更多关于 namespace 的信息，参阅这里。
cd cluster1/broker1bin/pulsar-admin namespaces create -c cluster1 public/namespace1
复制代码
验证结果
bin/pulsar-admin namespaces list public
复制代码
输出
"public/default""public/namespace1"
复制代码
2.设置 namespace1 的消息保留策略。
注意：如果不设置消息保留策略且 topic 未被订阅，一段时间后，topic 的数据会被自动清理。
bin/pulsar-admin namespaces set-retention -s 100M -t 3d public/namespace1
复制代码
3.在 namespace1 创建 topic1，并使用写入 1000 条数据。
提示：pulsar-client 是发送和消费数据的命令行工具。更多关于 Pulsar 命令行工具的信息，参阅这里。
bin/pulsar-client produce -m 'hello c1 to c2' -n 1000 public/namespace1/topic109:56:34.504 [main] INFO org.apache.pulsar.client.cli.PulsarClientTool - 1000 messages successfully produced
复制代码
验证结果
bin/pulsar-admin --admin-url http://localhost:8080 topics stats-internal public/namespace1/topic1
复制代码
输出
entriesAddedCounter 显示增加了 1000 条数据。
{ "entriesAddedC{ "entriesAddedCounter" : 1000, "numberOfEntries" : 1000, "totalSize" : 65616, "currentLedgerEntries" : 1000, "currentLedgerSize" : 65616, "lastLedgerCreatedTimestamp" : "2021-04-22T10:24:00.582+08:00", "waitingCursorsCount" : 0, "pendingAddEntriesCount" : 0, "lastConfirmedEntry" : "4:999", "state" : "LedgerOpened", "ledgers" : [ { "ledgerId" : 4, "entries" : 0, "size" : 0, "offloaded" : false } ], "cursors" : { }, "compactedLedger" : { "ledgerId" : -1, "entries" : -1, "size" : -1, "offloaded" : false }}ounter" : 1000, "numberOfEntries" : 1000, "totalSize" : 65616, "currentLedgerEntries" : 1000, "currentLedgerSize" : 65616, "lastLedgerCreatedTimestamp" : "2021-04-22T10:24:00.582+08:00", "waitingCursorsCount" : 0, "pendingAddEntriesCount" : 0, "lastConfirmedEntry" : "4:999", "state" : "LedgerOpened", "ledgers" : [ { "ledgerId" : 4, "entries" : 0, "size" : 0, "offloaded" : false } ], "cursors" : { }, "compactedLedger" : { "ledgerId" : -1, "entries" : -1, "size" : -1, "offloaded" : false }}
复制代码
4.通过 cluster2（localhost:8081）查看 public/namespace1/topic1 的数据。
bin/pulsar-admin --admin-url http://localhost:8081 topics stats-internal public/namespace1/topic1
复制代码
输出
查看失败。打印信息显示 public/namespace1 仅分配至 cluster1，未分配至 cluster 2。此时验证了数据已隔离。
Namespace missing local cluster name in clusters list: local_cluster=cluster2 ns=public/namespace1 clusters=[cluster1]Reason: Namespace missing local cluster name in clusters list: local_cluster=cluster2 ns=public/namespace1 clusters=[cluster1]
复制代码
5.在 cluster2 中，向 public/namespace1/topic1 写入数据。
cd cluster2/broker1bin/pulsar-client produce -m 'hello c1 to c2' -n 1000 public/namespace1/topic1
复制代码
输出
结果显示写入消息数量为 0，操作失败，因为 namespace1 仅分配至 cluster1，未分配至 cluster 2。此时验证了数据已隔离。
12:09:50.005 [main] INFO org.apache.pulsar.client.cli.PulsarClientTool - 0 messages successfully produced
复制代码
迁移数据
以下步骤接着前文「验证数据隔离」继续操作。
确认数据已隔离后，本章讲述如何同步（使用跨地域复制功能）并迁移集群数据。
分配 namespace1 至 cluster2，即添加 cluster2 至 namesapce1 的 cluster 列表。
该步骤启用跨地域复制功能，同步 cluster1 和 cluster2 的数据。
bin/pulsar-admin namespaces set-clusters --clusters cluster1,cluster2 public/namespace1
复制代码
验证结果
bin/pulsar-admin namespaces get-clusters public/namespace1
复制代码
输出
2.查看 cluster2 是否有 topic1。
bin/pulsar-admin --admin-url http://localhost:8081 topics stats-internal public/namespace1/topic1
复制代码
输出
结果显示 cluster2 的 topic1 有 1000 条数据，说明 cluster1 的 topic1 数据已成功复制至 cluster2。
{ "entriesAddedCounter" : 1000, "numberOfEntries" : 1000, "totalSize" : 75616, "currentLedgerEntries" : 1000, "currentLedgerSize" : 75616, "lastLedgerCreatedTimestamp" : "2021-04-23T12:02:52.929+08:00", "waitingCursorsCount" : 1, "pendingAddEntriesCount" : 0, "lastConfirmedEntry" : "1:999", "state" : "LedgerOpened", "ledgers" : [ { "ledgerId" : 1, "entries" : 0, "size" : 0, "offloaded" : false } ], "cursors" : { "pulsar.repl.cluster1" : { "markDeletePosition" : "1:999", "readPosition" : "1:1000", "waitingReadOp" : true, "pendingReadOps" : 0, "messagesConsumedCounter" : 1000, "cursorLedger" : 2, "cursorLedgerLastEntry" : 2, "individuallyDeletedMessages" : "[]", "lastLedgerSwitchTimestamp" : "2021-04-23T12:02:53.248+08:00", "state" : "Open", "numberOfEntriesSinceFirstNotAckedMessage" : 1, "totalNonContiguousDeletedMessagesRange" : 0, "properties" : { } } }, "compactedLedger" : { "ledgerId" : -1, "entries" : -1, "size" : -1, "offloaded" : false }}
复制代码
3.迁移 cluster1 的 producer 和 consumer 至 cluster2。
PulsarClient pulsarClient1 = PulsarClient.builder().serviceUrl("pulsar://localhost:6650").build();// migrate the client to cluster2 pulsar://localhost:6660PulsarClient pulsarClient2 = PulsarClient.builder().serviceUrl("pulsar://localhost:6660").build();
复制代码
4.从 namespace1 的 cluster 列表中移除 cluster1 。
bin/pulsar-admin namespaces set-clusters --clusters cluster2 public/namespace1
复制代码
5.检查 cluster1 的 topic1 的数据。
cd cluster1/broker1bin/pulsar-admin --admin-url http://localhost:8080 topics stats-internal public/namespace1/topic1
复制代码
输出
结果显示数据为空，说明数据已从 cluster1 的 topic1 中成功移除。
{ "entriesAddedCounter" : 0, "numberOfEntries" : 0, "totalSize" : 0, "currentLedgerEntries" : 0, "currentLedgerSize" : 0, "lastLedgerCreatedTimestamp" : "2021-04-23T15:20:08.1+08:00", "waitingCursorsCount" : 1, "pendingAddEntriesCount" : 0, "lastConfirmedEntry" : "3:-1", "state" : "LedgerOpened", "ledgers" : [ { "ledgerId" : 3, "entries" : 0, "size" : 0, "offloaded" : false } ], "cursors" : { "pulsar.repl.cluster2" : { "markDeletePosition" : "3:-1", "readPosition" : "3:0", "waitingReadOp" : true, "pendingReadOps" : 0, "messagesConsumedCounter" : 0, "cursorLedger" : 4, "cursorLedgerLastEntry" : 0, "individuallyDeletedMessages" : "[]", "lastLedgerSwitchTimestamp" : "2021-04-23T15:20:08.122+08:00", "state" : "Open", "numberOfEntriesSinceFirstNotAckedMessage" : 1, "totalNonContiguousDeletedMessagesRange" : 0, "properties" : { } } }, "compactedLedger" : { "ledgerId" : -1, "entries" : -1, "size" : -1, "offloaded" : false }}
复制代码
至此，我们已将 cluster1 的 topic1 数据成功复制至 cluster2，之后移除了 cluster1 的 topic1 数据。
扩缩容节点
本章讲述如何扩缩容 broker 和 bookie 节点。
Broker
增加 broker 节点
本示例在 cluster1/broker1 创建 2 个 partitioned topic，再增加 2 个 broker 节点。之后卸载 partitioned topic 数据，并查看数据在 3 个 broker 之间的分配情况。
查看 cluster1 的 broker 信息。
cd/cluster1/broker1bin/pulsar-admin brokers list cluster1
复制代码
输出
结果说明当前 cluster1 只有 broker1。
在 cluster1/broker1 上创建 2 个 partitioned topic。
为 partitioned-topic1 创建 6 个分区，为 partitioned-topic2 创建 7 个分区。
bin/pulsar-admin topics create-partitioned-topic -p 6 public/namespace1/partitioned-topic1bin/pulsar-admin topics create-partitioned-topic -p 7 public/namespace1/partitioned-topic2
复制代码
查看结果。
bin/pulsar-admin topics partitioned-lookup public/namespace1/partitioned-topic1
复制代码
输出
结果显示，partitioned-topic1 所有数据属于 broker1。
"persistent://public/namespace1/partitioned-topic1-partition-0 pulsar://192.168.0.105:6650""persistent://public/namespace1/partitioned-topic1-partition-1 pulsar://192.168.0.105:6650""persistent://public/namespace1/partitioned-topic1-partition-2 pulsar://192.168.0.105:6650""persistent://public/namespace1/partitioned-topic1-partition-3 pulsar://192.168.0.105:6650""persistent://public/namespace1/partitioned-topic1-partition-4 pulsar://192.168.0.105:6650""persistent://public/namespace1/partitioned-topic1-partition-5 pulsar://192.168.0.105:6650"
复制代码
输入
bin/pulsar-admin topics partitioned-lookup public/namespace1/partitioned-topic2
复制代码
输出
结果显示，partitioned-topic2 所有数据属于 broker1。
"persistent://public/namespace1/partitioned-topic2-partition-0 pulsar://192.168.0.105:6650""persistent://public/namespace1/partitioned-topic2-partition-1 pulsar://192.168.0.105:6650""persistent://public/namespace1/partitioned-topic2-partition-2 pulsar://192.168.0.105:6650""persistent://public/namespace1/partitioned-topic2-partition-3 pulsar://192.168.0.105:6650""persistent://public/namespace1/partitioned-topic2-partition-4 pulsar://192.168.0.105:6650""persistent://public/namespace1/partitioned-topic2-partition-5 pulsar://192.168.0.105:6650""persistent://public/namespace1/partitioned-topic2-partition-6 pulsar://192.168.0.105:6650"
复制代码
新增 2 个 broker 节点：broker2 和 broker3。
1)部署准备。在 cluster1 文件夹中创建 broker2 和 broker3 文件夹，复制解压后 Pulsar 文件夹中的内容至 broker2 和 broker3 文件夹。
|-separate-clusters |-configuration-store |-zk1 |-cluster1 |-zk1 |-bk1 |-broker1 |-broker2 |-broker3 |-cluster2 |-zk1 |-bk1 |-broker1
复制代码
2)部署 broker。
a). 配置 broker。
b). 启动 broker。
c). 查看 cluster1 中已启动的 broker。
bin/pulsar-admin brokers list cluster1
复制代码
输出
"192.168.0.105:8080" // broker1"192.168.0.105:8082" // broker2"192.168.0.105:8083" // broker3
复制代码
卸载 namespace 1 中 partitioned-topic1 的数据。
bin/pulsar-admin namespaces unload public/namespace1
复制代码
验证结果。
1). 查看 partitioned-topic1 的数据分配情况。
bin/pulsar-admin topics partitioned-lookup public/namespace1/partitioned-topic1
复制代码
输出
结果显示，partitioned-topic1 的数据平均分配在 broker1、broker2 和 broker3 上。
"persistent://public/namespace1/partitioned-topic1-partition-0 pulsar://192.168.0.105:6650""persistent://public/namespace1/partitioned-topic1-partition-1 pulsar://192.168.0.105:6653""persistent://public/namespace1/partitioned-topic1-partition-2 pulsar://192.168.0.105:6652""persistent://public/namespace1/partitioned-topic1-partition-3 pulsar://192.168.0.105:6653""persistent://public/namespace1/partitioned-topic1-partition-4 pulsar://192.168.0.105:6650""persistent://public/namespace1/partitioned-topic1-partition-5 pulsar://192.168.0.105:6653"
复制代码
2). 查看 partitioned-topic2 的数据分配情况。
bin/pulsar-admin topics partitioned-lookup public/namespace1/partitioned-topic2
复制代码
结果显示，partitioned-topic2 的数据平均分配在 broker1、broker2 和 broker3 上。
输出
"persistent://public/namespace1/partitioned-topic2-partition-0 pulsar://192.168.0.105:6653""persistent://public/namespace1/partitioned-topic2-partition-1 pulsar://192.168.0.105:6650""persistent://public/namespace1/partitioned-topic2-partition-2 pulsar://192.168.0.105:6653""persistent://public/namespace1/partitioned-topic2-partition-3 pulsar://192.168.0.105:6652""persistent://public/namespace1/partitioned-topic2-partition-4 pulsar://192.168.0.105:6653""persistent://public/namespace1/partitioned-topic2-partition-5 pulsar://192.168.0.105:6650""persistent://public/namespace1/partitioned-topic2-partition-6 pulsar://192.168.0.105:6653"
复制代码
减少 broker 节点
以下步骤接着前文「如何增加 broker 节点」继续操作。
本示例在 cluster1 中减少 1 个 broker 节点，并查看 partitioned topic 数据在其余 2 个 broker 之间的分配情况。
减少 1 个 broker 节点，即停止 broker3。
cd/cluster1/broker3bin/pulsar-daemon stop broker
复制代码
验证结果。
bin/pulsar-admin brokers list cluster1
复制代码
输出
结果显示，当前 cluster1 仅启动了 broker1 和 broker2。
"192.168.0.105:8080" // broker1"192.168.0.105:8082" // broker2
复制代码
查看 partitioned-topic1 数据的分配情况。
bin/pulsar-admin topics partitioned-lookup public/namespace1/partitioned-topic1
复制代码
输出
结果显示，partitioned-topic1 数据平均分配至 broker1 和 broker2，即原属于 broker3 的数据已被重新平均分配至 broker1 和 broker2。
"persistent://public/namespace1/partitioned-topic1-partition-0 pulsar://192.168.0.105:6650""persistent://public/namespace1/partitioned-topic1-partition-1 pulsar://192.168.0.105:6650""persistent://public/namespace1/partitioned-topic1-partition-2 pulsar://192.168.0.105:6652""persistent://public/namespace1/partitioned-topic1-partition-3 pulsar://192.168.0.105:6652""persistent://public/namespace1/partitioned-topic1-partition-4 pulsar://192.168.0.105:6650""persistent://public/namespace1/partitioned-topic1-partition-5 pulsar://192.168.0.105:6650"
复制代码
同理，partitioned-topic2 的数据也被平均分配至 broker1 和 broker2。
bin/pulsar-admin topics partitioned-lookup public/namespace1/partitioned-topic2
复制代码
输出
"persistent://public/namespace1/partitioned-topic2-partition-0 pulsar://192.168.0.105:6650""persistent://public/namespace1/partitioned-topic2-partition-1 pulsar://192.168.0.105:6650""persistent://public/namespace1/partitioned-topic2-partition-2 pulsar://192.168.0.105:6652""persistent://public/namespace1/partitioned-topic2-partition-3 pulsar://192.168.0.105:6652""persistent://public/namespace1/partitioned-topic2-partition-4 pulsar://192.168.0.105:6650""persistent://public/namespace1/partitioned-topic2-partition-5 pulsar://192.168.0.105:6650""persistent://public/namespace1/partitioned-topic2-partition-6 pulsar://192.168.0.105:6652"
复制代码
Bookie
增加 bookie 节点
本示例在 cluster1/bookkeeper 1 已有 bookie1，增加 2 个 bookie 节点后，向 topic1 写入数据，并查看数据是否保存了多个副本。
查看 cluster1 的 bookie 信息。
cd cluster1/bk1bin/bookkeeper shell listbookies -rw -h
复制代码
输出
结果说明当前 cluster1 只有 bookie1。
12:31:34.933 [main] INFO org.apache.bookkeeper.tools.cli.commands.bookies.ListBookiesCommand - ReadWrite Bookies :12:31:34.946 [main] INFO org.apache.bookkeeper.tools.cli.commands.bookies.ListBookiesCommand - BookieID:192.168.0.105:3181, IP:192.168.0.105, Port:3181, Hostname:192.168.0.105
复制代码
2.允许 3 个 bookie 节点服务。
更改 cluster1/broker1/conf/broker.conf 文件中以下配置项的值。
managedLedgerDefaultEnsembleSize=3 // 指定 bookie 节点服务的数量managedLedgerDefaultWriteQuorum=3 // 指定数据副本写入的数量managedLedgerDefaultAckQuorum=2 // 指定数据成功写入几个副本后，数据才算写入成功
复制代码
3.重启 broker1，使配置生效。
cd cluster1/broker1bin/pulsar-daemon stop brokerbin/pulsar-daemon start broker
复制代码
4.设置 public/default 的消息保留策略。
注意：如果不设置消息保留策略且 topic 未被订阅，一段时间后，topic 的数据会被自动清理。
cd cluster1/broker1bin/pulsar-admin namespaces set-retention -s 100M -t 3d public/default
复制代码
5.在 public/default 创建 topic1，并写入 100 条数据。
bin/pulsar-client produce -m 'hello' -n 100 topic1
复制代码
输出
结果显示 bookie 节点数量不足导致数据写入失败。
···12:40:38.886 [pulsar-client-io-1-1] WARN org.apache.pulsar.client.impl.ClientCnx - [id: 0x56f92aff, L:/192.168.0.105:53069 - R:/192.168.0.105:6650] Received error from server: org.apache.bookkeeper.mledger.ManagedLedgerException: Not enough non-faulty bookies available...12:40:38.886 [main] ERROR org.apache.pulsar.client.cli.PulsarClientTool - Error while producing messages…12:40:38.890 [main] INFO org.apache.pulsar.client.cli.PulsarClientTool - 0 messages successfully produced
复制代码
新增 2 个 bookie 节点：bookie2 和 bookie3。
1).部署准备。
在 cluster1 中新增 bk2 和 bk3 文件夹，复制解压后 Pulsar 文件夹中的内容至 bk2 和 bk3 文件夹。
|-separate-clusters |-configuration-store |-zk1 |-cluster1 |-zk1 |-bk1 |-bk2 |-bk3 |-broker1 |-cluster2 |-zk1 |-bk1 |-broker1
复制代码
2).部署 bookie。
a). 配置 bookie。
b). 启动 bookie。
c). 检查 cluster1 中已启动的 bookie。
bin/bookkeeper shell listbookies -rw -h
复制代码
输出
结果显示 cluster 1 中已启动 3 个 bookie：
bookie1：192.168.0.105:3181
bookie2：192.168.0.105:3183
bookie3：192.168.0.105:3184
12:12:47.574 [main] INFO org.apache.bookkeeper.tools.cli.commands.bookies.ListBookiesCommand - BookieID:192.168.0.105:3183, IP:192.168.0.105, Port:3183, Hostname:192.168.0.105 12:12:47.575 [main] INFO org.apache.bookkeeper.tools.cli.commands.bookies.ListBookiesCommand - BookieID:192.168.0.105:3184, IP:192.168.0.105, Port:3184, Hostname:192.168.0.10512:12:47.576 [main] INFO org.apache.bookkeeper.tools.cli.commands.bookies.ListBookiesCommand - BookieID:192.168.0.105:3181, IP:192.168.0.105, Port:3181, Hostname:192.168.0.105
复制代码
7.设置 public/default 的消息保留策略。
注意：如果不设置消息保留策略且 topic 未被订阅，一段时间后，topic 的数据会被自动清理。
cd cluster1/broker1bin/pulsar-admin namespaces set-retention -s 100M -t 3d public/default
复制代码
在 public/default 创建 topic1，并写入 100 条数据。
bin/pulsar-client produce -m 'hello' -n 100 topic1
复制代码
输出
结果显示数据写入成功。
...12:17:40.222 [main] INFO org.apache.pulsar.client.cli.PulsarClientTool - 100 messages successfully produced
复制代码
查看 topic1 的信息。
bin/pulsar-admin topics stats-internal topic1
复制代码
输出
结果显示 ledgerId 5 保存了 topic1 的数据。
{ "entriesAddedCounter" : 100, "numberOfEntries" : 100, "totalSize" : 5500, "currentLedgerEntries" : 100, "currentLedgerSize" : 5500, "lastLedgerCreatedTimestamp" : "2021-05-11T12:17:38.881+08:00", "waitingCursorsCount" : 0, "pendingAddEntriesCount" : 0, "lastConfirmedEntry" : "5:99", "state" : "LedgerOpened", "ledgers" : [ { "ledgerId" : 5, "entries" : 0, "size" : 0, "offloaded" : false } ], "cursors" : { }, "compactedLedger" : { "ledgerId" : -1, "entries" : -1, "size" : -1, "offloaded" : false }}
复制代码
查看 ledgerid 5 存储在哪些 bookie 节点上。
bin/bookkeeper shell ledgermetadata -ledgerid 5
复制代码
输出
结果显示正如前文所配置， ledgerid 5 存储在 bookie1（3181）、bookie2（3183） 和 bookie3（3184）上。
...12:23:17.705 [main] INFO org.apache.bookkeeper.tools.cli.commands.client.LedgerMetaDataCommand - ledgerID: 512:23:17.714 [main] INFO org.apache.bookkeeper.tools.cli.commands.client.LedgerMetaDataCommand - LedgerMetadata{formatVersion=3, ensembleSize=3, writeQuorumSize=3, ackQuorumSize=2, state=OPEN, digestType=CRC32C, password=base64:, ensembles={0=[192.168.0.105:3184, 192.168.0.105:3181, 192.168.0.105:3183]}, customMetadata={component=base64:bWFuYWdlZC1sZWRnZXI=, pulsar/managed-ledger=base64:cHVibGljL2RlZmF1bHQvcGVyc2lzdGVudC90b3BpYzE=, application=base64:cHVsc2Fy}}...
复制代码
减少 bookie 节点
以下步骤接着前文「如何增加 bookie 节点」继续操作。
本示例在 cluster1 中减少 2 个 bookie 节点，再向 topic2 写入数据，并查看数据保存在哪些节点。
允许 1 个 bookie 节点服务。
更改 cluster1/broker1/conf/broker.conf 文件中以下配置项的值。
managedLedgerDefaultEnsembleSize=1 // 指定 bookie 节点服务的数量managedLedgerDefaultWriteQuorum=1 // 指定数据副本写入的数量managedLedgerDefaultAckQuorum=1 // 指定数据成功写入几个副本后，数据才算写入成功
复制代码
2.重启 broker1，使配置生效。
cd cluster1/broker1bin/pulsar-daemon stop brokerbin/pulsar-daemon start broker
复制代码
查看 cluster1 的 bookie 信息。
cd cluster1/bk1bin/bookkeeper shell listbookies -rw -h
复制代码
输出
结果说明当前 cluster1 已启动了 bookie1（3181）、bookie2（3183） 和 bookie3（3184）。
...15:47:41.370 [main] INFO org.apache.bookkeeper.tools.cli.commands.bookies.ListBookiesCommand - ReadWrite Bookies :15:47:41.382 [main] INFO org.apache.bookkeeper.tools.cli.commands.bookies.ListBookiesCommand - BookieID:192.168.0.105:3183, IP:192.168.0.105, Port:3183, Hostname:192.168.0.10515:47:41.383 [main] INFO org.apache.bookkeeper.tools.cli.commands.bookies.ListBookiesCommand - BookieID:192.168.0.105:3184, IP:192.168.0.105, Port:3184, Hostname:192.168.0.10515:47:41.384 [main] INFO org.apache.bookkeeper.tools.cli.commands.bookies.ListBookiesCommand - BookieID:192.168.0.105:3181, IP:192.168.0.105, Port:3181, Hostname:192.168.0.105…
复制代码
减少 2 个 bookie 节点，即停止 bookie2 和 bookie3。
提示：更多关于如何减少 bookie 的信息，参阅这里。
cd cluster1/bk2bin/bookkeeper shell listunderreplicatedbin/pulsar-daemon stop bookiebin/bookkeeper shell decommissionbookie
复制代码
cd cluster1/bk3bin/bookkeeper shell listunderreplicatedbin/pulsar-daemon stop bookiebin/bookkeeper shell decommissionbookie
复制代码
5.查看 cluster1 的 bookie 信息。
cd cluster1/bk1bin/bookkeeper shell listbookies -rw -hcd cluster1/bk1bin/bookkeeper shell listbookies -rw -h
复制代码
输出
结果说明当前 cluster1 仅启动了 bookie1（3181）。
…16:05:28.690 [main] INFO org.apache.bookkeeper.tools.cli.commands.bookies.ListBookiesCommand - ReadWrite Bookies :16:05:28.700 [main] INFO org.apache.bookkeeper.tools.cli.commands.bookies.ListBookiesCommand - BookieID:192.168.0.105:3181, IP:192.168.0.105, Port:3181, Hostname:192.168.0.105...
复制代码
设置 public/default 的消息保留策略。
注意：如果不设置消息保留策略且没有订阅，一段时间后，数据会被自动清理。
cd cluster1/broker1bin/pulsar-admin namespaces set-retention -s 100M -t 3d public/default
复制代码
在 public/default 创建 topic2，并写入 100 条数据。
bin/pulsar-client produce -m 'hello' -n 100 topic2
复制代码
输出
结果显示数据写入成功。
…16:06:59.448 [main] INFO org.apache.pulsar.client.cli.PulsarClientTool - 100 messages successfully produced
复制代码
查看 topic2 的信息。
bin/pulsar-admin topics stats-internal topic2
复制代码
输出
结果显示 ledgerId 7 保存了 topic 2 的数据。
{ "entriesAddedCounter" : 100, "numberOfEntries" : 100, "totalSize" : 5400, "currentLedgerEntries" : 100, "currentLedgerSize" : 5400, "lastLedgerCreatedTimestamp" : "2021-05-11T16:06:59.058+08:00", "waitingCursorsCount" : 0, "pendingAddEntriesCount" : 0, "lastConfirmedEntry" : "7:99", "state" : "LedgerOpened", "ledgers" : [ { "ledgerId" : 7, "entries" : 0, "size" : 0, "offloaded" : false } ], "cursors" : { }, "compactedLedger" : { "ledgerId" : -1, "entries" : -1, "size" : -1, "offloaded" : false }}
复制代码
查看 ledgerid 7 存储在哪些 bookie 节点上。
bin/bookkeeper shell ledgermetadata -ledgerid 7
复制代码
输出
结果显示 ledgerid 7 存储在 bookie1（3181）上。
...16:11:28.843 [main] INFO org.apache.bookkeeper.tools.cli.commands.client.LedgerMetaDataCommand - ledgerID: 716:11:28.846 [main] INFO org.apache.bookkeeper.tools.cli.commands.client.LedgerMetaDataCommand - LedgerMetadata{formatVersion=3, ensembleSize=1, writeQuorumSize=1, ackQuorumSize=1, state=OPEN, digestType=CRC32C, password=base64:, ensembles={0=[192.168.0.105:3181]}, customMetadata={component=base64:bWFuYWdlZC1sZWRnZXI=, pulsar/managed-ledger=base64:cHVibGljL2RlZmF1bHQvcGVyc2lzdGVudC90b3BpYzM=, application=base64:cHVsc2Fy}}…
复制代码
总结
本文是「Pulsar 隔离策略系列」的第 2 篇，讲解了如何在「多个 Pulsar 集群」的环境中验证数据隔离、同步集群数据和扩缩容节点。
本系列的下一篇博客将详细分析如何在「共享 BookKeeper 集群」的环境中玩转 Pulsar 隔离策略，同样也会是「Pulsar 傻瓜手册」喔！敬请期待！
更多信息
如果你对「Pulsar 隔离策略系列」感兴趣，欢迎查阅更多相关资源：
点击链接，获取 Apache Pulsar 硬核干货资料！
评论