PostgreSQL触发器结合NOTIFY实现异步事件通知

NOTIFY channel_name;
NOTIFY channel_name, 'payload';

CREATE OR REPLACE FUNCTION send_notify()
  RETURNS trigger
  LANGUAGE plpgsql
AS $$
BEGIN
    -- NOTIFY命令用于向'my_channel'频道发送通知
    -- 这里的NEW.id::text 是一个示例，表示将新插入行的ID作为通知的payload
    -- 您可以根据需要构建更复杂的payload，例如JSON字符串
    NOTIFY "my_channel", NEW.id::text || ' inserted into ' || TG_TABLE_NAME;

    -- 对于AFTER触发器，函数通常返回NULL
    RETURN NULL;
END;
$$;

CREATE TRIGGER send_notify_air
  AFTER INSERT -- 指定在INSERT操作之后触发
  ON some_table_name -- 替换为您的目标表名
  FOR EACH ROW -- 指定对每一行受影响的数据执行触发器
  EXECUTE FUNCTION send_notify(); -- 调用之前创建的函数

-- 1. 创建示例表
CREATE TABLE products (
    id SERIAL PRIMARY KEY,
    name VARCHAR(100) NOT NULL,
    price NUMERIC(10, 2)
);

-- 2. 创建通知函数
CREATE OR REPLACE FUNCTION product_insert_notify()
  RETURNS trigger
  LANGUAGE plpgsql
AS $$
BEGIN
    -- 发送通知，包含新产品的ID和名称
    NOTIFY "product_updates", 
           '{"id": ' || NEW.id::text || ', "name": "' || NEW.name || '", "event": "insert"}';
    RETURN NULL;
END;
$$;

-- 3. 创建触发器
CREATE TRIGGER trg_product_insert_notify
  AFTER INSERT
  ON products
  FOR EACH ROW
  EXECUTE FUNCTION product_insert_notify();

-- 4. 插入数据以测试触发器
INSERT INTO products (name, price) VALUES ('Laptop', 1200.00);
INSERT INTO products (name, price) VALUES ('Mouse', 25.50);

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.SQLException;
import java.sql.Statement;
import org.postgresql.PGConnection;
import org.postgresql.PGNotification;

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";

        try (Connection conn = DriverManager.getConnection(url, user, password)) {
            // 转换为PGConnection以访问PostgreSQL特有的功能
            PGConnection pgconn = conn.unwrap(PGConnection.class);

            // 创建一个Statement来执行LISTEN命令
            try (Statement stmt = conn.createStatement()) {
                stmt.execute("LISTEN \"product_updates\""); // 监听上面定义的频道
                System.out.println("Listening for notifications on channel 'product_updates'...");
            }

            // 持续轮询通知
            while (true) {
                PGNotification[] notifications = pgconn.getNotifications();
                if (notifications != null && notifications.length > 0) {
                    for (PGNotification notification : notifications) {
                        System.out.println("Received notification:");
                        System.out.println("  Channel: " + notification.getName());
                        System.out.println("  Payload: " + notification.getParameter());
                        System.out.println("  PID: " + notification.getPID());
                    }
                }
                // 暂停一小段时间，避免过度占用CPU
                Thread.sleep(1000); 
            }

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