Pulsar该如何使用

# Pulsar该如何使用

## 目录
1. [什么是Pulsar](#什么是pulsar)
2. [Pulsar的核心概念](#pulsar的核心概念)
3. [安装与部署](#安装与部署)
4. [基础使用指南](#基础使用指南)
5. [高级功能](#高级功能)
6. [最佳实践](#最佳实践)
7. [常见问题解答](#常见问题解答)

---

## 什么是Pulsar

Apache Pulsar 是一个开源的分布式发布-订阅消息系统，由雅虎开发并捐赠给Apache基金会。它结合了高性能消息队列、流处理和轻量级函数计算的能力，具有以下特点：

- **多租户架构**：支持隔离的命名空间和细粒度权限控制
- **低延迟高吞吐**：单集群可支持百万级TPS
- **持久化存储**：通过分层存储（Tiered Storage）实现无限回溯
- **地理复制**：原生支持跨地域数据同步

典型应用场景包括：
- 金融交易实时处理
- IoT设备数据采集
- 微服务间异步通信
- 流式数据分析管道

---

## Pulsar的核心概念

### 1. 基础组件
| 组件          | 说明                                                                 |
|---------------|----------------------------------------------------------------------|
| **Broker**    | 无状态服务层，处理消息路由和协议转换                                 |
| **BookKeeper**| 持久化存储层，提供高可靠的日志存储                                   |
| **ZooKeeper** | 元数据存储和集群协调服务                                             |

### 2. 逻辑概念
- **Tenant**：多租户隔离单元
- **Namespace**：租户内的逻辑分区
- **Topic**：消息的发布/订阅通道（支持持久化/非持久化）
- **Subscription**：消费组的订阅关系（包含Exclusive/Failover/Shared/Key_Shared四种模式）

### 3. 消息模型
```python
# 消息结构示例
{
  "data": "二进制消息体",
  "properties": {"k1":"v1"},  # 用户自定义属性
  "sequence_id": 12345,       # 序列号
  "publish_time": 1630000000  # 发布时间戳
}

安装与部署

单机模式（开发环境）

# 使用官方Docker镜像
docker run -it -p 6650:6650 -p 8080:8080 \
  apachepulsar/pulsar:latest \
  bin/pulsar standalone

集群部署（生产环境）

1. 准备3台以上服务器
2. 安装ZooKeeper集群（建议3/5节点）
3. 部署BookKeeper存储节点
4. 配置Broker服务
5. 验证集群状态：

bin/pulsar-admin clusters list

基础使用指南

1. 管理Topic

# 创建Topic
bin/pulsar-admin topics create persistent://tenant/ns/topic1

# 查看Topic列表
bin/pulsar-admin topics list tenant/ns

2. 生产消息（Python示例）

import pulsar

client = pulsar.Client('pulsar://localhost:6650')
producer = client.create_producer('my-topic')

for i in range(10):
    producer.send(f'Message-{i}'.encode('utf-8'))

client.close()

3. 消费消息

consumer = client.subscribe('my-topic', 'my-sub')

while True:
    msg = consumer.receive()
    print(f"Received: {msg.data().decode()}")
    consumer.acknowledge(msg)

高级功能

1. 消息路由策略

• RoundRobin：轮询分区
• SinglePartition：固定分区
• CustomPartition：自定义路由

// Java自定义路由示例
MessageRouter customRouter = (msg, metadata) -> {
    String key = msg.getKey();
    return key.hashCode() % metadata.numPartitions();
};

2. 事务支持

# 开启事务
txn = client.new_transaction()
    .with_txn_timeout(5, TimeUnit.MINUTES)
    .build()

# 事务内操作
producer.send(msg, txn=txn)
consumer.acknowledge(msg, txn=txn)

# 提交/回滚
txn.commit()
# txn.abort()

3. 函数计算

# function-config.yaml
name: word-counter
inputs: ["persistent://public/default/input"]
output: "persistent://public/default/output"
runtime: python

最佳实践

1. 性能调优

• 批量发送：producer.batch_enable = True
• 内存优化：调整managedLedgerCacheSizeMB
• IO优化：使用SSD存储Journal日志

2. 监控指标

关键监控项： - 生产者/消费者延迟 - 积压消息数（backlog） - 存储写入延迟

# 获取统计信息
bin/pulsar-admin topics stats-internal persistent://tenant/ns/topic1

3. 安全配置

# broker.conf
authenticationEnabled=true
authorizationEnabled=true
superUserRoles=admin

常见问题解答

Q1: 消息堆积如何处理？

• 增加消费者数量
• 调整消费者接收窗口（receiverQueueSize）
• 启用消息TTL自动过期

Q2: 如何保证消息顺序？

• 使用单分区Topic
• 采用Key-Shared订阅模式
• 相同Key的消息路由到同一分区

Q3: Pulsar与Kafka如何选型？

提示：本文基于Pulsar 2.10版本编写，具体实现可能随版本变化。建议参考官方文档获取最新信息。 “`

这篇文章包含了约1600字，采用Markdown格式编写，覆盖了Pulsar的基础使用到高级特性。如需扩展特定部分或添加更多示例代码，可以进一步补充细节内容。