NATS + JetStream 消息队列实战：给个人站做异步任务解耦，Subject/Stream/Consumer 与五个必踩的坑

Author： zz1984
发布时间：October 12, 2026
5 views
No comments
12633 字数
Categories：技术教程

为什么个人站长也需要消息队列

很多个人站点的架构很简单：Nginx 接请求，PHP 或 Python 处理业务，MySQL 存数据。用户提交表单、发送邮件、生成图片、写日志，全都在请求生命周期内同步完成。平时看着没问题，一旦碰上批量操作就露馅了——比如一次导入 5000 条数据、一次群发 2000 封通知邮件，请求要么超时，要么把 PHP-FPM 进程池占满，整站跟着卡住。

消息队列解决的就是这个问题：把「马上要做的重活」变成「先记下来，慢慢做」。请求只负责往队列里丢一条消息就立刻返回，后台的消费者进程再按自己的节奏处理。用户感知到的响应时间从几十秒降到几十毫秒。

说到消息队列，大家第一反应是 RabbitMQ 或 Kafka。但这两个对个人站长都有点重：Kafka 要 JVM、要 ZooKeeper（新版才好点）、内存起步就得几个 G；RabbitMQ 虽然轻一些，但 Erlang 运行时和插件体系也不小。而 NATS 是一个用 Go 写的单二进制消息系统，核心服务只有十几 MB 内存，还自带现代消息队列该有的持久化能力（JetStream）。对一台 2G 内存的小 VPS 来说，NATS 是性价比最高的选择。

这篇文章我会从零跑通一套 NATS + JetStream 的环境，讲清核心概念，然后用 PHP 发消息、Python 消费消息，最后落到「个人站有哪些地方真的用得上」。

先分清：核心 NATS 和 JetStream 是两回事

这一点必须先讲明白，否则后面全是坑。

核心 NATS（core NATS） 是一个纯粹的「发布-订阅」系统，特点是「发了就忘」：发布者把消息丢出去，如果那一刻没有订阅者在线，这条消息就直接丢了，服务器不存储、不重发。它适合做实时通知、广播、服务间心跳这类「过了就过了」的场景。

JetStream 是在核心 NATS 之上加的一层持久化层，相当于给 NATS 装上了「磁盘 + 确认 + 重放」。消息会被写到磁盘上的 Stream（流）里，消费者可以拉取（pull）、可以主动确认（ack）、可以重放历史消息。这一层才是我们做「异步任务队列」真正需要的东西。

很多教程一上来就讲 core NATS，你照着写完发现「消息丢了」，就是因为没有开 JetStream 的存储语义。记住一句话：要求可靠投递，就必须用 JetStream 的 Stream + pull consumer。

用 Docker 把 NATS 跑起来

JetStream 需要数据目录，所以启动命令里必须带 -js 开关和存储路径。写成 docker-compose 最省事：

version: "3.8"
services:
  nats:
    image: nats:2.10-alpine
    container_name: nats
    restart: unless-stopped
    # -js 开启 JetStream；--store_dir 指定持久化目录
    # -m 8222 是监控/管理 HTTP 端口，只在容器内暴露
    command: >
      -js
      --store_dir /data
      -m 8222
    ports:
      # 客户端连接端口：务必绑定内网/回环，别裸奔公网
      - "127.0.0.1:4222:4222"
      # 监控端口也只给本机
      - "127.0.0.1:8222:8222"
    volumes:
      - ./nats-data:/data
    healthcheck:
      test: ["CMD", "wget", "-qO-", "http://localhost:8222/healthz"]
      interval: 10s
      timeout: 3s
      retries: 3

几个必须注意的点：

4222 端口只绑 127.0.0.1。NATS 默认没有任何认证，绑定到 0.0.0.0 等于把消息系统送人。如果确实需要多机通信，一定要在配置文件里加用户名密码或 token。
数据目录要挂出来。JetStream 的持久化全靠这个目录，不挂卷的话容器一删消息全没。
8222 是监控端口，能看连接数、消息吞吐、JetStream 的使用量，非常好用，但同样不要暴露公网。

启动并验证：

docker compose up -d
docker logs nats --tail 20
# 看到 "Server is ready" 且 "JetStream version: 2.10.x" 就对了

# 用 nats CLI 连一下（宿主机装 nats-top 或官方 nats 客户端）
curl -s http://127.0.0.1:8222/varz | grep -o '"jetstream":{[^}]*}'

如果你不想装 CLI，直接用官方的 nats 二进制（单文件，Go 编译）：

# 下载对应架构的 nats CLI
curl -L https://github.com/nats-io/natscli/releases/latest/download/nats-2.10.18-linux-amd64.zip -o /tmp/nats.zip
unzip -o /tmp/nats.zip -d /tmp/natsbin
sudo install /tmp/natsbin/nats-*/nats /usr/local/bin/nats

# 配置连接上下文
nats context save local --server nats://127.0.0.1:4222
nats context select local
nats server check connection

核心概念：Subject、Stream、Consumer

NATS 的概念比 Kafka 少得多，三个词就能说清：

Subject（主题）

Subject 就是一个用点分隔的字符串，比如 tasks.email、tasks.image、site.log。发布者往某个 subject 发消息，订阅者订阅某个 subject 收消息。它支持通配符：tasks.* 匹配 tasks 下一级，tasks.> 匹配 tasks 下任意多级。这套设计让你可以用层级把业务分门别类，而不用像 Kafka 那样一开始就纠结「建几个 topic、分几个 partition」。

Stream（流）

Stream 是把「一批 subject 上的消息存到磁盘」的规则。创建 Stream 时要指定 subjects，比如 subjects: ["tasks.>"] 表示「所有 tasks 开头的消息都进这个流」。关键参数：

retention：保留策略。limits 按大小/条数/时间限制（默认），workqueue 表示消息被确认后立刻删除（最省空间，适合任务队列）。
max_age：消息最长保留多久，比如 24h。
storage：file（落盘，支持重启恢复）或 memory（只存内存，快但不抗重启）。做任务队列选 file。
replicas：副本数。单机固定 1。

Consumer（消费者）

Consumer 是「谁在读这个 Stream、读到哪了」的记录。两种模式：

Pull consumer（拉模式）：消费者主动说「给我 10 条」，处理完再 ack。任务队列强烈推荐这种，因为消费速度由自己控制，不会被打爆。
Push consumer（推模式）：服务器主动推给消费者。适合低延迟通知，但控制力弱。

Pull consumer 还分 ephemeral（临时，断开就忘）和 durable（持久，记住 offset）。异步任务必须用 durable，否则消费者重启后无法从上次的位置继续。

用 CLI 把 Stream 和 Consumer 建起来

先建流，再建消费者：

# 建一个 workqueue 类型的流，只保留未被确认的消息
nats stream add TASKS \
  --subjects "tasks.>" \
  --retention workqueue \
  --storage file \
  --max-age 24h \
  --max-msgs 100000 \
  --replicas 1 \
  --discard old

# 建一个持久化的拉取消费者
nats consumer add TASKS worker \
  --pull \
  --deliver all \
  --ack explicit \
  --max-deliver 5 \
  --wait 30s

# 查看状态
nats stream info TASKS
nats consumer info TASKS worker

参数逐个说清楚：

--retention workqueue：消息被 ack 后就删掉，队列不会无限膨胀。注意 workqueue 模式下一条消息只能被一个消费者消费（这正是任务队列想要的）。
--ack explicit：手动确认。只有消费成功你才 ack，处理失败就不 ack，消息会被重投。这是「不丢消息」的基石。
--max-deliver 5：同一条消息最多投 5 次，超了就进「死信」状态，避免有毒消息无限循环。
--wait 30s：拉取请求最长挂起 30 秒，没有新消息就返回空，避免客户端疯狂空轮询。

手动发一条测试消息：

nats pub tasks.email '{"to":"admin@example.com","subject":"测试"}'
# 拉取并确认
nats consumer next TASKS worker --count 1

PHP 侧：把耗时的活丢进队列

WordPress 或自研 PHP 站要发消息，官方有 basis-company/nats.php 这个纯 PHP 客户端，通过 Composer 安装：

composer require basis-company/nats

发布一条任务消息（比如「用户注册后要发的欢迎邮件」）：

<?php
require __DIR__ . '/vendor/autoload.php';

use Basis\Nats\Client;
use Basis\Nats\Configuration;

function enqueue(string $subject, array $payload): bool {
    $config = new Configuration([
        'host' => '127.0.0.1',
        'port' => 4222,
        'timeout' => 2,          // 连接超时，别让页面跟着卡
        'reconnect' => false,    // Web 请求里不重连，失败就让上层降级
    ]);
    try {
        $client = new Client($config);
        // JetStream 的发布必须走 jetstream 层，否则不会落盘
        $client->jetstream()->publish($subject, json_encode($payload, JSON_UNESCAPED_UNICODE));
        return true;
    } catch (\Throwable $e) {
        error_log('nats enqueue failed: ' . $e->getMessage());
        return false;
    }
}

// 用户注册成功后调用
$ok = enqueue('tasks.email', [
    'to' => $user['email'],
    'template' => 'welcome',
    'vars' => ['name' => $user['name']],
]);

// 关键：入队失败不能让主流程崩，写日志 + 兜底（比如塞本地文件稍后补发）
if (!$ok) {
    file_put_contents('/var/log/nats_fallback.log', json_encode($user) . "\n", FILE_APPEND);
}

两个容易踩的点：

必须用 JetStream 层的 publish（客户端上的 jetstream()->publish()），而不是 core NATS 的 publish()。后者不落盘，消费者不在线消息就没了。
Web 请求里的 NATS 客户端要短超时、不重连。如果你让请求去等重连，一次网络抖动就能把整个页面拖死。队列的意义是「快」，不是「强」。

Python 侧：消费者进程怎么写才可靠

消费者通常是一个常驻的 Python 进程，用官方的 nats-py：

pip install nats-py

一个带手动 ack、失败重投、优雅退出的 worker：

import asyncio, json, signal
import nats

class Worker:
    def __init__(self):
        self.nc = None
        self.js = None
        self.sub = None
        self.running = True

    async def connect(self):
        self.nc = await nats.connect(
            "nats://127.0.0.1:4222",
            max_reconnect_attempts=-1,      # 常驻进程应该无限重连
            reconnect_time_wait=2,
            error_cb=self.on_error,
            disconnected_cb=lambda: print("NATS disconnected"),
        )
        self.js = self.nc.jetstream()
        # durable + pull + 手动 ack 三件套
        self.sub = await self.js.pull_subscribe(
            "tasks.email", "worker", stream="TASKS"
        )

    async def on_error(self, e):
        print("NATS error:", e)

    async def run(self):
        while self.running:
            try:
                # 一次拉 5 条，挂起 30 秒
                msgs = await self.sub.fetch(batch=5, timeout=30)
            except nats.errors.TimeoutError:
                continue                      # 没消息，正常，继续等
            for msg in msgs:
                try:
                    data = json.loads(msg.data)
                    await self.handle(data)
                    await msg.ack()           # 只有成功才确认
                except Exception as e:
                    # 不 ack，让 JetStream 按 backoff 重投（受 max_deliver 限制）
                    print("task failed:", e)

    async def handle(self, data):
        # 真正的业务：发邮件、生成缩略图、重新生成 sitemap
        print("sending mail to", data["to"])
        await asyncio.sleep(0.5)              # 模拟耗时操作

    def stop(self, *_):
        self.running = False

async def main():
    w = Worker()
    await w.connect()
    # 收到 SIGTERM 时优雅退出（systemd stop 会发这个信号）
    loop = asyncio.get_event_loop()
    for sig in (signal.SIGINT, signal.SIGTERM):
        loop.add_signal_handler(sig, w.stop)
    await w.run()
    await w.nc.drain()

asyncio.run(main())

这段代码里有几个「可靠性」设计值得展开：

只有处理成功才 ack。抛异常就不 ack，消息会重新投递（默认有 backoff）。这就是「至少一次投递」。代价是业务逻辑必须幂等——同一条消息可能被处理两次，发邮件前最好查一下「这封邮件是否已发过」，或者用消息 ID 去重。
用 drain() 而不是直接断连。drain 会先把「已拉取但没确认」的消息处理完、把未发送的 ack 发出去，再关闭连接，避免重启时丢消息。
监听 SIGTERM。systemd 停止服务时发的就是 SIGTERM，处理好这个信号，systemctl restart 才不会丢任务。
常驻进程用无限重连。和 Web 请求相反，worker 断线了就应该一直重连，直到 NATS 回来。

用 systemd 把 worker 守护起来

worker 必须开机自启、崩溃自拉起，直接交给 systemd：

# /etc/systemd/system/nats-worker.service
[Unit]
Description=NATS JetStream Email Worker
After=network-online.target docker.service
Wants=network-online.target

[Service]
Type=simple
User=www-data
WorkingDirectory=/opt/site-workers
Environment="PYTHONUNBUFFERED=1"
ExecStart=/opt/site-workers/venv/bin/python /opt/site-workers/email_worker.py
Restart=always
RestartSec=5
# 优雅退出给足时间，配合上面的 drain
TimeoutStopSec=60
# 基本的资源与安全收口
NoNewPrivileges=true
PrivateTmp=true

[Install]
WantedBy=multi-user.target

sudo systemctl daemon-reload
sudo systemctl enable --now nats-worker
systemctl status nats-worker
journalctl -u nats-worker -f

TimeoutStopSec=60 很关键：如果给的时间比 drain 还短，systemd 会直接 kill，前面的优雅退出就白写了。

监控：怎么知道队列有没有积压

队列最怕的是「消费者挂了没人发现，消息越积越多」。JetStream 自带监控接口，几行命令就够：

# 看 stream 里还有多少消息、多少字节
nats stream report

# 看 consumer 的待处理数（num_pending 就是积压量）和已确认数
nats consumer info TASKS worker

# 通过监控 HTTP 端口看 JetStream 整体指标
curl -s http://127.0.0.1:8222/jsz | python3 -m json.tool

关键指标是 consumer 的 num_pending（还没被确认的消息数）和 num_redelivered（被重投的次数）。把这两个数采集到你的监控里（比如用前文写过的 VictoriaMetrics + vmagent 抓 8222 的 /jsz），设个阈值告警：

num_pending 持续上涨 → 消费者处理不过来或挂了。
num_redelivered 快速上升 → 有「有毒消息」反复失败，去看 worker 日志。
stream 的 messages 只增不减 → 消费者可能根本没在拉。

个人站到底哪些地方用得上

不是所有操作都值得上队列，我总结了几类真正划算的场景：

邮件与通知：注册欢迎信、评论回复通知、密码重置。发邮件要走外网 SMTP，慢且容易失败，最该异步化。
图片处理：用户上传图片后压缩、转 WebP、生成多尺寸缩略图。CPU 密集、耗时长，放队列里让 worker 慢慢跑，还能配合限速避免把小机器 CPU 打满。
缓存与索引重建：文章发布后重新生成 sitemap、推送到搜索引擎、刷新 CDN 缓存。这些操作经常超时，异步做最稳。
日志与统计落库：高并发下的访问埋点先丢队列，再由 worker 批量写库，避免每次请求都去写数据库。
定时与延迟任务：订单 30 分钟后未支付就关闭这类延迟任务，可以用 JetStream 的定时消息或 worker 内部的时间轮实现。

反过来，这些不要用队列：需要同步返回结果的查询、几毫秒就能完成的轻量写库、需要事务一致性的操作。队列的正确位置是「可延迟、可重试、失败不影响主流程」的活。

五个必须避开的坑

拿 core NATS 当队列用：不开 JetStream，消费者不在线消息就丢。要求可靠投递就必须用 Stream + durable pull consumer。
4222 端口暴露公网：NATS 默认无认证，暴露公网等于把消息总线送人。要么绑回环，要么配 token/用户名密码 + TLS。
消费者不幂等：至少一次投递意味着同一条消息可能被处理多次。发邮件、扣款这类操作必须做去重。
workqueue 流当广播用：workqueue 模式下一条消息只会被一个消费者拿到。需要「多人都收到」要用 limits 策略 + 多个 durable consumer。
不监控就不知道积压：JetStream 不会主动告诉你「消费者挂了」，必须把 num_pending 接进监控，否则等磁盘写满才发现就晚了。

小结

NATS + JetStream 对个人站长来说是一套「刚刚好」的异步方案：单二进制、内存占用小、概念少（Subject / Stream / Consumer 三个词），却提供了磁盘持久化、手动确认、重投、重放这些做可靠任务队列必需的能力。它不像 Kafka 那样要吃几个 G 内存，也不像 RabbitMQ 那样插件体系复杂，一台 2G 的小 VPS 就能跑。

落地顺序建议是：先装 NATS 开 JetStream，建一个 workqueue 流，把「发邮件」这个最痛的点挪进去跑通；确认稳定后，再把图片处理、索引重建、日志落库这些重活陆续搬进来。等这几件事都跑顺了，你会发现整站响应快了一截，服务器负载曲线也平了——这才是消息队列对个人站最大的价值。

Last modification：October 12th, 2026 at 07:24 pm

NATS + JetStream 消息队列实战：给个人站做异步任务解耦，Subject/Stream/Consumer 与五个必踩的坑

zz1984 • 2026 年 10 月 12 日

为什么个人站长也需要消息队列

这篇文章我会从零跑通一套 NATS + JetStream 的环境，讲清核心概念，然后用 PHP 发消息、Python 消费消息，最后落到「个人站有哪些地方真的用得上」。

先分清：核心 NATS 和 JetStream 是两回事

这一点必须先讲明白，否则后面全是坑。

用 Docker 把 NATS 跑起来

JetStream 需要数据目录，所以启动命令里必须带 -js 开关和存储路径。写成 docker-compose 最省事：

version: "3.8"
services:
  nats:
    image: nats:2.10-alpine
    container_name: nats
    restart: unless-stopped
    # -js 开启 JetStream；--store_dir 指定持久化目录
    # -m 8222 是监控/管理 HTTP 端口，只在容器内暴露
    command: >
      -js
      --store_dir /data
      -m 8222
    ports:
      # 客户端连接端口：务必绑定内网/回环，别裸奔公网
      - "127.0.0.1:4222:4222"
      # 监控端口也只给本机
      - "127.0.0.1:8222:8222"
    volumes:
      - ./nats-data:/data
    healthcheck:
      test: ["CMD", "wget", "-qO-", "http://localhost:8222/healthz"]
      interval: 10s
      timeout: 3s
      retries: 3

几个必须注意的点：

4222 端口只绑 127.0.0.1。NATS 默认没有任何认证，绑定到 0.0.0.0 等于把消息系统送人。如果确实需要多机通信，一定要在配置文件里加用户名密码或 token。
数据目录要挂出来。JetStream 的持久化全靠这个目录，不挂卷的话容器一删消息全没。
8222 是监控端口，能看连接数、消息吞吐、JetStream 的使用量，非常好用，但同样不要暴露公网。

启动并验证：

docker compose up -d
docker logs nats --tail 20
# 看到 "Server is ready" 且 "JetStream version: 2.10.x" 就对了

# 用 nats CLI 连一下（宿主机装 nats-top 或官方 nats 客户端）
curl -s http://127.0.0.1:8222/varz | grep -o '"jetstream":{[^}]*}'

如果你不想装 CLI，直接用官方的 nats 二进制（单文件，Go 编译）：

# 下载对应架构的 nats CLI
curl -L https://github.com/nats-io/natscli/releases/latest/download/nats-2.10.18-linux-amd64.zip -o /tmp/nats.zip
unzip -o /tmp/nats.zip -d /tmp/natsbin
sudo install /tmp/natsbin/nats-*/nats /usr/local/bin/nats

# 配置连接上下文
nats context save local --server nats://127.0.0.1:4222
nats context select local
nats server check connection

核心概念：Subject、Stream、Consumer

NATS 的概念比 Kafka 少得多，三个词就能说清：

Subject（主题）

Stream（流）

retention：保留策略。limits 按大小/条数/时间限制（默认），workqueue 表示消息被确认后立刻删除（最省空间，适合任务队列）。
max_age：消息最长保留多久，比如 24h。
storage：file（落盘，支持重启恢复）或 memory（只存内存，快但不抗重启）。做任务队列选 file。
replicas：副本数。单机固定 1。

Consumer（消费者）

Consumer 是「谁在读这个 Stream、读到哪了」的记录。两种模式：

Pull consumer（拉模式）：消费者主动说「给我 10 条」，处理完再 ack。任务队列强烈推荐这种，因为消费速度由自己控制，不会被打爆。
Push consumer（推模式）：服务器主动推给消费者。适合低延迟通知，但控制力弱。

Pull consumer 还分 ephemeral（临时，断开就忘）和 durable（持久，记住 offset）。异步任务必须用 durable，否则消费者重启后无法从上次的位置继续。

用 CLI 把 Stream 和 Consumer 建起来

先建流，再建消费者：

# 建一个 workqueue 类型的流，只保留未被确认的消息
nats stream add TASKS \
  --subjects "tasks.>" \
  --retention workqueue \
  --storage file \
  --max-age 24h \
  --max-msgs 100000 \
  --replicas 1 \
  --discard old

# 建一个持久化的拉取消费者
nats consumer add TASKS worker \
  --pull \
  --deliver all \
  --ack explicit \
  --max-deliver 5 \
  --wait 30s

# 查看状态
nats stream info TASKS
nats consumer info TASKS worker

参数逐个说清楚：

--retention workqueue：消息被 ack 后就删掉，队列不会无限膨胀。注意 workqueue 模式下一条消息只能被一个消费者消费（这正是任务队列想要的）。
--ack explicit：手动确认。只有消费成功你才 ack，处理失败就不 ack，消息会被重投。这是「不丢消息」的基石。
--max-deliver 5：同一条消息最多投 5 次，超了就进「死信」状态，避免有毒消息无限循环。
--wait 30s：拉取请求最长挂起 30 秒，没有新消息就返回空，避免客户端疯狂空轮询。

手动发一条测试消息：

nats pub tasks.email '{"to":"admin@example.com","subject":"测试"}'
# 拉取并确认
nats consumer next TASKS worker --count 1

PHP 侧：把耗时的活丢进队列

WordPress 或自研 PHP 站要发消息，官方有 basis-company/nats.php 这个纯 PHP 客户端，通过 Composer 安装：

composer require basis-company/nats

发布一条任务消息（比如「用户注册后要发的欢迎邮件」）：

<?php
require __DIR__ . '/vendor/autoload.php';

use Basis\Nats\Client;
use Basis\Nats\Configuration;

function enqueue(string $subject, array $payload): bool {
    $config = new Configuration([
        'host' => '127.0.0.1',
        'port' => 4222,
        'timeout' => 2,          // 连接超时，别让页面跟着卡
        'reconnect' => false,    // Web 请求里不重连，失败就让上层降级
    ]);
    try {
        $client = new Client($config);
        // JetStream 的发布必须走 jetstream 层，否则不会落盘
        $client->jetstream()->publish($subject, json_encode($payload, JSON_UNESCAPED_UNICODE));
        return true;
    } catch (\Throwable $e) {
        error_log('nats enqueue failed: ' . $e->getMessage());
        return false;
    }
}

// 用户注册成功后调用
$ok = enqueue('tasks.email', [
    'to' => $user['email'],
    'template' => 'welcome',
    'vars' => ['name' => $user['name']],
]);

// 关键：入队失败不能让主流程崩，写日志 + 兜底（比如塞本地文件稍后补发）
if (!$ok) {
    file_put_contents('/var/log/nats_fallback.log', json_encode($user) . "\n", FILE_APPEND);
}

两个容易踩的点：

必须用 JetStream 层的 publish（客户端上的 jetstream()->publish()），而不是 core NATS 的 publish()。后者不落盘，消费者不在线消息就没了。
Web 请求里的 NATS 客户端要短超时、不重连。如果你让请求去等重连，一次网络抖动就能把整个页面拖死。队列的意义是「快」，不是「强」。

Python 侧：消费者进程怎么写才可靠

消费者通常是一个常驻的 Python 进程，用官方的 nats-py：

pip install nats-py

一个带手动 ack、失败重投、优雅退出的 worker：

import asyncio, json, signal
import nats

class Worker:
    def __init__(self):
        self.nc = None
        self.js = None
        self.sub = None
        self.running = True

    async def connect(self):
        self.nc = await nats.connect(
            "nats://127.0.0.1:4222",
            max_reconnect_attempts=-1,      # 常驻进程应该无限重连
            reconnect_time_wait=2,
            error_cb=self.on_error,
            disconnected_cb=lambda: print("NATS disconnected"),
        )
        self.js = self.nc.jetstream()
        # durable + pull + 手动 ack 三件套
        self.sub = await self.js.pull_subscribe(
            "tasks.email", "worker", stream="TASKS"
        )

    async def on_error(self, e):
        print("NATS error:", e)

    async def run(self):
        while self.running:
            try:
                # 一次拉 5 条，挂起 30 秒
                msgs = await self.sub.fetch(batch=5, timeout=30)
            except nats.errors.TimeoutError:
                continue                      # 没消息，正常，继续等
            for msg in msgs:
                try:
                    data = json.loads(msg.data)
                    await self.handle(data)
                    await msg.ack()           # 只有成功才确认
                except Exception as e:
                    # 不 ack，让 JetStream 按 backoff 重投（受 max_deliver 限制）
                    print("task failed:", e)

    async def handle(self, data):
        # 真正的业务：发邮件、生成缩略图、重新生成 sitemap
        print("sending mail to", data["to"])
        await asyncio.sleep(0.5)              # 模拟耗时操作

    def stop(self, *_):
        self.running = False

async def main():
    w = Worker()
    await w.connect()
    # 收到 SIGTERM 时优雅退出（systemd stop 会发这个信号）
    loop = asyncio.get_event_loop()
    for sig in (signal.SIGINT, signal.SIGTERM):
        loop.add_signal_handler(sig, w.stop)
    await w.run()
    await w.nc.drain()

asyncio.run(main())

这段代码里有几个「可靠性」设计值得展开：

只有处理成功才 ack。抛异常就不 ack，消息会重新投递（默认有 backoff）。这就是「至少一次投递」。代价是业务逻辑必须幂等——同一条消息可能被处理两次，发邮件前最好查一下「这封邮件是否已发过」，或者用消息 ID 去重。
用 drain() 而不是直接断连。drain 会先把「已拉取但没确认」的消息处理完、把未发送的 ack 发出去，再关闭连接，避免重启时丢消息。
监听 SIGTERM。systemd 停止服务时发的就是 SIGTERM，处理好这个信号，systemctl restart 才不会丢任务。
常驻进程用无限重连。和 Web 请求相反，worker 断线了就应该一直重连，直到 NATS 回来。

用 systemd 把 worker 守护起来

worker 必须开机自启、崩溃自拉起，直接交给 systemd：

# /etc/systemd/system/nats-worker.service
[Unit]
Description=NATS JetStream Email Worker
After=network-online.target docker.service
Wants=network-online.target

[Service]
Type=simple
User=www-data
WorkingDirectory=/opt/site-workers
Environment="PYTHONUNBUFFERED=1"
ExecStart=/opt/site-workers/venv/bin/python /opt/site-workers/email_worker.py
Restart=always
RestartSec=5
# 优雅退出给足时间，配合上面的 drain
TimeoutStopSec=60
# 基本的资源与安全收口
NoNewPrivileges=true
PrivateTmp=true

[Install]
WantedBy=multi-user.target

sudo systemctl daemon-reload
sudo systemctl enable --now nats-worker
systemctl status nats-worker
journalctl -u nats-worker -f

TimeoutStopSec=60 很关键：如果给的时间比 drain 还短，systemd 会直接 kill，前面的优雅退出就白写了。

监控：怎么知道队列有没有积压

队列最怕的是「消费者挂了没人发现，消息越积越多」。JetStream 自带监控接口，几行命令就够：

# 看 stream 里还有多少消息、多少字节
nats stream report

# 看 consumer 的待处理数（num_pending 就是积压量）和已确认数
nats consumer info TASKS worker

# 通过监控 HTTP 端口看 JetStream 整体指标
curl -s http://127.0.0.1:8222/jsz | python3 -m json.tool

num_pending 持续上涨 → 消费者处理不过来或挂了。
num_redelivered 快速上升 → 有「有毒消息」反复失败，去看 worker 日志。
stream 的 messages 只增不减 → 消费者可能根本没在拉。

个人站到底哪些地方用得上

不是所有操作都值得上队列，我总结了几类真正划算的场景：

邮件与通知：注册欢迎信、评论回复通知、密码重置。发邮件要走外网 SMTP，慢且容易失败，最该异步化。
图片处理：用户上传图片后压缩、转 WebP、生成多尺寸缩略图。CPU 密集、耗时长，放队列里让 worker 慢慢跑，还能配合限速避免把小机器 CPU 打满。
缓存与索引重建：文章发布后重新生成 sitemap、推送到搜索引擎、刷新 CDN 缓存。这些操作经常超时，异步做最稳。
日志与统计落库：高并发下的访问埋点先丢队列，再由 worker 批量写库，避免每次请求都去写数据库。
定时与延迟任务：订单 30 分钟后未支付就关闭这类延迟任务，可以用 JetStream 的定时消息或 worker 内部的时间轮实现。

五个必须避开的坑

拿 core NATS 当队列用：不开 JetStream，消费者不在线消息就丢。要求可靠投递就必须用 Stream + durable pull consumer。
4222 端口暴露公网：NATS 默认无认证，暴露公网等于把消息总线送人。要么绑回环，要么配 token/用户名密码 + TLS。
消费者不幂等：至少一次投递意味着同一条消息可能被处理多次。发邮件、扣款这类操作必须做去重。
workqueue 流当广播用：workqueue 模式下一条消息只会被一个消费者拿到。需要「多人都收到」要用 limits 策略 + 多个 durable consumer。
不监控就不知道积压：JetStream 不会主动告诉你「消费者挂了」，必须把 num_pending 接进监控，否则等磁盘写满才发现就晚了。

为什么个人站长也需要消息队列

先分清：核心 NATS 和 JetStream 是两回事

用 Docker 把 NATS 跑起来

核心概念：Subject、Stream、Consumer

Subject（主题）

Stream（流）

Consumer（消费者）

用 CLI 把 Stream 和 Consumer 建起来

PHP 侧：把耗时的活丢进队列

Python 侧：消费者进程怎么写才可靠

用 systemd 把 worker 守护起来

监控：怎么知道队列有没有积压

个人站到底哪些地方用得上

五个必须避开的坑

小结

相关文章