PulseMQ

面向金融行情与实时数据流的高性能消息中间件，基于 ZeroMQ ROUTER/DEALER 架构。

• Client/Server 架构 — 服务端双 ROUTER（数据面 + 控制面）集中路由；客户端 DEALER 发布/订阅
• PLAIN + bcrypt 认证 — ZAP 认证链：bcrypt 哈希凭据存储，随机 admin token 保护监控接口
• 类型保真 — DataFrame / dict / str / bytes 端到端保真（_restore_type 自动还原原始 Python 类型）
• 低延迟数据面 — 独立同步线程转发，端到端 p50 可低至亚毫秒级
• 完整监控 — Web UI（ECharts）+ REST API + SSE 实时推送，流量趋势、延迟分位（P50/P95/P99）、丢弃统计、在线客户端、事件流
• 流控与丢弃监控 — 信用窗口流控 + DONTWAIT 非阻塞发送，消费端有界队列满时按 topic 统计丢弃
• 运行期重连 — 断线自动指数退避重连（1s → 2s → 4s … → 30s 封顶），订阅自动恢复

安装

Python >= 3.13

pip install pulse-mq

依赖项：pyzmq、msgspec、python-snappy、lz4、zstandard、pyarrow、pandas、bcrypt、loguru。

import pulsemq            # Python 模块名（无连字符）
from pulsemq import Server

PyPI 分发名是 pulse-mq（pip install 用），import 名是 pulsemq（Python 标识符不允许连字符）。与 python-dateutil → import dateutil 同一模式。

快速开始

启动服务端

最简单的方式是直接用 CLI（零配置，首次启动自动生成默认 admin 用户，密码输出到 stderr）：

pulsemq          # 或： pulsemq-server

也可以在代码中启动（start() 是协程，需要 asyncio.run 包裹）：

import asyncio
from pulsemq import Server

async def main():
    srv = Server(
        data_endpoint="tcp://0.0.0.0:5555",
        control_endpoint="tcp://0.0.0.0:5556",
        admin_endpoint="0.0.0.0:9090",
        credentials={"user1": "pass1"},  # 或省略，用 credentials_file
    )
    await srv.start()
    await srv.wait_for_shutdown()  # Ctrl+C / srv.stop() 后返回

asyncio.run(main())

Server 构造参数：

用户管理是独立 CLI（pulsemq-users，不连 Server，直接读写凭据文件）：

pulsemq-users add user1 --password pass1 --roles publisher,subscriber
pulsemq-users list

服务端内置定时推送

无需外部生产者客户端，直接在 Server 上注册定时回调：

import asyncio
from pulsemq import Server

async def main():
    srv = Server(credentials={"u": "p"})

    @srv.producer("market.tick", interval=2.0, serializer="msgpack")
    async def gen_tick():
        return {"symbol": "AAPL", "price": 180.5, "volume": 1000}

    @srv.producer("market.quote", interval=0.5, serializer="pyarrow", compression="lz4")
    async def gen_quote():
        import pandas as pd
        return pd.DataFrame({"price": [10, 20], "vol": [100, 200]})

    # burst_producer：无间隔连续推送，回调返回 None 即停止本 producer
    seq = 0
    @srv.burst_producer("bench", serializer="msgpack")
    async def bench():
        nonlocal seq
        if seq >= 1000:   # 发满 1000 条后停止
            return None
        seq += 1
        return {"seq": seq}

    await srv.start()       # 注册的 producer 自动开始调度
    await srv.wait_for_shutdown()

asyncio.run(main())

回调返回值支持 DataFrame / dict / str / bytes，自动编码为协议帧并路由到所有匹配的消费者。

生产者

先创建用户（一次性），再运行示例（Server 必须已在运行）：

pulsemq-users add publisher --password pass1 --roles publisher
pulsemq-users add subscriber --password pass2 --roles subscriber

import asyncio
from pulsemq.client import ProducerClient

async def main():
    prod = ProducerClient(
        "tcp://127.0.0.1:5555", "tcp://127.0.0.1:5556",
        username="publisher", password="pass1",
    )
    await prod.start()
    await prod.publish("market.stock.AAPL", {"price": 180.5, "volume": 1000})
    await prod.stop()

asyncio.run(main())

消费者

import asyncio
from pulsemq.client import ConsumerClient

async def main():
    cons = ConsumerClient(
        "tcp://127.0.0.1:5555", "tcp://127.0.0.1:5556",
        username="subscriber", password="pass2",
    )
    await cons.start()

    def on_msg(msg):
        print(msg.topic, msg.payload, msg.timestamp_ns)

    await cons.subscribe("market.*", on_msg)  # 可在 start 前预注册
    await cons.run_forever()  # 运行直到 Ctrl+C，重连致命错误自动抛出

asyncio.run(main())

subscribe 支持 header_only=True——回调只接收 FrameHeader（topic / record_count / timestamp_ns），跳过完整反序列化，适合只需头部信息的低延迟场景：

await cons.subscribe("market.*", on_header, header_only=True)

ProducerClient / ConsumerClient 分别屏蔽订阅 / 发布能力；通用 Client 同时支持两者。

架构速览

                      ┌──────────────────────────┐
                      │         Server           │
   生产者 DEALER ────→ │  ┌─ 数据面 ────────────┐ │ ────→ 消费者 DEALER
                      │  │ ROUTER :5555 (同步线程)│ │
                      │  └──────────────────────┘ │
                      │  ┌─ 控制面 ────────────┐ │ ←──── DEALER（REGISTER/HEARTBEAT/…）
                      │  │ ROUTER :5556 (异步) │ │
                      │  └──────────────────────┘ │
                      │  ┌─ Admin ─────────────┐ │ ──── REST / SSE / Web UI
                      │  │ HTTP :9090 (独立线程)│ │
                      │  └──────────────────────┘ │
                      │     ZAP PLAIN (bcrypt)    │
                      └──────────────────────────┘

数据流

生产者 DEALER ──encode→  数据面 ROUTER  decode_header  match topic ──→ 消费者 DEALER
                              ↓
                    TrafficStats.record  LatencyStats.sample（半程）

服务端不解压、不反序列化 payload（decode_header 仅提取头部），转发后由消费者完整 decode 还原。

控制面命令

详细设计见 docs/ARCHITECTURE.md。

数据类型与序列化

支持的数据类型

encode 会自动推断 data_type 并选择默认序列化器，无需手动指定。DataFrame 在 msgpack/json 下先转 list[dict]（data_type=DATAFRAME），接收端 _restore_type 还原回 DataFrame；pyarrow 直接序列化 IPC 流。

序列化格式

压缩算法

小消息场景压缩通常是负收益（计算开销 > 传输节省）；批量 DataFrame 场景 lz4/zstd 有明显效果。

客户端生命周期

首次启动

启动认证检测采用 ZMQ monitor 设计：

start()
  ├─ 数据面 DEALER + PLAIN + monitor → 等待认证裁定（超时 5s）
  │   ├─ handshake_ok → 继续
  │   ├─ auth_failed → AuthenticationError（exit 3）
  │   └─ 超时 / 其他 → ClientStartupError（exit 4）
  ├─ 控制面 DEALER（复用数据面认证态，不开 monitor）
  ├─ REGISTER → OK / 超时（exit 4） / 被拒（exit 4）
  ├─ 恢复既有订阅（重连场景）
  ├─ recv_loop + heartbeat_loop（每秒）
  └─ 切换到运行期 monitor（接管断线重连）

运行期重连

断线后按指数退避自动重连（初始 1s，×2，封顶 30s）：

disconnected → cancel bg tasks → 新 Transport → PLAIN 认证 → REGISTER（同 client_id）
  ├─ ALREADY_ONLINE → 退避重试（等心跳超时释放旧记录）
  ├─ auth_failed → 致命错误，exit 3
  └─ OK → 恢复订阅 → 重启 recv/heartbeat

业务无感：订阅自动恢复，消息继续接收，业务层无需重新 subscribe()。

两线程消费模型（可选）

默认单线程：recv 线程同时负责解码与回调。当解码成为瓶颈时，传入 decode_queue_size > 0 启用两线程模式：

recv 线程：header 解码 + 延迟采样 + 路由匹配 → 入队 _DropQueue
worker 线程：批量出队 → 完整 decode + 回调分发

队列满时丢弃最老消息并按 topic 计数，通过心跳上报给服务端，在 Web UI 的 topic 卡片上可见。

监控

Web UI

浏览器打开 http://localhost:9090/?token=<admin_token> 查看实时面板：

• 指标卡片：活跃主题、消息量/秒、流量/秒、运行时间
• 客户端卡片：在线用户、生产者数、消费者数、订阅数
• ECharts 流量趋势折线图（分钟级，1H/8H 切换，最多 5 topic 叠加）+ msg/s、bytes/s sparkline
• 延迟趋势曲线（P50/P95/P99 时间序列，半程/全程可切换）+ 端到端延迟列表
• 实时事件流（认证 / 连接 / 断线 / 订阅）
• topic 卡片含丢弃指示；在线 Client 详情

REST API

# 实时指标（topics / latency / drops / 在线计数 / 最近事件）
curl 'http://localhost:9090/api/v1/stats/realtime?token=<token>'

# 主题列表
curl 'http://localhost:9090/api/v1/topics?token=<token>'

# 主题分钟级历史（内存 + SQLite 合并）
curl 'http://localhost:9090/api/v1/topics/market.tick/history?minutes=60&token=<token>'

# 延迟历史（kind=half 半程 / e2e 全程）
curl 'http://localhost:9090/api/v1/latency/topics/market.tick/history?minutes=60&kind=e2e&token=<token>'

# 在线客户端明细
curl 'http://localhost:9090/api/v1/clients?token=<token>'

# 生命周期事件
curl 'http://localhost:9090/api/v1/events?limit=50&token=<token>'

# 系统状态
curl 'http://localhost:9090/api/v1/system/status?token=<token>'

# 健康检查（无需 token）
curl http://localhost:9090/healthz

token 通过 ?token=... 或 Authorization: Bearer ... 传递。

SSE 实时流

curl -N 'http://localhost:9090/api/v1/stats/stream?token=<token>'

每 1 秒推送一帧 JSON，包含 topics / latency_half / latency_e2e / drops / online_users / sse_events / server_time / start_time 等。

协议帧格式

单 bytes 帧（非 ZMQ 多帧，通过 DEALER/ROUTER 传输）：

magic(2) ver(1) msg_type(1) flags(1) data_type(1) topic_len(2 BE)
topic(N) ts(8 BE ns) record_count(4 BE) payload(变长) [CRC32?(4)]

• magic = "PM"，ver = 0x01
• msg_type = DATA(0x01) / CONTROL(0x02)
• flags 位域：序列化器（bit 0-2）+ 压缩算法（bit 3-4）+ CRC（bit 7）+ reserved（bit 5-6）
• data_type = UNKNOWN(0x00) / DICT(0x01) / DATAFRAME(0x02) / STR(0x03) / BYTES(0x04)
• record_count 上限 1,000,000
• CRC 可选（由 flags bit 7 指示，默认关闭）

位级布局与 flags 编码细节见 docs/ARCHITECTURE.md。

凭据管理

CLI

用户管理是独立入口 pulsemq-users（不连 Server，直接读写凭据文件，密码自动 bcrypt 哈希）：

# 添加用户（密码自动 bcrypt 哈希；--password 可省略，改为交互输入）
pulsemq-users add trader1 --password secret123 --roles publisher,subscriber

# 列出所有用户
pulsemq-users list

# 禁用 / 启用用户
pulsemq-users disable trader1
pulsemq-users enable  trader1

# 修改密码
pulsemq-users passwd trader1 --password new_secret

# 热加载凭据（向 Server 进程发 SIGHUP；需 PULSEMQ_PID 环境变量，仅 POSIX）
pulsemq-users reload

启动服务端用另一个入口 pulsemq（等价于 pulsemq-server）：

pulsemq            # 启动 Server，Ctrl+C 优雅关闭

指定凭据文件：把 --file <path> 放在子命令前后均可（如 pulsemq-users list --file /etc/pulsemq/users.toml）。

文件格式（pulsemq_users.toml）

[users.admin]
hashed_password = "$2b$12$..."
roles = ["admin"]
enabled = true
created_at = "2026-06-27T00:00:00Z"

文件采用原子写（临时文件 + rename），用户名/角色做准入校验（仅字母数字 _ -），阻断危险字符流入 TOML 防注入。

Admin Token

优先级：显式 admin_token= 参数 > 配置文件 > 环境变量 PULSEMQ_ADMIN_TOKEN > 随机生成。

首次启动自动生成 32 字节随机 base64url token，写入 ./data/pulsemq_admin.token（POSIX 下 0600 权限），并在 stderr 输出一次。Web UI 和 REST API 通过 ?token=... 或 Authorization: Bearer ... 传递。

日志

日志输出到 data/logs/ 目录，每日滚动，保留 30 天：

data/logs/
├── pulsemq_2026-08-01.log
├── pulsemq_2026-08-02.log
└── ...

stderr 同步输出（容器/交互可见）。

性能基准

仓库自带多套基准脚本：

# 1. 单进程快速基准（Server + Producer + Consumer 同进程）
python scripts/bench_simple.py --duration 5
python scripts/bench_simple.py --duration 5 --records-per-frame 1000 --serializer pyarrow --compression lz4
#   可选：--serializer {msgpack,json,pyarrow,str,bytes}
#         --compression {none,snappy,lz4,zstd}

# 2. 全面基准（协议层微基准 + 端到端矩阵 + 扇出，单进程）
python scripts/bench_full.py                  # 跑全部
python scripts/bench_full.py --duration 10 --part 2   # 只跑 Part 2（端到端矩阵）

# 3. 多进程基准（生产端/服务端/消费端独立进程，28 组合全覆盖）
python scripts/bench_multiprocess.py          # 跑全部 28 组合
python scripts/bench_multiprocess.py --count 3000 --data-type dict

# 4. 跨机器基准（远程 Linux 跑 Server，本地跑 Producer/Consumer）
python scripts/bench_dist.py --remote <ip> --ssh root@<ip>
python scripts/bench_dist.py --part a         # 只跑 Part A

脚本输出帧/记录吞吐量与 p50/p90/p99/max 帧延迟。

性能特征（可从实现推断，实际数值请自行运行脚本获取）：

• DataFrame + msgpack/json：encode 限速、无积压，端到端延迟最低（p50 常在亚毫秒级）
• 小消息（dict/str/bytes）：吞吐最高（可达 10 万+ f/s），但 burst 发送易致队列积压，p50 较高
• pyarrow：encode 快，但 consumer 端 to_pandas() 转换是延迟瓶颈
• zstd：对大 payload 压缩比最优，对 < 200B 小消息通常为负收益
• 数据面同步线程：转发路径无 asyncio 调度，路由匹配有结果缓存

配置

环境变量

配置文件（TOML）

更多参数（stats_db、event_ring_size、stats_archive_batch_size、admin_thread、ui_enabled、decode_queue_size 等）通过 TOML 配置文件设置。在 Server(config=...) 传入自定义 ServerConfig，或调用 load_server_config(path) / load_client_config(path) 加载。完整字段见 ServerConfig / ClientConfig。

ServerConfig 默认值：零配置即可启动。

许可证

MIT