3. 消息队列基础：主流MQ选型与PostgreSQL集成场景

消息队列（Message Queue, MQ）是分布式系统中实现异步解耦、流量削峰和数据最终一致性的核心中间件。在集成测试场景中，MQ 与 PostgreSQL 的配合尤为关键——MQ 负责事件流转，PostgreSQL 负责状态持久化与事务保障。本章聚焦三大主流 MQ（RabbitMQ、Kafka、Pulsar）的选型对比，并梳理它们与 PostgreSQL 集成的典型场景。

3.1 主流MQ选型对比

下表从协议、模型、性能、持久化、运维复杂度等维度，对 RabbitMQ、Kafka、Pulsar 进行横向对比：

3.2 消息队列与PostgreSQL集成的典型场景

MQ 与 PostgreSQL 的集成主要围绕以下三类模式展开：

1. 事件驱动 + 状态持久化：应用将业务事件发送到 MQ，消费者处理后更新 PG 状态表。
2. CDC 数据同步：通过 Debezium 等工具捕获 PG 的 WAL 日志变更，推送到 MQ 供下游消费。
3. 异步任务调度：将耗时任务（如报表生成、数据导出）封装为消息，由 Worker 消费并写入 PG。

下图展示了三种集成模式的架构对比：

3.3 各MQ与PostgreSQL集成的技术要点

3.3.1 RabbitMQ + PostgreSQL

• 原生集成方式：利用 PostgreSQL 的 LISTEN / NOTIFY 机制，结合 pg_notify 扩展将数据库事件推送到 RabbitMQ。
• 事务消息：通过 RabbitMQ 的 Publisher Confirm 和 Consumer Ack 机制，配合 PG 的本地事务，实现“数据库操作 + 消息发送”的最终一致性。
• 典型代码片段（Spring Boot + RabbitMQ + PG）：

@Transactional
public void createOrder(Order order) {
    // 1. 写入 PostgreSQL 订单表
    orderRepository.save(order);
    // 2. 发送消息到 RabbitMQ（利用事务同步器确保提交后发送）
    TransactionSynchronizationManager.registerSynchronization(
        new TransactionSynchronization() {
            @Override
            public void afterCommit() {
                rabbitTemplate.convertAndSend("order.exchange", "order.created", order);
            }
        }
    );
}

3.3.2 Kafka + PostgreSQL

• CDC 集成：Debezium 的 PostgreSQL Connector 读取 WAL（Write-Ahead Log）日志，将 INSERT/UPDATE/DELETE 事件转换为 Kafka 消息。
• Exactly-Once 语义：Kafka 的幂等生产者 + 事务 API，配合 PG 的幂等消费逻辑（通过唯一键去重），实现端到端精确一次。
• 典型配置（Debezium PG Connector）：

{
  "name": "pg-connector",
  "config": {
    "connector.class": "io.debezium.connector.postgresql.PostgresConnector",
    "database.hostname": "localhost",
    "database.port": "5432",
    "database.user": "debezium",
    "database.password": "dbz",
    "database.dbname": "mydb",
    "database.server.name": "pg-server",
    "plugin.name": "pgoutput",
    "slot.name": "debezium_slot",
    "publication.name": "dbz_publication",
    "table.include.list": "public.orders"
  }
}

3.3.3 Pulsar + PostgreSQL

• Pulsar IO Connector：Pulsar 提供原生的 JDBC Sink Connector，可直接将消息写入 PostgreSQL 表。
• 分层存储：Pulsar 的 BookKeeper 存储层支持将冷数据卸载到 PG 或对象存储，适合长时间回溯的集成测试场景。
• 典型配置（Pulsar JDBC Sink）：

{
  "tenant": "public",
  "namespace": "default",
  "name": "pg-sink",
  "inputs": ["persistent://public/default/orders"],
  "configs": {
    "jdbcUrl": "jdbc:postgresql://localhost:5432/mydb",
    "tableName": "orders",
    "userName": "pulsar",
    "password": "pulsar",
    "insertMode": "UPSERT",
    "keyConverter": "org.apache.pulsar.client.impl.schema.AvroConverter"
  }
}

3.4 选型决策树

根据实际业务需求，可参考以下决策路径选择 MQ：