消息队列对比：Kafka vs RocketMQ vs Pulsar

// Kafka Producer
Properties props = new Properties();
props.put("bootstrap.servers", "kafka1:9092,kafka2:9092");
props.put("key.serializer", StringSerializer.class.getName());
props.put("value.serializer", StringSerializer.class.getName());
props.put("acks", "all");               // 等待所有 ISR 副本确认
props.put("retries", 3);
props.put("enable.idempotence", true);   // 幂等生产者

KafkaProducer<String, String> producer = new KafkaProducer<>(props);

ProducerRecord<String, String> record = new ProducerRecord<>(
    "order-topic",                       // topic
    order.getId(),                       // key (决定分区)
    JsonUtils.toJson(order)              // value
);

producer.send(record, (metadata, exception) -> {
    if (exception == null) {
        log.info("Sent to partition={}, offset={}",
            metadata.partition(), metadata.offset());
    } else {
        log.error("Send failed", exception);
    }
});

// Kafka Consumer
Properties consumerProps = new Properties();
consumerProps.put("bootstrap.servers", "kafka1:9092");
consumerProps.put("group.id", "order-processor");
consumerProps.put("enable.auto.commit", false);
consumerProps.put("auto.offset.reset", "earliest");

KafkaConsumer<String, String> consumer = new KafkaConsumer<>(consumerProps);
consumer.subscribe(List.of("order-topic"));

while (true) {
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
    for (ConsumerRecord<String, String> record : records) {
        processOrder(record.value());
    }
    consumer.commitSync(); // 手动提交 offset
}

// RocketMQ Producer
DefaultMQProducer producer = new DefaultMQProducer("order-producer-group");
producer.setNamesrvAddr("namesrv1:9876;namesrv2:9876");
producer.setSendMsgTimeout(3000);
producer.setRetryTimesWhenSendFailed(3);
producer.start();

Message msg = new Message(
    "order-topic",          // topic
    "OrderCreated",         // tag
    order.getId(),          // key
    JsonUtils.toJson(order).getBytes(StandardCharsets.UTF_8) // body
);

// 同步发送
SendResult result = producer.send(msg);
log.info("Send result: {}, msgId: {}", result.getSendStatus(), result.getMsgId());

// 延迟消息
msg.setDelayTimeLevel(3);  // 延迟10s (1s 5s 10s 30s ...)
producer.send(msg);

// RocketMQ Consumer
DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("order-consumer-group");
consumer.setNamesrvAddr("namesrv1:9876");
consumer.subscribe("order-topic", "OrderCreated || OrderPaid");
consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);

consumer.registerMessageListener((MessageListenerConcurrently) (msgs, context) -> {
    for (MessageExt msg : msgs) {
        try {
            Order order = JsonUtils.fromJson(new String(msg.getBody()), Order.class);
            processOrder(order);
        } catch (Exception e) {
            log.error("Process message failed, msgId={}", msg.getMsgId(), e);
            return ConsumeConcurrentlyStatus.RECONSUME_LATER; // 稍后重试
        }
    }
    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
});

consumer.start();

// Pulsar Producer
PulsarClient client = PulsarClient.builder()
    .serviceUrl("pulsar://pulsar1:6650,pulsar2:6650")
    .build();

Producer<Order> producer = client.newProducer(Schema.JSON(Order.class))
    .topic("persistent://public/default/orders")
    .batchingMaxPublishDelay(10, TimeUnit.MILLISECONDS)
    .sendTimeout(30, TimeUnit.SECONDS)
    .blockIfQueueFull(true)
    .create();

// 异步发送
producer.sendAsync(order).thenAccept(msgId -> {
    log.info("Published message: {}", msgId);
}).exceptionally(ex -> {
    log.error("Publish failed", ex);
    return null;
});

// Pulsar Consumer - 支持多种订阅模式
Consumer<Order> consumer = client.newConsumer(Schema.JSON(Order.class))
    .topic("persistent://public/default/orders")
    .subscriptionName("order-processor")
    .subscriptionType(SubscriptionType.Key_Shared) // 按 key 分发
    .ackTimeout(30, TimeUnit.SECONDS)
    .subscribe();

while (true) {
    Message<Order> msg = consumer.receive();
    try {
        processOrder(msg.getValue());
        consumer.acknowledge(msg);
    } catch (Exception e) {
        consumer.negativeAcknowledge(msg); // 否定确认，触发重投
    }
}

// Kafka 顺序消息
producer.send(new ProducerRecord<>(
    "order-topic",
    orderId,       // 相同 orderId 路由到同一 partition
    orderEvent
));

// RocketMQ 顺序消息
producer.send(msg, (mqs, msg1, arg) -> {
    String orderId = (String) arg;
    int index = Math.abs(orderId.hashCode()) % mqs.size();
    return mqs.get(index); // 相同 orderId 路由到同一 queue
}, orderId);

// 顺序消费
consumer.registerMessageListener((MessageListenerOrderly) (msgs, context) -> {
    for (MessageExt msg : msgs) {
        processOrderEvent(msg);
    }
    return ConsumeOrderlyStatus.SUCCESS;
});

// Kafka Exactly-Once (事务生产者 + 幂等消费者)
props.put("enable.idempotence", true);
props.put("transactional.id", "order-tx-001");

KafkaProducer<String, String> producer = new KafkaProducer<>(props);
producer.initTransactions();

try {
    producer.beginTransaction();

    // 发送消息
    producer.send(new ProducerRecord<>("output-topic", key, value));

    // 提交消费者 offset（作为事务的一部分）
    producer.sendOffsetsToTransaction(
        Map.of(new TopicPartition("input-topic", 0),
               new OffsetAndMetadata(offset + 1)),
        consumerGroupMetadata
    );

    producer.commitTransaction();
} catch (Exception e) {
    producer.abortTransaction();
}