如何将PostgreSQL与RabbitMQ对接，实现数据更新触发消息队列？

一、抛弃老旧工具，采用原生+轻量Docker服务方案 #

1. 用PostgreSQL原生触发器捕获行变更 #

PostgreSQL自带的pg_notify函数可在触发器中发送变更通知，无需依赖废弃扩展，完全适配PostgreSQL 12+版本：

• 创建触发器函数，监听目标表行更新并发送JSON格式通知到指定通道：

CREATE OR REPLACE FUNCTION notify_row_update()
RETURNS TRIGGER AS $$
BEGIN
  PERFORM pg_notify(
    'row_updates', 
    json_build_object(
      'table', TG_TABLE_NAME,
      'old_row', row_to_json(OLD),
      'new_row', row_to_json(NEW)
    )::text
  );
  RETURN NEW;
END;
$$ LANGUAGE plpgsql;

• 给目标表绑定触发器：

CREATE TRIGGER trigger_row_update
AFTER UPDATE ON your_target_table
FOR EACH ROW EXECUTE FUNCTION notify_row_update();

2. 自定义Docker化中间服务转发通知到RabbitMQ #

放弃多年未维护的area51/notify-rabbit，自己写极简Node.js/Python服务并打包成Docker镜像，可部署在EC2或直接放入EKS，完全满足容器化要求：
示例Node.js服务（依赖pg和amqplib库）：

const { Client } = require('pg');
const amqp = require('amqplib');

// PostgreSQL配置
const pgClient = new Client({
  host: 'your-postgres-ec2-ip',
  port: 5432,
  user: 'your-user',
  password: 'your-password',
  database: 'your-db'
});

// RabbitMQ配置
const rabbitmqUrl = 'amqp://your-rabbitmq-ec2-ip:5672';
const queueName = 'postgres_row_updates';

async function start() {
  // 连接PostgreSQL并监听通知通道
  await pgClient.connect();
  await pgClient.query('LISTEN row_updates;');
  
  // 连接RabbitMQ并声明持久化队列
  const rabbitConn = await amqp.connect(rabbitmqUrl);
  const channel = await rabbitConn.createChannel();
  await channel.assertQueue(queueName, { durable: true });

  // 转发PostgreSQL通知到RabbitMQ
  pgClient.on('notification', (msg) => {
    if (msg.channel === 'row_updates') {
      channel.sendToQueue(queueName, Buffer.from(msg.payload), { persistent: true });
      console.log('Forwarded update to RabbitMQ:', msg.payload);
    }
  });

  console.log('Service running: listening for PostgreSQL updates');
}

start().catch(err => console.error('Service startup failed:', err));

配套Dockerfile：

FROM node:20-alpine
WORKDIR /app
COPY package*.json ./
RUN npm install --production
COPY . .
CMD ["node", "index.js"]

构建镜像后，可直接在EC2用Docker运行，或部署到EKS Pod中，不会出现启动即停止的问题。

二、RabbitMQ消息投递到EKS的实现 #

1. EKS内部部署消费者Pod #

在EKS中部署Deployment，运行Docker化的消费者服务，监听RabbitMQ队列并执行任务：
示例Kubernetes Deployment配置：

apiVersion: apps/v1
kind: Deployment
metadata:
  name: rabbitmq-consumer
spec:
  replicas: 2
  selector:
    matchLabels:
      app: rabbitmq-consumer
  template:
    metadata:
      labels:
        app: rabbitmq-consumer
    spec:
      containers:
      - name: consumer
        image: your-docker-repo/rabbitmq-consumer:latest
        env:
        - name: RABBITMQ_URL
          value: amqp://your-rabbitmq-ec2-ip:5672
        - name: QUEUE_NAME
          value: postgres_row_updates

2. 云原生替代方案：AWS SQS + Lambda #

若倾向AWS托管服务，可替换RabbitMQ为SQS：

• 用上述PostgreSQL触发器+轻量服务把变更通知转发到SQS
• 配置Lambda监听SQS，收到消息后调用EKS API创建Job/CronJob执行任务
  此方案无需维护消息代理服务器，更贴合AWS生态。

三、旧工具失效原因说明 #

• pg_amqp扩展多年未更新，不兼容PostgreSQL 12+，会破坏数据库运行环境，直接丢弃即可。
• area51/notify-rabbit依赖的Node.js版本和第三方库均已过时，启动失败属于必然情况。

四、当前最佳实践总结 #

1. 用PostgreSQL原生触发器+pg_notify捕获行变更，避免依赖老旧组件。
2. 自定义Docker化中间服务转发通知，可控性强，适配最新版本的PostgreSQL和RabbitMQ。
3. 优先在EKS内部部署消费者执行任务；若偏好托管服务，采用SQS+Lambda触发EKS任务更贴合云原生架构。

内容的提问来源于stack exchange，提问作者JackLidge