PostgreSQL触发器实现异步事件通知：利用NOTIFY向客户端发送消息

CREATE OR REPLACE FUNCTION send_data_notification()
  RETURNS TRIGGER
  LANGUAGE plpgsql
AS $$
DECLARE
    notification_payload TEXT;
BEGIN
    -- 构建通知内容，可以包含新行的数据 (NEW) 或旧行的数据 (OLD)
    -- TG_OP 变量表示触发器操作类型 (INSERT, UPDATE, DELETE)
    -- TG_TABLE_NAME 变量表示触发器所在表的名称

    IF TG_OP = 'INSERT' THEN
        -- 示例：当有新数据插入时，发送新行的ID和名称
        -- 假设表有 'id' 和 'name' 字段
        notification_payload := '新数据插入：表 ' || TG_TABLE_NAME || ', ID: ' || NEW.id || ', 名称: ' || NEW.name;
    ELSIF TG_OP = 'UPDATE' THEN
        -- 示例：当数据更新时，发送旧ID和新ID
        notification_payload := '数据更新：表 ' || TG_TABLE_NAME || ', 旧ID: ' || OLD.id || ', 新ID: ' || NEW.id;
    ELSIF TG_OP = 'DELETE' THEN
        -- 示例：当数据删除时，发送被删除行的ID
        notification_payload := '数据删除：表 ' || TG_TABLE_NAME || ', ID: ' || OLD.id;
    END IF;

    -- 使用NOTIFY命令发送消息
    -- "my_channel" 是通知通道的名称，客户端将监听此通道
    -- notification_payload 是要发送的消息内容
    NOTIFY "my_channel", notification_payload;

    -- 触发器函数必须返回一个行图像（NEW, OLD）或NULL。
    -- 对于AFTER触发器，返回NULL是常见的做法，因为它不影响数据操作。
    RETURN NULL;
END;
$$;

CREATE TABLE IF NOT EXISTS products (
    id SERIAL PRIMARY KEY,
    name VARCHAR(255) NOT NULL,
    price DECIMAL(10, 2)
);

CREATE TRIGGER product_data_change_notify_trigger
  AFTER INSERT OR UPDATE OR DELETE ON products
  FOR EACH ROW
  EXECUTE FUNCTION send_data_notification();

NOTIFY
Asynchronous notification "my_channel" with payload "新数据插入：表 products, ID: 1, 名称: Laptop" received from server process with PID 12345.
NOTIFY
Asynchronous notification "my_channel" with payload "数据更新：表 products, 旧ID: 1, 新ID: 1" received from server process with PID 12345.
NOTIFY
Asynchronous notification "my_channel" with payload "数据删除：表 products, ID: 1" received from server process with PID 12345.

import java.sql.*;

public class NotificationListener {

    public static void main(String[] args) {
        String url = "jdbc:postgresql://localhost:5432/your_database";
        String user = "your_user";
        String password = "your_password";
        String channel = "my_channel";

        try (Connection conn = DriverManager.getConnection(url, user, password);
             Statement stmt = conn.createStatement()) {

            // 1. 监听通道
            stmt.execute("LISTEN \"" + channel + "\"");
            System.out.println("Listening on channel: " + channel);

            // 2. 持续检查通知
            while (true) {
                // 检查是否有通知到达
                // PostgreSQL JDBC驱动的Statement.getNotifications()方法可以获取通知
                // 注意：这通常需要在一个独立的线程中执行，并且需要定期调用
                // 实际的JDBC驱动可能需要更复杂的实现，例如通过Connection.unwrap()获取PGConnection
                // 并使用PGConnection.getNotifications()

                // 简化示例，实际生产环境需根据具体JDBC驱动和版本实现
                // 以下代码片段是一个概念性的演示，具体实现需查阅JDBC驱动文档
                if (conn instanceof org.postgresql.PGConnection) {
                    org.postgresql.PGNotification[] notifications = 
                        ((org.postgresql.PGConnection) conn).getNotifications();
                    if (notifications != null) {
                        for (org.postgresql.PGNotification notification : notifications) {
                            System.out.println("Received notification: Channel=" + notification.getName() +
                                               ", PID=" + notification.getPID() +
                                               ", Payload=" + notification.getParameter());
                        }
                    }
                } else {
                    System.out.println("Current JDBC driver does not support PGConnection notifications directly.");
                    // Fallback for generic JDBC or polling approach
                }

                Thread.sleep(1000); // 每秒检查一次
            }

        } catch (SQLException | InterruptedException e) {
            e.printStackTrace();
        }
    }
}