Postgres - Listen/Notify构建轻量级发布订阅系统
<h2 id="概述">概述</h2><p>原先设计一个内部系统的消息模块和缓存模块时，只有一个Postgres依赖。想着没多大用户量，没必要额外安装Redis，徒增运维工夫。缓存好解决，配个UNLOGGED表即可。吭吭哧哧琢磨怎么用数据表实现消息的时候，发现PostgreSQL 提供了内置命令 <code>LISTEN</code> 和 <code>NOTIFY</code>，用于在数据库服务器和连接的客户端之间实现异步通信。这个 PostgreSQL 特有的扩展功能使得数据库可以作为一个轻量级的消息队列（MQ）系统使用，允许应用程序从数据库中生成事件，并由其他客户端实时响应。于是一拍即合，上手体验一下。</p>
<h3 id="核心特性">核心特性</h3>
<ul>
<li><strong>轻量级实现</strong>：无需额外的消息中间件，直接利用 PostgreSQL 内置功能</li>
<li><strong>异步通信</strong>：支持发布-订阅模式，实现解耦的组件通信</li>
<li><strong>内存高效</strong>：通道（Channel）是纯内存对象，不占用磁盘空间</li>
<li><strong>零配置</strong>：无需预先创建或管理通道，随用随建</li>
</ul>
<h3 id="适用场景">适用场景</h3>
<ol>
<li><strong>实时仪表盘</strong>：数据变更时实时推送更新</li>
<li><strong>缓存失效</strong>：数据更新时通知缓存层刷新</li>
<li><strong>数据审计</strong>：跟踪重要数据变更事件</li>
<li><strong>任务调度</strong>：构建简单的分布式任务队列</li>
<li><strong>事件驱动架构</strong>：实现微服务间的事件通信</li>
</ol>
<h3 id="通道channel机制">通道（Channel）机制</h3>
<p><strong>重要特性</strong>：</p>
<ul>
<li>Channel 是纯内存对象，随 <code>LISTEN</code> 命令隐式创建</li>
<li>当所有监听会话断开或执行 <code>UNLISTEN</code> 时自动回收</li>
<li>无需手动创建或删除通道，也不支持此操作</li>
</ul>
<h3 id="消息传递模型">消息传递模型</h3>
<p>PostgreSQL 的 <code>NOTIFY</code> 采用典型的 <strong>"无监听即丢弃"</strong> 机制：</p>
<ul>
<li>没有监听者时，消息不会入队</li>
<li>不占用磁盘空间</li>
<li>不消耗持久化内存</li>
<li>消息仅在存在活跃监听者时传递</li>
</ul>
<h2 id="基础使用">基础使用</h2>
<h3 id="psql-命令行示例">psql 命令行示例</h3>
<pre><code class="language-sql">-- 监听指定通道
LISTEN task_channel;
-- 向通道发送消息
NOTIFY task_channel, '123456';
-- 取消监听所有通道
UNLISTEN *;
-- 查看当前监听的通道
SELECT pg_listening_channels();
-- 查看系统通知状态
SELECT * FROM pg_stat_activity WHERE backend_type = 'client backend';
</code></pre>
<h3 id="动态消息生成">动态消息生成</h3>
<p>标准的 <code>NOTIFY</code> 命令要求消息内容必须明确指定，不支持动态字符串拼接。但可以使用 <code>pg_notify()</code> 函数来生成动态通知：</p>
<pre><code class="language-sql">-- 使用 pg_notify 函数支持动态消息
SELECT pg_notify('my_channel', 'Hello, ' || 'World!');
-- 带参数的动态消息
SELECT pg_notify('audit_channel', 'User ' || current_user || ' logged in at ' || now()::text);
</code></pre>
<h2 id="python-实现示例">Python 实现示例</h2>
<h3 id="项目结构">项目结构</h3>
<pre><code>├── main.py # 核心实现：TaskWorker 和 TaskProducer
├── conf/
│ └── config.toml # 配置文件
└── pkg/
└── config/ # 配置管理模块
</code></pre>
<h3 id="安装依赖">安装依赖</h3>
<pre><code class="language-shell">uv add "psycopg>=3.3.3"
</code></pre>
<h3 id="配置管理">配置管理</h3>
<p>首先，通过配置文件管理数据库连接和通道设置：</p>
<pre><code class="language-toml"># conf/config.toml
host = "127.0.0.1"
port = 5432
user = "username"
password = "password"
dbname = "database_name"
pool_min_size = 2
pool_max_size = 10
channel = "task_channel"# 默认通道名称
</code></pre>
<p>配置模块的代码示例：<code>pkg/config/config.py</code></p>
<pre><code class="language-python">import tomllib
from typing import Any, Dict
class BaseConfig:
def __init__(self, cfg_file: str):
self._cfg_file = cfg_file
self._data: Dict = {}
self._load_config()
def _load_config(self) -> None:
if self._data:
return
try:
with open(self._cfg_file, "rb") as f:
self._data = tomllib.load(f)
except FileNotFoundError:
raise RuntimeError(f"配置文件不存在: {self._cfg_file}")
except Exception as e:
raise RuntimeError(f"加载配置文件失败: {e}")
class PostgresConfigMixin(BaseConfig):
def postgres_host(self) -> str:
return self._data.get("database", {}).get("postgres", {}).get("host", "")
def postgres_port(self) -> int:
return self._data.get("database", {}).get("postgres", {}).get("port", 5432)
def postgres_user(self) -> str:
return self._data.get("database", {}).get("postgres", {}).get("user", "")
def postgres_password(self) -> str:
return self._data.get("database", {}).get("postgres", {}).get("password", "")
def postgres_dbname(self) -> str:
return self._data.get("database", {}).get("postgres", {}).get("dbname", "")
def postgres_pool_min_size(self) -> int:
return (
self._data.get("database", {}).get("postgres", {}).get("pool_min_size", 2)
)
def postgres_pool_max_size(self) -> int:
return (
self._data.get("database", {}).get("postgres", {}).get("pool_max_size", 10)
)
def postgres_channel(self) -> str:
return (
self._data.get("database", {})
.get("postgres", {})
.get("channel", "default_channel")
)
def get_postgres_dsn(self, hide_password: bool = False) -> str:
"""获取PostgreSQL连接DSN"""
password = self.postgres_password()
if hide_password and password:
password = "***"
# psycopg3 使用标准的 PostgreSQL 连接字符串格式
return (
f"postgresql://{self.postgres_user()}:{password}@"
f"{self.postgres_host()}:{self.postgres_port()}/{self.postgres_dbname()}"
)
class LLMConfigMixin(BaseConfig):
"""LLM配置Mixin"""
def llm_model(self) -> str:
return self._data.get("llm", {}).get("model", "")
def llm_base_url(self) -> str:
return self._data.get("llm", {}).get("base_url", "")
def llm_api_key(self) -> str:
return self._data.get("llm", {}).get("api_key", "")
class Config(PostgresConfigMixin, LLMConfigMixin):
def __init__(self, cfg_file: str = "conf/config.toml"):
super().__init__(cfg_file)
def reload(self) -> None:
"""重新加载配置"""
self._data = {}# 清空数据
self._load_config()
</code></pre>
<h3 id="核心组件实现">核心组件实现</h3>
<h4 id="1-任务消费者taskworker">1. 任务消费者（TaskWorker）</h4>
<p><code>TaskWorker</code> 负责监听指定通道并处理接收到的任务：</p>
<pre><code class="language-python"># main.py - TaskWorker 类核心部分
import asyncio
import json
import signal
from typing import Set
from psycopg import AsyncConnection, Notify, sql
from psycopg_pool import AsyncConnectionPool
from pkg.config import cfg
MAX_CONCURRENCY = 10
class TaskWorker:
def __init__(self, dsn: str, channel: str):
self._dsn = dsn
self.channel = channel
self.pool: AsyncConnectionPool | None = None
self.listener_conn: AsyncConnection | None = None
self.sem = asyncio.Semaphore(MAX_CONCURRENCY)
self.active_tasks: Set = set()
async def start(self) -> None:
self.pool = AsyncConnectionPool(
self._dsn,
min_size=cfg.postgres_pool_min_size(),
max_size=cfg.postgres_pool_max_size(),
open=False,# 延迟打开, 避免阻塞
)
await self.pool.open()
# 独立监听连接，防止 LISTEN 状态随连接回收丢失
self.listener_conn = await AsyncConnection.connect(self._dsn, autocommit=True)
await self.listener_conn.execute(
sql.SQL("LISTEN {}").format(sql.Identifier(self.channel))
)
print(f"Listening on channel: {self.channel}")
try:
async for notify in self.listener_conn.notifies():
if notify.channel == self.channel and notify.payload:
await self._dispatch_task(notify)
except asyncio.CancelledError:
print("Listener cancelled")
except Exception as e:
print(f"Listener error: {e}")
finally:
await self.stop()
async def _dispatch_task(self, notify: Notify) -> None:
"""接收通知并分发任务"""
task_info = notify.payload.strip()
try:
task_data = json.loads(task_info)# 假设 payload 是 JSON 格式的字符串
print(f"Received task notification: {task_data}")
except json.JSONDecodeError:
task_data = {"task_id": task_info}# 如果不是 JSON 格式，使用原始字符串
print(f"Received non-JSON task notification: {task_data}")
except Exception as e:
print(f"Invalid task ID received: {task_info}")
return
if not isinstance(task_data, dict) or "task_id" not in task_data:
print(
f"Missing task_id in notification: {task_data}, or task_data is not a dict"
)
return
task = asyncio.create_task(self._process_task(task_data))
self.active_tasks.add(task)
task.add_done_callback(self.active_tasks.discard)
async def _process_task(self, task_data: dict) -> None:
"""执行业务逻辑"""
if not self.pool:
raise RuntimeError("Connection pool is not initialized")
async with self.sem:
async with self.pool.connection() as conn:
try:
await self._execute_business(task_data)
print(f"<= Task {task_data['task_id']} completed successfully")
except Exception as e:
print(f"Error processing task {task_data['task_id']}: {e}")
# 这里可以添加重试逻辑或错误记录
await self._log_failure(conn, task_data["task_id"], str(e))
async def _execute_business(self, task_data: dict) -> None:
"""执行业务逻辑"""
print(f"<= Processing task {task_data}...")
await asyncio.sleep(5)# 模拟耗时操作
print(f"<= Task {task_data} done.")
async def _log_failure(self, conn: AsyncConnection, task_id: int, error_msg: str):
"""记录失败日志"""
try:
# 记录失败日志到独立表，便于后续重试或告警
print(f"Logging failure for task {task_id}: {error_msg}")
except Exception as e:
print(f"Failed to log error for task {task_id}: {e}")
async def stop(self) -> None:
print("Stopping TaskWorker...")
if self.active_tasks:
await asyncio.gather(*self.active_tasks, return_exceptions=True)
if self.listener_conn:
await self.listener_conn.close()
if self.pool:
await self.pool.close()
print("TaskWorker stopped gracefully.")
</code></pre>
<h4 id="2-任务发布者taskpublisher">2. 任务发布者（TaskPublisher）</h4>
<p><code>TaskPublisher</code> 负责向通道发布任务消息：</p>
<pre><code class="language-python"># main.py - TaskPublisher 类
class TaskPublisher:
def __init__(self, dsn: str):
self._dsn = dsn
self._pool: AsyncConnectionPool | None = None
async def start(self):
if not self._pool:
self._pool = AsyncConnectionPool(
self._dsn,
min_size=cfg.postgres_pool_min_size(),
max_size=cfg.postgres_pool_max_size(),
open=False,# 延迟打开, 避免阻塞
)
await self._pool.open()
print("TaskPublisher started.")
async def publish(self, channel: str, payload: dict):
if not self._pool:
raise RuntimeError("Connection pool is not initialized")
async with self._pool.connection() as conn:
try:
payload_str = json.dumps(
payload, default=str
)# 将 dict 转换为 JSON 字符串
await conn.execute(
sql.SQL("NOTIFY {}, {}").format(
sql.Identifier(channel), sql.Literal(payload_str)
)
)
print(f"=> Published task to channel {channel}: {payload_str}")
return True
except Exception as e:
print(f"Failed to publish task to channel {channel}: {e}")
return False
async def publish_batch(self, channel: str, payloads: list):
if not self._pool:
raise RuntimeError("Connection pool is not initialized")
count = 0
async with self._pool.connection() as conn:
for payload in payloads:
try:
payload_str = json.dumps(payload, default=str)
await conn.execute(
sql.SQL("NOTIFY {}, {}").format(
sql.Identifier(channel), sql.Literal(payload_str)
)
)
print(f"=> Published task to channel {channel}: {payload_str}")
count += 1
except Exception as e:
print(f"Failed to publish batch tasks to channel {channel}: {e}")
return count
async def stop(self):
if self._pool:
await self._pool.close()
print("TaskPublisher stopped.")
async def __aenter__(self):
await self.start()
return self
async def __aexit__(self, exc_type, exc_val, exc_tb):
await self.stop()
</code></pre>
<h3 id="调用演示">调用演示</h3>
<p>以下是<code>main.py</code>中的演示部分：</p>
<pre><code class="language-python"># main.py
async def run_worker():
worker = TaskWorker(cfg.get_postgres_dsn(), cfg.postgres_channel())
task = asyncio.create_task(worker.start())
return task
async def run_publisher():
async with TaskPublisher(cfg.get_postgres_dsn()) as publisher:
for i in range(1, 11):
payload = {"task_id": i, "data": f"Task data {i}"}
await publisher.publish(cfg.postgres_channel(), payload)
await asyncio.sleep(0.5)# 模拟发布间隔
async def main():
# worker = TaskWorker(cfg.get_postgres_dsn(), cfg.postgres_channel())
# loop = asyncio.get_running_loop()
# stop_evt = asyncio.Event()
# for sig in (signal.SIGINT, signal.SIGTERM):
# loop.add_signal_handler(sig, stop_evt.set)
# listen_task = asyncio.create_task(worker.start())
# await stop_evt.wait()
# print("Shutdown signal received, stopping worker...")
# listen_task.cancel()
# await listen_task
worker_task = await run_worker()
await asyncio.sleep(2)# 确保 worker 已经启动并监听
await run_publisher()# 发布任务
print("All tasks published successfully.")
await asyncio.sleep(30)# 等待 worker 处理完任务
worker_task.cancel()# 停止 worker
try:
await worker_task
except asyncio.CancelledError:
pass
if __name__ == "__main__":
asyncio.run(main())
</code></pre>
<h2 id="使用场景扩展">使用场景扩展</h2>
<h3 id="场景一实时数据同步">场景一：实时数据同步</h3>
<pre><code class="language-python">async def sync_data_change(self, table_name: str, record_id: str, operation: str):
"""数据变更时发送同步通知"""
message = f"{table_name}:{record_id}:{operation}"
await self.publish("data_sync_channel", message)
</code></pre>
<h3 id="场景二分布式锁通知">场景二：分布式锁通知</h3>
<pre><code class="language-python">async def notify_lock_release(self, lock_name: str):
"""锁释放时通知等待者"""
await self.publish("distributed_lock_channel", f"RELEASE:{lock_name}")
</code></pre>
<h3 id="场景三缓存失效广播">场景三：缓存失效广播</h3>
<pre><code class="language-python">async def invalidate_cache(self, cache_key: str):
"""缓存失效时广播通知"""
await self.publish("cache_invalidation_channel", cache_key)
</code></pre>
<h2 id="注意事项">注意事项</h2>
<h3 id="技术限制">技术限制</h3>
<ol>
<li><strong>消息大小</strong>：NOTIFY 消息最大为 8000 字节</li>
<li><strong>无持久化</strong>：消息不持久化，重启后丢失</li>
<li><strong>无确认机制</strong>：发送方无法知道消息是否被接收</li>
<li><strong>无顺序保证</strong>：消息可能不按发送顺序到达</li>
</ol>
<h3 id="生产环境建议">生产环境建议</h3>
<ol>
<li><strong>监控告警</strong>：实现通道监听状态监控</li>
<li><strong>错误处理</strong>：添加完善的错误处理和日志记录</li>
<li><strong>备份机制</strong>：重要消息应有备份存储</li>
<li><strong>性能测试</strong>：在高负载下测试系统表现</li>
</ol>
<h2 id="补充">补充</h2>
<h3 id="asyncpg版">asyncpg版</h3>
<p><code>asyncpg</code>是python连接postgres的纯异步驱动，性能更好</p>
<p>安装</p>
<pre><code class="language-shell">uv add asyncpg
</code></pre>
<p>代码示例：</p>
<pre><code class="language-python">import asyncio
import signal
from typing import Set
import asyncpg
from pkg.config import cfg
MAX_CONCURRENT_TASKS = 5# 最大并发任务数
class TaskWorker:
def __init__(self, dsn: str, channel: str):
self._dsn = dsn
self._channel = channel
self._pool: asyncpg.Pool | None = None
self._listener_conn: asyncpg.Connection | None = None
self._sem = asyncio.Semaphore(MAX_CONCURRENT_TASKS)
self._active_tasks: Set = set()
async def start(self):
"""启动监听器"""
if not self._pool:
self._pool = await asyncpg.create_pool(
dsn=self._dsn,
min_size=cfg.postgres_pool_min_size(),
max_size=cfg.postgres_pool_max_size(),
)
# 创建专用连接用于监听
if not self._listener_conn:
self._listener_conn = await asyncpg.connect(dsn=self._dsn)
await self._listener_conn.add_listener(self._channel, self._on_notify)
print(f"Listening on channel: {self._channel}")
try:
# 保持监听状态
await asyncio.Future()
except asyncio.CancelledError:
pass
finally:
await self.stop()
async def _on_notify(
self, conn: asyncpg.Connection, pid: int, channel: str, payload: str
):
"""收到 NOTIFY 时的回调函数"""
if not payload:
return
task_id = payload.strip()
print(f"Received notification: {task_id} on channel: {channel}")
# 提交到事件循环，带并发限制
task = asyncio.create_task(self._handle_task(task_id))
self._active_tasks.add(task)
task.add_done_callback(self._active_tasks.discard)
async def _handle_task(self, task_id: str):
"""任务处理核心逻辑"""
if not self._pool:
raise RuntimeError("数据库连接池未初始化")
async with self._sem:# 并发控制
async with self._pool.acquire() as conn:
try:
# 1. 执行业务逻辑
await self._execute_business_logic(task_id)
# 2. 更新任务状态（示例）
print(f"Task {task_id} completed successfully.")
except Exception as e:
print(f"Task {task_id} failed: {e}")
await self._log_failure(task_id, str(e))
async def _execute_business_logic(self, task_id: str):
"""模拟业务逻辑处理"""
print(f"Processing task {task_id}...")
await asyncio.sleep(5)# 模拟耗时操作
print(f"Task {task_id} completed.")
async def _log_failure(self, task_id: str, error: str):
"""记录失败日志"""
print(f"Task {task_id} failure logged: {error}")
# 实际项目中可记录到数据库
async def stop(self):
"""优雅关闭"""
print("Shutting down gracefully...")
# 等待正在运行的任务完成
if self._active_tasks:
await asyncio.gather(*self._active_tasks, return_exceptions=True)
# 清理资源
if self._listener_conn:
await self._listener_conn.remove_listener(self._channel, self._on_notify)
await self._listener_conn.close()
if self._pool:
await self._pool.close()
print("Shutdown complete.")
async def main():
worker = TaskWorker(cfg.get_postgres_dsn(), cfg.postgres_channel())
loop = asyncio.get_running_loop()
stop_evt = asyncio.Event()
# 注册信号处理器，优雅关闭
for sig in (signal.SIGINT, signal.SIGTERM):
loop.add_signal_handler(sig, stop_evt.set)
# 启动监听器
listen_task = asyncio.create_task(worker.start())
# 等待停止事件
await stop_evt.wait()
print("Shutdown signal received, stopping worker...")
# 取消监听任务并等待完成
listen_task.cancel()
await listen_task
if __name__ == "__main__":
asyncio.run(main())
</code></pre>
</div>
<div id="MySignature" role="contentinfo">
<p>本文来自博客园，作者：花酒锄作田，转载请注明原文链接：https://www.cnblogs.com/XY-Heruo/p/19832406</p><br><br>
来源：https://www.cnblogs.com/XY-Heruo/p/19832406
頁:
[1]