4、生产者开发基础：创建生产者实例、同步发送消息、异步发送消息、发送回调处理、消息发送结果解析（MessageId、发送状态）

生产者（Producer）是消息队列中消息的发起端，负责将业务数据封装为消息并发送至 Broker。本章围绕 OpenAPI 的生产者客户端，系统讲解从实例创建到发送结果解析的完整流程，涵盖同步与异步两种发送模式及其回调处理机制。

4.1 生产者实例创建

创建生产者实例是消息发送的第一步。通过 OpenAPI 提供的客户端工厂或 Builder 模式，开发者可以配置连接参数、线程模型与序列化方式。

// 创建生产者实例示例（Java）
Producer producer = new ProducerBuilder()
    .endpoint("192.168.1.100:9876")
    .namespace("production-order")
    .accessKey("your-access-key")
    .secretKey("your-secret-key")
    .sendTimeout(3000)
    .retryTimes(2)
    .build();
producer.start();

4.2 同步发送消息

同步发送是最基础的发送模式。调用 send() 方法后，当前线程会阻塞等待 Broker 返回发送结果。适用于对消息发送可靠性要求极高、且可以容忍短暂延迟的场景。

// 同步发送消息示例
Message message = new Message()
    .topic("order-topic")
    .tag("create")
    .body("{\"orderId\":\"20250301001\"}".getBytes(StandardCharsets.UTF_8));

SendResult result = producer.send(message);
System.out.println("MessageId: " + result.getMessageId());
System.out.println("SendStatus: " + result.getSendStatus());

4.3 异步发送消息

异步发送允许调用立即返回，发送结果通过回调接口在后台线程中处理。适用于高吞吐、低延迟要求的场景，如日志采集、实时监控数据上报。

// 异步发送消息示例
producer.sendAsync(message, new SendCallback() {
    @Override
    public void onSuccess(SendResult result) {
        System.out.println("异步发送成功，MessageId: " + result.getMessageId());
    }

    @Override
    public void onException(Throwable e) {
        System.err.println("异步发送失败，异常: " + e.getMessage());
    }
});

4.4 发送回调处理

回调处理是异步发送的核心环节。合理的回调设计能够提升系统的容错能力与可观测性。下表对比了两种常见的回调处理策略：

// 异步队列中转回调示例
private final BlockingQueue<SendResult> successQueue = new LinkedBlockingQueue<>(10000);
private final BlockingQueue<Throwable> failureQueue = new LinkedBlockingQueue<>(1000);

producer.sendAsync(message, new SendCallback() {
    @Override
    public void onSuccess(SendResult result) {
        successQueue.offer(result);
    }

    @Override
    public void onException(Throwable e) {
        failureQueue.offer(e);
    }
});

// 独立线程批量处理成功结果
new Thread(() -> {
    while (true) {
        SendResult result = successQueue.take();
        // 批量写入数据库或日志
    }
}).start();

4.5 消息发送结果解析

发送结果对象 SendResult 包含两个关键字段：MessageId 和 SendStatus。MessageId 是消息在消息队列中的全局唯一标识，用于后续的消息追踪、消费确认与死信处理。SendStatus 表示消息是否被 Broker 成功接收。

// 发送结果解析示例
SendResult result = producer.send(message);

// 获取 MessageId
String messageId = result.getMessageId();
// 建议将 messageId 与业务主键关联存储，便于后续追踪

// 获取发送状态
SendStatus status = result.getSendStatus();
switch (status) {
    case SEND_OK:
        // 发送成功，业务逻辑继续
        break;
    case FLUSH_DISK_TIMEOUT:
        // Broker 刷盘超时，消息可能已写入但未落盘
        // 高可靠性场景建议重试
        break;
    case FLUSH_SLAVE_TIMEOUT:
        // 主从同步超时，消息可能未同步到从节点
        // 对一致性要求高的场景建议重试
        break;
    default:
        // 其他异常状态，记录日志并告警
        break;
}

4.6 同步与异步发送对比总结