1
// BookKeeper 客户端写入
2
BookKeeper bk = new BookKeeper(zkConnectString);
3

4
// 创建 Ledger
5
LedgerHandle ledger = bk.createLedger(
6
    3,    // ensembleSize: Bookie 数量
7
    3,    // writeQuorumSize: 写入副本数
8
    2,    // ackQuorumSize: 确认副本数
9
    BookKeeper.DigestType.CRC32,
10
    "password".getBytes()
11
);
12

13
// 写入 Entry
14
long entryId = ledger.addEntry("Hello Pulsar".getBytes());
15

16
// 关闭 Ledger
17
ledger.close();

1
persistent://tenant/namespace/topic
2
   |           |       |       |
3
   |           |       |       └── Topic 名称
4
   |           |       └── 命名空间（隔离单元）
5
   |           └── 租户（多租户隔离）
6
   └── 持久化类型（persistent / non-persistent）

1
# 创建租户
2
pulsar-admin tenants create acme-corp \
3
    --admin-roles admin \
4
    --allowed-clusters us-east,us-west
5

6
# 创建命名空间
7
pulsar-admin namespaces create acme-corp/production \
8
    --clusters us-east,us-west \
9
    --retention 7d
10

11
# 创建 Topic
12
pulsar-admin topics create persistent://acme-corp/production/orders \
13
    --partitions 8

1
// 创建不同订阅模式的消费者
2
Consumer<byte[]> exclusiveConsumer = client.newConsumer()
3
    .topic("persistent://acme-corp/production/orders")
4
    .subscriptionName("order-processor")
5
    .subscriptionType(SubscriptionType.Exclusive)
6
    .subscribe();
7

8
Consumer<byte[]> sharedConsumer = client.newConsumer()
9
    .topic("persistent://acme-corp/production/orders")
10
    .subscriptionName("order-processor")
11
    .subscriptionType(SubscriptionType.Shared)
12
    .subscribe();
13

14
Consumer<byte[]> keySharedConsumer = client.newConsumer()
15
    .topic("persistent://acme-corp/production/orders")
16
    .subscriptionName("order-processor")
17
    .subscriptionType(SubscriptionType.Key_Shared)
18
    .subscribe();

1
# 配置 Geo 复制
2
# 1. 创建全局命名空间
3
pulsar-admin namespaces create acme-corp/global-orders \
4
    --clusters us-east,us-west,eu-central
5

6
# 2. 设置复制策略
7
pulsar-admin namespaces set-replication-clusters acme-corp/global-orders \
8
    --clusters us-east,us-west,eu-central
9

10
# 3. 查看复制状态
11
pulsar-admin namespaces get-replication-clusters acme-corp/global-orders

1
// Producer：发送消息到本地集群，自动复制到远程
2
Producer<byte[]> producer = client.newProducer()
3
    .topic("persistent://acme-corp/global-orders/orders")
4
    .create();
5

6
// 消息会自动复制到所有配置的集群
7
producer.send("Order data".getBytes());
8

9
// Consumer：从最近的集群消费
10
Consumer<byte[]> consumer = client.newConsumer()
11
    .topic("persistent://acme-corp/global-orders/orders")
12
    .subscriptionName("order-processor")
13
    .subscribe();
14

15
// 消费者会收到来自所有集群的消息（去重后）
16
Message<byte[]> msg = consumer.receive();

1
# 创建租户并设置配额
2
pulsar-admin tenants create acme-corp \
3
    --admin-roles admin@acme \
4
    --allowed-clusters us-east
5

6
# 设置命名空间配额
7
pulsar-admin namespaces set-quota acme-corp/production \
8
    --max-producers-per-topic 50 \
9
    --max-consumers-per-topic 100 \
10
    --max-consumers-per-subscription 50
11

12
# 设置消息保留策略
13
pulsar-admin namespaces set-retention acme-corp/production \
14
    --size 100G \
15
    --time 7d
16

17
# 设置限流
18
pulsar-admin namespaces set-rate-limit acme-corp/production \
19
    --publish-rate 1000 \
20
    --subscribe-rate 1000

1
# Broker 隔离：指定命名空间只能使用特定 Broker
2
pulsar-admin namespaces set-bundle-range acme-corp/production \
3
    --broker-affinity-group premium-brokers
4

5
# Bookie 隔离：指定命名空间只能使用特定 Bookie
6
pulsar-admin namespaces set-bookie-affinity-group acme-corp/production \
7
    --bookie-affinity-group premium-bookies \
8
    --primary-bookie-group us-east-premium

1
// Pulsar Function：订单金额过滤
2
public class OrderFilterFunction implements Function<Order, Order> {
3
    @Override
4
    public Order process(Order input, Context context) {
5
        if (input.getAmount() > 1000) {
6
            context.publish("high-value-orders", input);
7
            context.incrCounter("high-value-count", 1);
8
        }
9
        return input;
10
    }
11
}

1
# 部署 Function
2
pulsar-admin functions create \
3
    --name order-filter \
4
    --inputs persistent://acme-corp/production/orders \
5
    --output persistent://acme-corp/production/filtered-orders \
6
    --classname com.example.OrderFilterFunction \
7
    --jar /path/to/function.jar
8

9
# 查看状态
10
pulsar-admin functions status --name order-filter

1
# 配置分层存储（S3）
2
pulsar-admin namespaces set-offload-policies acme-corp/production \
3
    --offload-threshold 100MB \
4
    --offload-deletion-lag 2d \
5
    --s3-endpoint s3.amazonaws.com \
6
    --s3-bucket pulsar-offload \
7
    --s3-region us-east-1