Postgres LISTEN/NOTIFY: pub/sub real-time không cần Redis hay Kafka

Có một pattern tôi gặp đi gặp lại ở các team Việt Nam: hệ thống mới chỉ vài nghìn user, một API server và một Postgres, nhưng team đã vội kéo Redis vào chỉ để làm pub/sub cho cache invalidation. Thêm một dependency nghĩa là thêm một thứ phải monitor, phải backup, phải debug khi nó hết RAM lúc 2 giờ sáng. Trong khi đó Postgres có sẵn một cơ chế pub/sub built-in mà ít người dùng đến: LISTEN/NOTIFY.

Hôm trước mình refactor một service cũ — cái đó dùng polling DB mỗi 2 giây để biết khi có config thay đổi, một thiết kế đáng xấu hổ. Đổi sang LISTEN/NOTIFY mất đúng 30 phút và xóa được nguyên một cron job. Bài này kể lại cái mình học được: nó hoạt động thế nào, code thực tế ra sao, và quan trọng nhất — chỗ nào nó sẽ vỡ.

LISTEN/NOTIFY là gì

Đây là cặp lệnh SQL có trong Postgres từ rất lâu (trước version 9):

• NOTIFY channel, 'payload' — publish một message vào channel
• LISTEN channel — đăng ký nhận message của channel đó
• UNLISTEN channel — hủy đăng ký

Connection nào đang LISTEN sẽ nhận được payload mỗi khi có ai đó NOTIFY cùng channel. Message được gửi qua async notification — driver Postgres (psycopg, pg, asyncpg, pq cho Go…) đều support đọc nó.

-- Terminal 1
LISTEN cache_invalidate;
-- chờ...

-- Terminal 2
NOTIFY cache_invalidate, 'user:123';

Terminal 1 sẽ nhận được message với channel cache_invalidate, payload user:123. Đơn giản đến mức gần như khó tin.

Trigger NOTIFY tự động khi data thay đổi

Cái hay là bạn có thể gọi pg_notify(channel, payload) từ trigger. Tức là mỗi khi một row được insert/update/delete, Postgres tự động phát event ra ngoài.

CREATE OR REPLACE FUNCTION notify_product_change()
RETURNS trigger AS $$
BEGIN
  PERFORM pg_notify(
    'product_changes',
    json_build_object(
      'op', TG_OP,
      'id', COALESCE(NEW.id, OLD.id),
      'updated_at', extract(epoch from now())
    )::text
  );
  RETURN COALESCE(NEW, OLD);
END;
$$ LANGUAGE plpgsql;

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

Bây giờ mọi thay đổi trên bảng products đều phát một event lên channel product_changes. Cache layer (hay search indexer, hay WebSocket fan-out) chỉ cần LISTEN channel đó để biết phải invalidate cái gì.

Listener phía Python (asyncpg)

Mình dùng asyncpg vì nó hỗ trợ LISTEN gọn gàng:

import asyncio
import asyncpg
import json

async def listen_product_changes(conn):
    async def callback(connection, pid, channel, payload):
        data = json.loads(payload)
        print(f"[{channel}] {data['op']} product {data['id']}")
        # Invalidate cache, push WebSocket, trigger indexer...
        await invalidate_cache(f"product:{data['id']}")

    await conn.add_listener("product_changes", callback)
    # giữ connection sống
    while True:
        await asyncio.sleep(3600)

async def main():
    conn = await asyncpg.connect(DATABASE_URL)
    try:
        await listen_product_changes(conn)
    finally:
        await conn.close()

asyncio.run(main())

Connection này phải là dedicated connection, không đi qua PgBouncer ở mode transaction hoặc statement — sẽ giải thích ngay dưới.

Những cú đau khi đưa vào production

Cái dễ nhất là viết được demo. Cái khó là chạy 24/7 không sót event nào. Đây là những bài học mình va phải.

1. PgBouncer transaction mode KHÔNG support LISTEN

Đây là cái cắt cổ phổ biến nhất. Hầu hết hệ thống Postgres trong sản xuất đều có PgBouncer đứng trước ở mode transaction (pool connection theo transaction, không phải session). Mode đó không support session-level state như LISTEN.

# pgbouncer.ini cho phần listener
[databases]
notify_pool = host=postgres port=5432 dbname=app pool_mode=session

[pgbouncer]
# pool chính vẫn transaction mode
pool_mode = transaction

Giải pháp là tạo một pool riêng pool_mode=session chỉ dành cho listener, hoặc cho listener bypass PgBouncer kết nối trực tiếp tới Postgres. Đừng share connection với code application bình thường.

2. Payload tối đa 8000 bytes

Limit này hard-coded trong Postgres (NAMEDATALEN - 1 thực tế là khoảng 8000 bytes cho payload). Đừng nhét cả object lớn vào — chỉ gửi {id, op} rồi để consumer query lại nếu cần. Pattern này gọi là lightweight notification + heavy fetch, an toàn hơn nhiều.

3. Notification bị mất nếu listener offline

NOTIFY là fire-and-forget. Nếu listener đang restart, đang reconnect, hay vừa bị OOM — message đó bay luôn, không có queue, không có retry. Đây là điểm khác biệt to nhất với Redis Streams hay Kafka.

Pattern an toàn là kết hợp với bảng outbox:

CREATE TABLE event_outbox (
  id BIGSERIAL PRIMARY KEY,
  channel TEXT NOT NULL,
  payload JSONB NOT NULL,
  created_at TIMESTAMPTZ DEFAULT now(),
  processed_at TIMESTAMPTZ
);

CREATE OR REPLACE FUNCTION emit_event(p_channel TEXT, p_payload JSONB)
RETURNS void AS $$
BEGIN
  INSERT INTO event_outbox(channel, payload) VALUES (p_channel, p_payload);
  PERFORM pg_notify(p_channel, p_payload::text);
END;
$$ LANGUAGE plpgsql;

Listener khi nhận được notify thì xử lý và đánh dấu processed_at. Khi restart, nó query WHERE processed_at IS NULL để catch up những event đã bỏ lỡ. Đây thực ra là transactional outbox pattern thu nhỏ — đảm bảo at-least-once delivery với chính cái DB bạn đang có.

4. NOTIFY trong transaction chỉ gửi đi khi COMMIT

Đây thực ra là tính năng, không phải bug — nhưng dễ làm bạn ngạc nhiên. Nếu bạn NOTIFY rồi ROLLBACK, event không bao giờ được phát. Tốt cho consistency: subscriber không nhìn thấy data chưa committed. Nhưng nhớ rằng độ trễ của event ít nhất bằng độ dài transaction.

5. Cùng một payload trong một transaction sẽ bị dedup

Postgres tự deduplicate notification giống hệt nhau trong cùng transaction. NOTIFY foo, 'x' rồi NOTIFY foo, 'x' lần nữa chỉ sinh ra 1 notification khi commit. Nếu bạn cần đếm event chính xác, thêm UUID hay sequence number vào payload.

Khi nào đủ, khi nào không đủ

LISTEN/NOTIFY phù hợp khi:

• Throughput dưới 1000 event/giây, payload nhỏ (vài KB)
• Subscriber luôn online hoặc có outbox để catch up
• Logic đơn giản: cache invalidation, push WebSocket fan-out, đánh thức worker, broadcast config change
• Bạn đang dùng Postgres rồi và muốn giảm số component

Nó không đủ khi:

• Event throughput cao (vài chục nghìn/giây) — bottleneck ở chính NOTIFY queue trong Postgres
• Cần persistent log để replay (Kafka làm cái này)
• Cần ordering chặt chẽ giữa nhiều partition
• Subscriber có thể offline lâu (giờ, ngày) — sẽ mất event nếu không dùng outbox

Một quy tắc nhỏ: nếu một ngày nào đó bạn thấy mình implement retry, dead-letter queue, consumer group, partition trên top của LISTEN/NOTIFY — đó là tín hiệu nên đổi sang broker thật. Đừng cố thắng nó cho đủ tính năng của Kafka.

Ví dụ thật: cache invalidation cho catalog service

Mình từng dùng pattern này cho service catalog có khoảng 50 instance behind load balancer, mỗi instance giữ Caffeine cache trong memory:

[Admin update product]
        |
        v
  UPDATE products SET ... RETURNING id;
        |
        v
  Trigger -> pg_notify('product_invalidate', '123')
        |
        v
  Postgres broadcast tới 50 listeners
        |
        v
  Mỗi instance: cache.evict("product:123")

Trước đó team đang định cài thêm Redis pub/sub cho việc này. Sau khi đo: peak khoảng 200 invalidation/giây, mỗi event 20 bytes — Postgres LISTEN/NOTIFY xử lý ngon ơ, latency p99 dưới 30ms từ lúc UPDATE đến lúc cache evict ở instance xa nhất. Tiết kiệm được nguyên một Redis cluster và một mớ code để quản nó.

Tổng kết

LISTEN/NOTIFY là một trong những tính năng “ngầm” của Postgres mà ai biết thì sẽ tiết kiệm được rất nhiều. Nó không thay thế được Kafka, nhưng cho 80% use case pub/sub đời thường — cache invalidation, real-time UI update, đánh thức background worker — nó dư sức. Quan trọng nhất là bạn không phải gánh thêm một component nữa trong stack.

Lần tới khi định kéo Redis hay RabbitMQ vào chỉ để publish event đơn giản, hãy thử pg_notify() trước. Nếu nó vỡ ở scale của bạn — lúc đó hẵng nâng cấp. Còn không, đỡ được một con hạc trong chuồng.