在实际的生产环境里，消息中间件选什么用，往往不是看哪个名气大，而是看你的业务场景到底需要什么。电商订单流和日志采集这两类场景，虽然都是消息驱动，但对中间件的要求截然不同。订单流要的是"一条都不能丢"，日志采集要的是"海量吞吐且别拖垮系统"。今天要聊的就是 Pulsar 怎么同时应对这两种需求，以及在真实项目里到底该怎么选型、怎么落地。
Pulsar 是一个由 Apache 基金会孵化的分布式消息系统，最早出自 StreamNative，后来捐赠给了 Apache。它和 Kafka 一样能发消息，但内部架构完全不一样——Pulsar 把存储和计算分开了，计算层是 Broker，存储层是 BookKeeper。这种设计让它在弹性扩缩容上比 Kafka 灵活得多。
下面我们就从订单流和日志采集两个场景出发，看看怎么在同一个 Pulsar 集群上跑起来。
一、订单流场景：消息可靠性的命门所在
1.1 订单流到底要什么
电商订单流的核心诉求其实就两个字：可靠。用户下单那一刻，消息必须送达下游——库存扣减、支付回调、物流预约、积分发放……任何一个环节丢了消息，都可能引发客诉甚至资金损失。
所以订单流场景的设计原则是：宁可慢一点，也不能丢一条。
1.2 Topic 命名与分区策略
在 Pulsar 里，Topic 是消息流动的基本单位。订单流场景推荐用分区 Topic（Partitioned Topic），把不同状态的消息分到不同分区里，既保证吞吐，也方便按业务维度做消费。
下面是一个 Topic 命名规范示例：
partition://ecommerce/order-created/10
partition://ecommerce/order-paid/10
partition://ecommerce/order-shipped/10
partition://ecommerce/order-completed/10
partition://ecommerce/order-cancelled/10
这里 ecommerce 是 namespace，order-created 等是 topic 名，10 是分区数。分区数要根据单机处理能力来定，一般建议是消费者数量的整数倍。
1.3 生产者代码示例
技术栈：Java 17 + Apache Pulsar Java Client 3.1 + Maven
/**
* 电商订单消息生产者
* 技术栈：Java 17 + Apache Pulsar Client 3.1
* 职责：将订单事件发送到 Pulsar 的订单 Topic
*/
package com.demo.pulsar.producer;
import org.apache.pulsar.client.api.*;
import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.UUID;
import java.util.concurrent.TimeUnit;
import java.util.logging.Level;
import java.util.logging.Logger;
public class OrderProducer {
private static final Logger LOG = Logger.getLogger(OrderProducer.class.getName());
// Pulsar 服务地址
private static final String PULSAR_SERVICE_URL = "pulsar://localhost:6650";
// 订单 Topic 前缀
private static final String ORDER_TOPIC_PREFIX = "partition://ecommerce/order";
private PulsarClient client;
private Producer<String> producer;
/**
* 初始化生产者
*/
public void init() throws PulsarClientException {
// 创建 Pulsar 客户端，设置发送超时 10 秒
client = PulsarClient.builder()
.serviceUrl(PULSAR_SERVICE_URL)
.sendTimeout(10, TimeUnit.SECONDS)
.operationTimeout(15, TimeUnit.SECONDS)
.build();
// 创建生产者，使用 Key 进行消息路由，确保同一订单走同一分区
producer = client.newProducer(StandardCharsets.UTF_8)
.topic(ORDER_TOPIC_PREFIX + "-created")
.producerName("order-producer-" + UUID.randomUUID().toString().substring(0, 8))
.hashingScheme(HashingScheme.xxHash) // 使用 xxHash 做消息键哈希
.messageRoutingMode(MessageRoutingMode.KeyHash) // 按 Key 哈希路由到同一分区
.blockIfQueueFull(true) // 队列满时阻塞，绝不丢弃消息
.maxPendingMessages(50000)
.maxPendingMessagesAcrossPartitions(100000)
.compressionType(CompressionType.LZ4) // LZ4 压缩减少网络带宽
.build();
LOG.info("订单生产者初始化完成");
}
/**
* 发送订单创建消息（异步方式，兼顾可靠性与性能）
*
* @param orderJson 订单 JSON 字符串
* @param orderId 订单 ID，用作消息 Key
*/
public void sendOrderCreated(String orderJson, String orderId) {
// 构建消息：设置 Key 保证同订单同分区，设置 Properties 方便过滤
producer.newMessage()
.key(orderId) // 订单 ID 作为 Key，Pulsar 会用 xxHash 计算目标分区
.value(orderJson)
.property("event_type", "ORDER_CREATED")
.property("order_id", orderId)
.property("timestamp", String.valueOf(System.currentTimeMillis()))
.sendAsync()
.whenComplete((messageId, throwable) -> {
if (throwable != null) {
// 发送失败记录错误日志，需要重试或写入死信队列
LOG.log(Level.SEVERE,
String.format("订单消息发送失败，orderId=%s, error=%s",
orderId, throwable.getMessage()));
} else {
LOG.info(String.format("订单消息发送成功，orderId=%s, msgId=%s",
orderId, messageId));
}
});
}
/**
* 发送订单支付消息（同步方式，关键节点要求确认）
* 支付是资金操作，必须同步确认落盘后才继续
*/
public void sendOrderPaid(String orderJson, String orderId) throws Exception {
producer.newMessage()
.key(orderId)
.value(orderJson)
.property("event_type", "ORDER_PAID")
.send(); // 同步发送，等待服务端确认后再返回
}
/**
* 关闭生产者资源
*/
public void close() {
try {
if (producer != null) {
producer.flush(); // 先刷未发送的消息
producer.close();
}
if (client != null) {
client.close();
}
LOG.info("订单生产者已关闭");
} catch (PulsarClientException e) {
LOG.log(Level.SEVERE, "关闭生产者时出错", e);
}
}
/**
* 测试入口
*/
public static void main(String[] args) throws Exception {
OrderProducer producer = new OrderProducer();
producer.init();
// 模拟发送 10 条订单消息
String sampleOrder = "{\"orderId\":\"ORD-20240101-001\","
+ "\"userId\":\"user-1001\","
+ "\"amount\":299.00,"
+ "\"items\":[{\"skuId\":\"SKU-001\",\"quantity\":2}],"
+ "\"status\":\"CREATED\"}";
for (int i = 0; i < 10; i++) {
String orderId = "ORD-20240101-" + String.format("%03d", i + 1);
producer.sendOrderCreated(sampleOrder, orderId);
Thread.sleep(100); // 模拟生产速率
}
// 等待异步消息发送完成
Thread.sleep(5000);
producer.close();
}
}
上面这个生产者有几个关键设计点值得注意。MessageRoutingMode.KeyHash 确保同一个订单的所有消息走同一个分区，这样消费端可以按订单维度做顺序处理。blockIfQueueFull(true) 保证在队列满的时候阻塞而不是丢弃消息，这是订单场景的核心保障。压缩选择 LZ4 是因为它在压缩率和解压速度之间取得了很好的平衡。
1.4 消费者代码示例
订单消费者需要做到：消费确认、失败重试、死信队列兜底。
/**
* 电商订单消息消费者
* 技术栈：Java 17 + Apache Pulsar Client 3.1
* 职责：消费订单创建消息，执行后续业务逻辑
*/
package com.demo.pulsar.consumer;
import org.apache.pulsar.client.api.*;
import com.google.gson.Gson;
import com.google.gson.annotations.SerializedName;
import java.util.concurrent.TimeUnit;
import java.util.logging.Level;
import java.util.logging.Logger;
public class OrderConsumer {
private static final Logger LOG = Logger.getLogger(OrderConsumer.class.getName());
private static final String PULSAR_SERVICE_URL = "pulsar://localhost:6650";
private static final String ORDER_TOPIC = "partition://ecommerce/order-created";
private static final String SUBSCRIPTION_NAME = "order-processor-sub";
private static final String DEAD_LETTER_TOPIC = "ecommerce/order-dead-letter";
private PulsarClient client;
private Consumer<String> consumer;
private Gson gson = new Gson();
/**
* 订单事件实体
*/
public static class OrderEvent {
@SerializedName("orderId")
public String orderId;
@SerializedName("userId")
public String userId;
@SerializedName("amount")
public double amount;
@SerializedName("status")
public String status;
@SerializedName("timestamp")
public long timestamp;
@Override
public String toString() {
return String.format("OrderEvent{orderId='%s', userId='%s', amount=%.2f, status='%s'}",
orderId, userId, amount, status);
}
}
/**
* 初始化消费者
*/
public void init() throws PulsarClientException {
client = PulsarClient.builder()
.serviceUrl(PULSAR_SERVICE_URL)
.operationTimeout(30, TimeUnit.SECONDS)
.build();
consumer = client.newConsumer(StandardCharsets.UTF_8)
.topic(ORDER_TOPIC)
.subscriptionName(SUBSCRIPTION_NAME)
.subscriptionType(SubscriptionType.Failover) // Failover：同一分组只有一个实例消费
.subscriptionInitialPosition(SubscriptionInitialPosition.Earliest) // 从最早的消息开始
.receiverQueueSize(1000) // 本地接收队列大小
.deadLetterPolicy(DeadLetterPolicy.builder()
.maxRedeliverCount(5) // 最多重试 5 次
.deadLetterTopic(DEAD_LETTER_TOPIC) // 重试失败后进入死信队列
.build())
.subscribe();
LOG.info("订单消费者初始化完成，订阅模式=Failover");
}
/**
* 开始消费消息
*/
public void startConsuming() {
LOG.info("开始消费订单消息...");
while (true) {
try {
// 使用 MessageListener 异步监听，避免轮询
MessageListener<String> listener = (consumerInst, message) -> {
processOrder(message);
try {
consumerInst.acknowledge(message); // 处理成功后确认
} catch (PulsarClientException e) {
LOG.log(Level.SEVERE, "消息确认失败", e);
}
};
consumer.listen(listener);
Thread.sleep(3600000); // 保持线程存活
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
break;
} catch (PulsarClientException e) {
LOG.log(Level.SEVERE, "消费异常，3秒后重试", e);
try { Thread.sleep(3000); } catch (InterruptedException ex) {
Thread.currentThread().interrupt(); break;
}
}
}
}
/**
* 处理订单消息的核心业务逻辑
*/
private void processOrder(Message<String> message) {
try {
String payload = message.getValue();
OrderEvent event = gson.fromJson(payload, OrderEvent.class);
LOG.info("收到订单消息: " + event);
// 1. 校验订单数据完整性
if (event.orderId == null || event.orderId.isEmpty()) {
throw new IllegalArgumentException("订单ID不能为空");
}
// 2. 幂等性检查：避免重复处理
if (isDuplicate(event.orderId)) {
LOG.warning("订单已处理过，跳过: " + event.orderId);
return;
}
// 3. 执行业务逻辑（库存扣减、积分发放等）
deductInventory(event);
sendNotification(event);
updateOrderStatus(event, "PROCESSING");
LOG.info("订单处理完成: " + event.orderId);
} catch (IllegalArgumentException e) {
// 数据校验失败，不重试
LOG.log(Level.SEVERE, "订单数据校验失败", e);
} catch (Exception e) {
// 业务处理失败，不确认消息，Pulsar 会自动重发
LOG.log(Level.SEVERE, "订单处理失败，等待重试", e);
throw new RuntimeException("订单处理失败", e);
}
}
/**
* 检查订单是否重复处理（实际项目中查 Redis 或数据库）
*/
private boolean isDuplicate(String orderId) {
return false; // 示例中返回 false
}
private void deductInventory(OrderEvent event) {
LOG.info("扣减库存: " + event.orderId);
}
private void sendNotification(OrderEvent event) {
LOG.info("发送订单通知: " + event.userId);
}
private void updateOrderStatus(OrderEvent event, String newStatus) {
LOG.info("更新订单状态: " + event.orderId + " -> " + newStatus);
}
/**
* 关闭消费者资源
*/
public void close() {
try {
if (consumer != null) consumer.close();
if (client != null) client.close();
LOG.info("订单消费者已关闭");
} catch (PulsarClientException e) {
LOG.log(Level.SEVERE, "关闭消费者时出错", e);
}
}
}
关于死信队列，Pulsar 提供了内置的 Dead Letter Queue 支持。上面示例里设置了 maxRedeliverCount(5)，意思是同一条消息最多重试 5 次，如果 5 次都失败就自动把消息投到死信 Topic 里。这样的好处是：消费者不会因为一条毒消息无限重试而阻塞整个队列。生产环境建议单独做一个死信消费者，专门处理死信队列里的消息，人工介入或触发告警。
二、日志采集场景：海量数据下的吞吐与成本平衡
2.1 日志采集的特殊需求
日志采集和订单流最大的区别在于：日志是"允许少量丢失"的，但吞吐量极高。一次系统重启可能产生几十 GB 的日志，如果每条都同步确认，系统性能会被严重拖慢。
日志场景的核心矛盾是：既要处理海量数据，又不能因为日志太多把业务系统搞挂。
2.2 日志 Topic 设计策略
日志 Topic 的设计建议按照"系统 + 模块"来划分，同时使用批量发送和批量消费来最大化吞吐。
persistent://logging/java-app/system-log # 系统级日志（INFO/WARN/ERROR）
persistent://logging/java-app/access-log # 访问日志（API请求）
persistent://logging/java-app/audit-log # 审计日志（用户操作）
分区数要开大一些，日志场景通常按机器数量来定。比如你有 50 台应用服务器，那分区数设成 50 或者 100 比较合适。
2.3 日志生产者代码示例
日志生产者的核心策略是异步批量发送，减少网络往返次数。
/**
* 应用日志采集生产者
* 技术栈：Java 17 + Apache Pulsar Client 3.1
* 职责：将应用日志异步批量发送到 Pulsar
*/
package com.demo.pulsar.logging;
import org.apache.pulsar.client.api.*;
import com.google.gson.Gson;
import com.google.gson.GsonBuilder;
import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.concurrent.TimeUnit;
import java.util.logging.Level;
import java.util.logging.Logger;
public class LogProducer {
private static final Logger LOG = Logger.getLogger(LogProducer.class.getName());
private static final String PULSAR_SERVICE_URL = "pulsar://localhost:6650";
private static final String SYSTEM_LOG_TOPIC = "persistent://logging/java-app/system-log";
private static final String ACCESS_LOG_TOPIC = "persistent://logging/java-app/access-log";
private PulsarClient client;
private Producer<String> systemLogProducer;
private Producer<String> accessLogProducer;
private Gson gson;
/**
* 初始化日志生产者
*/
public void init() throws PulsarClientException {
gson = new GsonBuilder().setPrettyPrinting(false).create();
client = PulsarClient.builder()
.serviceUrl(PULSAR_SERVICE_URL)
.operationTimeout(60, TimeUnit.SECONDS)
.ioThreads(4) // 增加 IO 线程数以提升并发能力
.build();
// 系统日志生产者：批量发送优化
systemLogProducer = client.newProducer(StandardCharsets.UTF_8)
.topic(SYSTEM_LOG_TOPIC)
.producerName("system-log-producer")
.batchingMaxPublishDelay(10, TimeUnit.MILLISECONDS) // 每 10ms 批量发送一次
.batchingMaxMessages(1000) // 或攒够 1000 条再发，取先到条件
.batchingMaxBytes(5 * 1024 * 1024) // 或攒够 5MB 再发
.batchingMaxBucketCount(10) // 批量桶数，控制并发发送
.compressionType(CompressionType.LZ4)
.blockIfQueueFull(true)
.maxPendingMessages(100000) // 日志场景队列可以设大
.maxPendingMessagesAcrossPartitions(200000)
.enableBatching(true) // 开启批量
.sendTimeout(30, TimeUnit.SECONDS)
.build();
// 访问日志生产者
accessLogProducer = client.newProducer(StandardCharsets.UTF_8)
.topic(ACCESS_LOG_TOPIC)
.producerName("access-log-producer")
.batchingMaxPublishDelay(20, TimeUnit.MILLISECONDS) // 访问日志可以攒更久
.batchingMaxMessages(2000)
.compressionType(CompressionType.LZ4)
.blockIfQueueFull(true)
.maxPendingMessages(100000)
.build();
LOG.info("日志生产者初始化完成，系统日志批量延迟=10ms");
}
/**
* 发送系统日志
*
* @param level 日志级别（INFO/WARN/ERROR）
* @param logger 日志来源类名
* @param message 日志消息
*/
public void sendSystemLog(String level, String logger, String message) {
LogEntry entry = new LogEntry();
entry.timestamp = LocalDateTime.now().format(DateTimeFormatter.ISO_LOCAL_DATE_TIME);
entry.level = level;
entry.logger = logger;
entry.message = message;
entry.thread = Thread.currentThread().getName();
entry.appName = "java-web-app";
entry.env = "production";
entry.instanceId = getInstanceId();
systemLogProducer.newMessage()
.key(entry.instanceId + "-" + entry.timestamp)
.value(gson.toJson(entry))
.property("log_level", level)
.property("app_name", entry.appName)
.sendAsync()
.whenComplete((messageId, throwable) -> {
if (throwable != null) {
LOG.warning("日志发送失败: " + throwable.getMessage());
}
});
}
/**
* 发送访问日志
*/
public void sendAccessLog(String requestId, String method, String uri,
int status, long duration) {
AccessLogEntry entry = new AccessLogEntry();
entry.timestamp = LocalDateTime.now().format(DateTimeFormatter.ISO_LOCAL_DATE_TIME);
entry.requestId = requestId;
entry.method = method;
entry.uri = uri;
entry.status = status;
entry.duration = duration;
entry.appName = "java-web-app";
entry.env = "production";
entry.instanceId = getInstanceId();
// 按状态码路由：正常请求和错误请求走不同分区，方便后续统计
String routingKey = (status >= 400) ? "error-" + status : "success-" + status;
accessLogProducer.newMessage()
.key(routingKey + "-" + requestId)
.value(gson.toJson(entry))
.property("http_status", String.valueOf(status))
.property("duration", String.valueOf(duration))
.sendAsync();
}
/**
* 日志条目实体
*/
public static class LogEntry {
public String timestamp;
public String level;
public String logger;
public String message;
public String thread;
public String appName;
public String env;
public String instanceId;
}
/**
* 访问日志条目实体
*/
public static class AccessLogEntry {
public String timestamp;
public String requestId;
public String method;
public String uri;
public int status;
public long duration;
public String appName;
public String env;
public String instanceId;
}
/**
* 获取实例 ID
*/
private static String getInstanceId() {
String id = System.getenv("INSTANCE_ID");
return (id != null) ? id : "unknown-instance";
}
/**
* 优雅关闭
*/
public void close() {
try {
systemLogProducer.flush();
accessLogProducer.flush();
systemLogProducer.close();
accessLogProducer.close();
client.close();
LOG.info("日志生产者已关闭");
} catch (PulsarClientException e) {
LOG.log(Level.SEVERE, "关闭日志生产者时出错", e);
}
}
/**
* 使用示例：模拟批量发送日志
*/
public static void main(String[] args) throws Exception {
LogProducer logProducer = new LogProducer();
logProducer.init();
// 模拟发送 1000 条系统日志
for (int i = 0; i < 1000; i++) {
String level = (i % 10 == 0) ? "ERROR" :
(i % 5 == 0) ? "WARN" : "INFO";
logProducer.sendSystemLog(level, "com.demo.OrderService",
"处理订单#" + i + "，耗时" + (long)(Math.random() * 100) + "ms");
}
// 模拟发送访问日志
for (int i = 0; i < 500; i++) {
int status = (i % 10 == 0) ? 500 :
(i % 5 == 0) ? 404 : 200;
logProducer.sendAccessLog(
"REQ-" + i,
(i % 2 == 0) ? "GET" : "POST",
"/api/orders/" + i,
status,
(long)(Math.random() * 500)
);
}
Thread.sleep(10000); // 等待批量消息发送完成
logProducer.close();
}
}
日志生产者这里有个关键的优化点：批量发送。batchingMaxPublishDelay(10, TimeUnit.MILLISECONDS) 的意思是，每攒 10 毫秒的消息就发一次批次。如果 10 毫秒内攒够了 1000 条或 5MB，就提前发。这样在高峰期可以把网络请求数降低几个数量级——原来是 1000 条消息发 1000 次请求，现在是发 1 次请求带 1000 条消息。
2.4 日志消费者代码示例
日志消费者通常用于：日志归档到存储、日志分析、告警触发。
/**
* 日志消息消费者（用于归档与分析）
* 技术栈：Java 17 + Apache Pulsar Client 3.1
* 职责：消费日志消息，写入存储或触发告警
*/
package com.demo.pulsar.logging;
import org.apache.pulsar.client.api.*;
import com.google.gson.Gson;
import java.util.concurrent.*;
import java.util.logging.Level;
import java.util.logging.Logger;
public class LogConsumer {
private static final Logger LOG = Logger.getLogger(LogConsumer.class.getName());
private static final String PULSAR_SERVICE_URL = "pulsar://localhost:6650";
private static final String SYSTEM_LOG_TOPIC = "persistent://logging/java-app/system-log";
private static final String SUBSCRIPTION_NAME = "log-archive-sub";
private PulsarClient client;
private Consumer<String> consumer;
private Gson gson = new Gson();
// 日志归档缓冲区
private final BlockingQueue<String> archiveBuffer = new LinkedBlockingQueue<>(10000);
// 异步归档线程池
private final ExecutorService archiveExecutor = Executors.newFixedThreadPool(3);
/**
* 初始化日志消费者
*/
public void init() throws PulsarClientException {
client = PulsarClient.builder()
.serviceUrl(PULSAR_SERVICE_URL)
.build();
consumer = client.newConsumer(StandardCharsets.UTF_8)
.topic(SYSTEM_LOG_TOPIC)
.subscriptionName(SUBSCRIPTION_NAME)
.subscriptionType(SubscriptionType.Failover)
.receiverQueueSize(5000) // 日志量大，接收队列设大
.ackTimeout(10, TimeUnit.SECONDS)
.negativeAckRedeliveryDelay(30, TimeUnit.SECONDS)
.subscribe();
LOG.info("日志消费者初始化完成");
}
/**
* 开始消费日志消息
*/
public void startConsuming() {
LOG.info("开始消费日志消息...");
consumer.listen((consumerInst, message) -> {
try {
String payload = message.getValue();
LogProducer.LogEntry logEntry = gson.fromJson(payload, LogProducer.LogEntry.class);
// 按日志级别处理
switch (logEntry.level) {
case "ERROR":
handleErrorMessage(logEntry);
break;
case "WARN":
handleWarningMessage(logEntry);
break;
default:
break;
}
archiveBuffer.offer(gson.toJson(logEntry));
consumerInst.acknowledge(message);
} catch (Exception e) {
LOG.log(Level.SEVERE, "处理日志消息失败", e);
}
});
}
/**
* 处理错误日志：触发告警
*/
private void handleErrorMessage(LogProducer.LogEntry entry) {
String alertMessage = String.format(
"[生产告警] 应用=%s, 级别=%s, 来源=%s, 消息=%s, 时间=%s",
entry.appName, entry.level, entry.logger, entry.message, entry.timestamp
);
// 异步发送告警，不阻塞消费线程
archiveExecutor.submit(() -> sendAlert(alertMessage));
}
private void handleWarningMessage(LogProducer.LogEntry entry) {
LOG.warning("WARN 日志: " + entry.message);
}
/**
* 发送告警通知
*/
private void sendAlert(String message) {
System.out.println("告警通知: " + message);
}
/**
* 启动日志归档后台线程
*/
public void startArchiveThread() {
archiveExecutor.submit(() -> {
LOG.info("日志归档线程已启动");
StringBuilder batchBuffer = new StringBuilder();
int count = 0;
while (true) {
try {
String entry = archiveBuffer.poll(5, TimeUnit.SECONDS);
if (entry != null) {
batchBuffer.append(entry).append("\n");
count++;
}
// 缓冲区超过 100 条或等待超时则刷入存储
if (count >= 100 || (entry == null && batchBuffer.length() > 0)) {
System.out.println("归档 " + count + " 条日志");
batchBuffer.setLength(0);
count = 0;
}
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
break;
}
}
});
}
/**
* 优雅关闭
*/
public void close() {
archiveExecutor.shutdown();
try { archiveExecutor.awaitTermination(30, TimeUnit.SECONDS); }
catch (InterruptedException e) { archiveExecutor.shutdownNow(); }
try {
consumer.close();
client.close();
} catch (PulsarClientException e) {
LOG.log(Level.SEVERE, "关闭日志消费者时出错", e);
}
LOG.info("日志消费者已关闭");
}
public static void main(String[] args) throws Exception {
LogConsumer logConsumer = new LogConsumer();
logConsumer.init();
logConsumer.startConsuming();
logConsumer.startArchiveThread();
Thread.currentThread().join();
}
}
日志消费者这里的设计思路是：用 MessageListener 做异步消费，避免单线程瓶颈。每条消息消费后先做分级处理——ERROR 日志立即触发告警，WARN 日志做统计后延迟告警，INFO 日志直接归档。归档部分用了独立的线程池和缓冲区，这样消费线程和归档线程互不干扰，日志量大时不会因为归档慢而拖慢消费。
三、技术优缺点分析
3.1 Pulsar 在订单流场景的优势
Pulsar 的分层架构（计算与存储分离）在订单流场景下有几个实实在在的好处。首先是弹性扩缩容方便——流量高峰时加 Broker 节点就能提升消费能力，不需要像 Kafka 那样重平衡分区。其次是消息不丢——Pulsar 的 ACK 机制是消息写入 BookKeeper 后才确认，比 Kafka 的 Leader 确认更可靠。第三是 Topic 数量可以非常大，Pulsar 单集群可以轻松支撑百万级 Topic，这在订单流场景里很实用，每个业务线甚至每个租户都可以有独立的 Topic。
3.2 Pulsar 在日志采集场景的优势
日志采集场景主要看重吞吐和成本。Pulsar 的批量发送能力、多层压缩（LZ4、ZSTD、SNAPPY）以及分层存储（热数据在 SSD、冷数据自动下沉到对象存储）让它在日志场景下非常经济。特别是分层存储功能——如果不用 Pulsar，日志冷数据要么自己搭存储，要么全放在 SSD 上成本很高。Pulsar 可以自动把 24 小时前的日志下沉到 S3，查询时再从对象存储拉取，成本能降好几倍。
3.3 Pulsar 的劣势与局限
Pulsar 不是万能的。它的劣势也很明显：
第一，架构复杂度高。Pulsar 由 ZooKeeper + BookKeeper + Broker 三个组件组成，部署运维的复杂度比 Kafka 高不少。如果你团队没有足够的运维能力，可能觉得 Kafka 更省心。
第二，小消息场景的延迟略高。因为 Pulsar 的元数据存储在 ZooKeeper 里，当 Topic 数量非常大时（比如几万个），ZooKeeper 可能成为瓶颈。不过对于电商订单流和日志采集这两种场景来说，Topic 数量通常在几千级别，不会碰到这个问题。
第三，Java Client 的批量发送在低延迟场景下表现一般。如果你需要微秒级延迟的消息传递，Pulsar 可能不是最佳选择。但对于订单流和日志采集来说，毫秒级延迟完全够用。
3.4 与 Kafka 的简单对比
- Kafka 的优势：生态成熟、社区活跃、运维资料多、KRaft 模式简化架构
- Pulsar 的优势：计算存储分离、弹性伸缩、多租户天然支持、分层存储、低延迟
对于同时需要订单流和日志采集的场景，Pulsar 的一个大优势是可以用同一个集群同时跑两种业务。Kafka 的话，通常建议把日志和交易消息分到不同的集群，因为日志的写模式是"大量小写入"，交易消息是"少量大写入"，混在一起会影响彼此性能。
四、应用场景分析
4.1 电商订单流场景的落地要点
在电商订单流里，Pulsar 的链式消费非常自然：
用户下单 → 订单服务产生消息 → Pulsar → 库存服务消费扣库存 → Pulsar → 支付服务消费发起支付 → Pulsar → 物流服务消费预约发货 → Pulsar → 通知服务消费发送短信/邮件
每一跳都用独立的 Topic，消费者之间通过 Pulsar 解耦。库存服务挂了，消息不会丢，它会停在 Pulsar 里等库存服务恢复后再消费。
关键配置汇总：订阅模式选 Failover，消息路由用 KeyHash，ACK 模式用 Individual，开启死信队列。
4.2 日志采集场景的落地要点
典型架构是：应用服务 → 日志SDK/Agent → Pulsar → 日志消费者 → Elasticsearch/S3/ClickHouse。
应用服务不需要直接依赖 Pulsar SDK，而是通过一个日志 Agent 把日志文件读取出来，批量发送到 Pulsar。关键配置：订阅模式选 Shared 让多个消费者分担消费，消息路由用 RoundRobin（日志不需要顺序），ACK 模式用 Cumulative 减少 ACK 次数，开启批量发送，压缩用 LZ4。
五、注意事项
5.1 Topic 管理
建议按照业务线划分 namespace，比如 ecommerce、logging、payment 各自独立。Topic 命名要有统一规范。不要在代码里硬编码 Topic 名，而是通过配置文件注入，这样可以灵活切换环境。
5.2 消息可靠性保障
消息不丢靠三个层面配合：生产端开启 blockIfQueueFull(true) 并区分同步异步发送，服务端配置合理的 replicationClusters 和 BookKeeper 的 ensembleSize/writeQuorum，消费端处理完消息后再 ACK 并用独立字段做幂等判断。
5.3 性能调优实战建议
批量发送：日志场景一定开，订单场景可选。接收队列：日志场景设 5000-10000。压缩：LZ4 是压缩率和速度的平衡点，ZSTD 压缩率更好但 CPU 开销更大。分区数：消费者数量乘以 2 到 4 比较合理。
5.4 运维监控要点
建议监控这些指标：pulsar_producer_messages_published_total（发送量）、pulsar_consumer_messages_consumed_total（消费量）、pulsar_backlog_size_bytes（积压大小）、pulsar_pending_messages（待确认消息数）。积压是订单流最关键的监控指标，如果积压持续增长说明消费跟不上生产。
5.5 多租户隔离
在一个集群上同时跑订单流和日志采集时，建议用不同的 tenant 做资源隔离。tenant 之间配额独立，一个租户的流量暴涨不会影响另一个租户。
# 创建两个租户的示例命令
pulsar-admin tenants create ecommerce
pulsar-admin tenants create logging
# 创建 namespace 并设置配额
pulsar-admin namespaces create ecommerce/order
pulsar-admin namespaces create logging/app-logs
# 为订单 namespace 设置消息保留策略：保留 7 天
pulsar-admin namespaces set-retention -n ecommerce/order -s 1TB -t 7d
# 为日志 namespace 设置保留策略：保留 24 小时
pulsar-admin namespaces set-retention -n logging/app-logs -s 500GB -t 24h
# 开启日志 namespace 的分层存储（冷数据自动下沉到 S3）
pulsar-admin namespaces set-tiered-storage -n logging/app-logs -p s3://log-archive-bucket
六、文章总结
Pulsar 在电商订单流和日志采集这两个场景下都能很好地胜任，但需要注意的核心点不同。订单流关注消息可靠性和顺序，日志采集关注吞吐和成本。同一个 Pulsar 集群可以承载这两种业务，只要配置得当、Topic 规划合理。
选 Pulsar 还是 Kafka，取决于你的团队能力和业务需求。如果你需要多租户隔离、分层存储、弹性扩缩容这些能力，Pulsar 是很好的选择。如果团队更熟悉 Kafka 生态且业务规模不需要太复杂的多租户能力，Kafka 也完全够用。
实际落地的建议：先在测试环境用 Docker Compose 部署一个最小集群跑通流程，验证消息可靠性和吞吐量后，再逐步迁移生产环境。最后一点很重要：不管选什么中间件，消息可靠性最终靠的是代码逻辑。幂等设计、死信队列处理、告警监控、灰度发布，这些软功夫比中间件本身的配置更关键。
评论
围绕“Pulsar多场景适配：电商订单流与日志采集的架构选型与落地要点”参与讨论