如何用Apollo-Server订阅Postgres数据库，实时推送用户数据至React客户端？

完全满足你的需求：当第三方服务（Airflow/NiFi）向Postgres的User表插入新数据时，Apollo Server捕获该事件并主动推送给已订阅的React客户端。下面是具体实现步骤、代码示例和实践注意事项。

1. 核心原理 #

依赖Postgres的**变更数据捕获（CDC）**能力，结合Apollo Server的Subscription机制：

• Postgres端：通过触发器+LISTEN/NOTIFY机制，在User表插入新数据时发送事件通知
• Apollo Server端：监听Postgres的通知，将事件包装为GraphQL Subscription推送给客户端
• React客户端：通过Apollo Client订阅该事件，实时接收并渲染新用户数据

2. 分步实现代码示例 #

第一步：Postgres端配置变更通知 #

创建触发器，在新用户插入时向Apollo Server发送事件：

-- 创建触发器函数：插入新用户时发送JSON格式的通知
CREATE OR REPLACE FUNCTION notify_new_user()
RETURNS TRIGGER AS $$
BEGIN
  -- 发送名为"new_user_event"的通知，携带新用户的JSON数据
  PERFORM pg_notify('new_user_event', row_to_json(NEW)::TEXT);
  RETURN NEW;
END;
$$ LANGUAGE plpgsql;

-- 给User表绑定INSERT触发器
CREATE TRIGGER trigger_new_user
AFTER INSERT ON "User"
FOR EACH ROW EXECUTE FUNCTION notify_new_user();

第二步：Apollo Server端配置Subscription #

使用apollo-server-express+graphql-ws实现WebSocket订阅（Apollo v4+推荐该方案）：

安装依赖

npm install apollo-server-express express graphql graphql-ws ws pg

Server核心代码

const express = require('express');
const { ApolloServer } = require('apollo-server-express');
const { createServer } = require('http');
const { WebSocketServer } = require('ws');
const { useServer } = require('graphql-ws/lib/use/ws');
const { Client } = require('pg');
const { gql } = require('apollo-server-express');

// 1. 定义GraphQL Schema
const typeDefs = gql`
  type User {
    id: ID!
    name: String!
    email: String!
    # 根据你的User表结构补充其他字段
  }

  type Query {
    # 保留原有查询接口
    users: [User!]!
  }

  type Subscription {
    # 定义订阅接口：推送新用户数据
    newUser: User!
  }
`;

// 2. 定义Resolver
const resolvers = {
  Query: {
    async users(_, __, { pgClient }) {
      const result = await pgClient.query('SELECT * FROM "User"');
      return result.rows;
    },
  },
  Subscription: {
    newUser: {
      // 订阅逻辑：监听Postgres的通知并返回异步迭代器
      async subscribe(_, __, { pgClient }) {
        // 监听Postgres的"new_user_event"频道
        await pgClient.query('LISTEN new_user_event');

        return {
          async next() {
            return new Promise(resolve => {
              pgClient.once('notification', msg => {
                if (msg.channel === 'new_user_event') {
                  const newUser = JSON.parse(msg.payload);
                  resolve({ value: { newUser }, done: false });
                }
              });
            });
          },
          // 取消订阅时停止监听
          async return() {
            await pgClient.query('UNLISTEN new_user_event');
            return { done: true };
          },
          // 错误处理时停止监听
          async throw(err) {
            await pgClient.query('UNLISTEN new_user_event');
            throw err;
          },
        };
      },
    },
  },
};

// 3. 初始化Postgres客户端
const pgClient = new Client({
  connectionString: 'postgresql://username:password@localhost:5432/your_db_name',
  // 替换为你的Postgres连接信息
});

// 4. 启动服务器
async function startServer() {
  await pgClient.connect();

  const app = express();
  const httpServer = createServer(app);

  // 配置WebSocket服务器
  const wsServer = new WebSocketServer({
    server: httpServer,
    path: '/graphql',
  });

  const apolloServer = new ApolloServer({
    typeDefs,
    resolvers,
    context: () => ({ pgClient }),
  });

  await apolloServer.start();
  apolloServer.applyMiddleware({ app });

  // 绑定GraphQL Schema到WebSocket服务器
  useServer({ schema: apolloServer.schema }, wsServer);

  const PORT = 4000;
  httpServer.listen(PORT, () => {
    console.log(`HTTP服务运行：http://localhost:${PORT}${apolloServer.graphqlPath}`);
    console.log(`WebSocket服务运行：ws://localhost:${PORT}${apolloServer.graphqlPath}`);
  });
}

startServer().catch(err => console.error('服务启动失败：', err));

第三步：React客户端订阅实现 #

使用Apollo Client的useSubscription钩子实时接收数据：

安装依赖

npm install @apollo/client graphql graphql-ws

客户端核心代码

import { ApolloClient, InMemoryCache, ApolloProvider, useSubscription, gql } from '@apollo/client';
import { GraphQLWsLink } from '@apollo/client/link/subscriptions';
import { createClient } from 'graphql-ws';

// 创建WebSocket链接
const wsLink = new GraphQLWsLink(createClient({
  url: 'ws://localhost:4000/graphql',
}));

// 初始化Apollo Client
const client = new ApolloClient({
  uri: 'http://localhost:4000/graphql',
  cache: new InMemoryCache(),
  link: wsLink,
});

// 订阅新用户的组件
const NewUserAlert = () => {
  const { data, loading, error } = useSubscription(gql`
    subscription {
      newUser {
        id
        name
        email
      }
    }
  `);

  if (loading) return <p>等待新用户...</p>;
  if (error) return <p>订阅失败：{error.message}</p>;

  return data?.newUser ? (
    <div className="alert alert-success">
      <h4>新用户加入！</h4>
      <p>姓名：{data.newUser.name}</p>
      <p>邮箱：{data.newUser.email}</p>
    </div>
  ) : null;
};

// 主应用组件
function App() {
  return (
    <ApolloProvider client={client}>
      <div className="App">
        <h1>用户管理系统</h1>
        <NewUserAlert />
        {/* 其他用户列表组件 */}
      </div>
    </ApolloProvider>
  );
}

export default App;

3. 实践注意事项 #

• Postgres连接稳定性：LISTEN/NOTIFY是会话级别的，要确保Apollo Server的Postgres客户端连接保持活跃，可添加重连逻辑避免断开后丢失监听。
• 复杂场景扩展：如果需要监听更新/删除事件，或处理大量数据变更，可考虑使用Debezium等专业CDC工具，替代轻量的LISTEN/NOTIFY。
• 权限控制：确保Postgres用户拥有数据库USAGE权限和User表的SELECT权限（触发器函数需要访问NEW行数据）。
• 错误处理：在Server和Client端都要处理WebSocket断开、Postgres连接失败等异常情况，添加自动重连逻辑。
• 版本兼容：Apollo Server v4+推荐使用graphql-ws，旧版的subscriptions-transport-ws已被弃用。

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