1
// Kafka 的核心抽象：分区日志
2
// 每个 Topic 由多个 Partition 组成，Partition 内消息有序
3
// 消费者通过 Offset 标记消费位置
4
Properties props = new Properties();
5
props.put("bootstrap.servers", "kafka:9092");
6
props.put("key.serializer", StringSerializer.class.getName());
7
props.put("value.serializer", StringSerializer.class.getName());
8
props.put("acks", "all");              // 等待所有 ISR 确认
9
props.put("enable.idempotence", "true"); // 幂等生产者
10
props.put("compression.type", "lz4");    // 压缩减少网络开销
11

12
KafkaProducer<String, String> producer = new KafkaProducer<>(props);
13
// 发送到指定分区（按 Key hash），保证同 Key 有序
14
producer.send(new ProducerRecord<>("order-events", orderId, orderJson));

1
// RabbitMQ 的核心抽象：Exchange → Binding → Queue
2
// Producer 发到 Exchange，Exchange 路由到 Queue
3
ConnectionFactory factory = new ConnectionFactory();
4
factory.setHost("rabbitmq");
5
try (Connection conn = factory.newConnection();
6
     Channel ch = conn.createChannel()) {
7
    // 声明 Topic Exchange
8
    ch.exchangeDeclare("order-events", "topic", true);
9
    // 绑定：order.new → 支付队列
10
    ch.queueBind("payment-queue", "order-events", "order.new");
11
    // 绑定：order.# → 通知队列（# 匹配多级）
12
    ch.queueBind("notify-queue", "order-events", "order.#");
13
    // 发布消息
14
    ch.basicPublish("order-events", "order.new",
15
        MessageProperties.PERSISTENT_TEXT_PLAIN, orderJson.getBytes());
16
}

1
// RocketMQ 事务消息：两阶段提交 + 回查
2
TransactionMQProducer producer = new TransactionMQProducer("order-group");
3
producer.setTransactionListener(new TransactionListener() {
4
    @Override
5
    public LocalTransactionState executeLocalTransaction(Message msg, Object arg) {
6
        try {
7
            // 执行本地事务：创建订单
8
            orderService.createOrder(parseOrder(msg));
9
            return LocalTransactionState.COMMIT_MESSAGE;
10
        } catch (Exception e) {
11
            return LocalTransactionState.ROLLBACK_MESSAGE;
12
        }
13
    }
14

15
    @Override
16
    public LocalTransactionState checkLocalTransaction(MessageExt msg) {
17
        // 回查：订单是否创建成功？
18
        Order order = orderService.getById(parseOrderId(msg));
19
        return order != null ? COMMIT_MESSAGE : ROLLBACK_MESSAGE;
20
    }
21
});

1
// Pulsar 的多租户 + 多订阅模式
2
PulsarClient client = PulsarClient.builder()
3
    .serviceUrl("pulsar://pulsar:6650").build();
4

5
// 租户/命名空间隔离
6
Producer<String> producer = client.newProducer(Schema.STRING)
7
    .topic("persistent://tenant-1/order-ns/order-events")
8
    .enableBatching(true)
9
    .batchingMaxMessages(1000)
10
    .create();
11

12
// 多种订阅模式：Exclusive / Failover / Shared / Key_Shared
13
Consumer<String> consumer = client.newConsumer(Schema.STRING)
14
    .topic("persistent://tenant-1/order-ns/order-events")
15
    .subscriptionName("payment-sub")
16
    .subscriptionType(SubscriptionType.Key_Shared) // Key 有序
17
    .subscribe();

1
// 金融交易：RocketMQ 事务消息 + 顺序消息
2
// 1. 转账操作：事务消息保证 DB 和 MQ 一致
3
TransactionMQProducer producer = new TransactionMQProducer("transfer-group");
4
producer.setTransactionListener(new TransactionListener() {
5
    @Override
6
    public LocalTransactionState executeLocalTransaction(Message msg, Object arg) {
7
        // 扣减余额 + 记录流水（本地事务）
8
        accountService.transfer(transferRequest);
9
        return LocalTransactionState.COMMIT_MESSAGE;
10
    }
11
    @Override
12
    public LocalTransactionState checkLocalTransaction(MessageExt msg) {
13
        TransferRecord record = transferService.getById(parseId(msg));
14
        return record != null ? COMMIT_MESSAGE : ROLLBACK_MESSAGE;
15
    }
16
});
17

18
// 2. 顺序消息：同一账户的操作必须有序
19
Message msg = new Message("transfer-events", tag, transferJson.getBytes());
20
// 使用 accountId 作为 Key，保证同一账户的事件在同一个队列
21
msg.setKeys(accountId);
22
SendResult result = producer.send(msg, new MessageQueueSelector() {
23
    @Override
24
    public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {
25
        int index = Math.abs(accountId.hashCode()) % mqs.size();
26
        return mqs.get(index);
27
    }
28
}, accountId);

1
# Kafka 日志收集：Filebeat → Kafka → Logstash → Elasticsearch
2
# Filebeat 配置
3
filebeat.inputs:
4
  - type: log
5
    paths: ["/var/log/app/*.log"]
6
    fields:
7
      app: order-service
8
      env: production
9

10
output.kafka:
11
  hosts: ["kafka:9092"]
12
  topic: "app-logs"
13
  partition.round_robin:
14
    reachable_only: true
15
  required_acks: 1
16
  compression: gzip
17
  max_message_bytes: 1000000

1
# Pulsar IoT 接入：MQTT 协议 + 分层存储
2
# Pulsar MQTT 协议适配器
3
mqtt:
4
  enabled: true
5
  port: 1883
6
  maxConnections: 100000
7

8
# 分层存储策略：热数据 7 天，冷数据转 S3
9
policies:
10
  retention:
11
    retentionTimeInMinutes: 10080  # 7 天热数据
12
    retentionSizeInMB: -1
13
  offload:
14
    offloadThresholdInSeconds: 604800  # 7 天后转冷
15
    offloadPolicies:
16
      managedLedgerOffloadDriver: s3
17
      s3Region: us-east-1
18
      s3Bucket: pulsar-offload

1
// 微服务通信：RabbitMQ Topic Exchange 精细路由
2
// 订单服务发布事件
3
channel.basicPublish("order-events", "order.created", null, event1);
4
channel.basicPublish("order-events", "order.cancelled", null, event2);
5
channel.basicPublish("order-events", "order.shipped", null, event3);
6

7
// 支付服务只订阅创建事件
8
channel.queueBind("payment-queue", "order-events", "order.created");
9

10
// 通知服务订阅所有事件
11
channel.queueBind("notify-queue", "order-events", "order.#");
12

13
// 物流服务只订阅发货事件
14
channel.queueBind("shipping-queue", "order-events", "order.shipped");

1
// 实时分析：Kafka Streams 窗口聚合
2
StreamsBuilder builder = new StreamsBuilder();
3
KStream<String, OrderEvent> orders = builder.stream("order-events");
4

5
// 按客户分组，5 分钟窗口统计
6
KTable<Windowed<String>, OrderStats> stats = orders
7
    .groupBy((key, event) -> event.getCustomerId())
8
    .windowedBy(TimeWindows.of(Duration.ofMinutes(5)))
9
    .aggregate(
10
        OrderStats::new,
11
        (customerId, event, aggregate) -> aggregate.add(event),
12
        Materialized.with(Serdes.String(), new JsonSerde<>(OrderStats.class))
13
    );
14

15
// 输出到统计 Topic
16
stats.toStream()
17
    .map((key, value) -> new KeyValue<>(key.key(), value))
18
    .to("order-stats");

1
// Kafka → RabbitMQ 桥接
2
@KafkaListener(topics = "order-events")
3
public void onOrderEvent(ConsumerRecord<String, String> record) {
4
    // 将 Kafka 事件转发到 RabbitMQ
5
    rabbitTemplate.convertAndSend("order-exchange", "order.new", record.value());
6
}
7

8
// RabbitMQ → Kafka 桥接
9
@RabbitListener(queues = "analytics-queue")
10
public void onAnalyticsMessage(Message message) {
11
    // 将 RabbitMQ 消息转发到 Kafka
12
    kafkaTemplate.send("analytics-topic", new String(message.getBody()));
13
}

1
// 可靠桥接：使用本地消息表保证一致性
2
@Transactional
3
public void bridgeToRabbitMQ(ConsumerRecord<String, String> record) {
4
    // 1. 检查是否已桥接（幂等）
5
    if (bridgeLogRepository.exists(record.topic(), record.offset())) {
6
        return;  // 已桥接，跳过
7
    }
8

9
    // 2. 转发到 RabbitMQ
10
    rabbitTemplate.convertAndSend("order-exchange", "order.new", record.value());
11

12
    // 3. 记录桥接日志
13
    bridgeLogRepository.save(new BridgeLog(record.topic(), record.offset()));
14
}

1
# 监控桥接状态
2
# Kafka Connect 状态
3
curl http://localhost:8083/connectors/rabbitmq-sink/status
4

5
# RabbitMQ Shovel 状态
6
rabbitmqctl list_shovels

1
// 迁移第一步：双写适配器
2
@Service
3
public class DualWriteMessageService {
4
    private final RabbitTemplate rabbitTemplate;
5
    private final KafkaTemplate<String, String> kafkaTemplate;
6
    private final FeatureFlagService featureFlag;
7

8
    public void sendOrderEvent(String routingKey, String payload) {
9
        // 始终写入 RabbitMQ（旧系统）
10
        rabbitTemplate.convertAndSend("order-exchange", routingKey, payload);
11

12
        // 特性开关控制是否写入 Kafka
13
        if (featureFlag.isEnabled("kafka-dual-write")) {
14
            try {
15
                kafkaTemplate.send("order-events", routingKey, payload).get(5, TimeUnit.SECONDS);
16
            } catch (Exception e) {
17
                // 双写阶段 Kafka 写入失败不影响主流程
18
                log.warn("Kafka dual-write failed", e);
19
                metricsService.increment("kafka.dual-write.failure");
20
            }
21
        }
22
    }
23
}
24

25
// 迁移第二步：对比验证
26
@Scheduled(fixedRate = 60000)
27
public void verifyDualWriteConsistency() {
28
    // 抽样对比 RabbitMQ 和 Kafka 的消息
29
    long rabbitCount = rabbitAdmin.getQueueMessageCount("order-queue");
30
    long kafkaLag = kafkaAdmin.getConsumerGroupLag("order-consumer-group");
31
    metricsService.gauge("migration.consistency.gap", Math.abs(rabbitCount - kafkaLag));
32
}