Beanstalkd 1.13 协议命令与工具详解

Beanstalkd 无管理 CLI，本文用 telnet/nc 逐条实测 1.13 全部协议命令（生产、消费、peek、kick、tube 管理、统计、错误响应）并给出 Python 生产者/消费者实战脚本。

Beanstalkd消息队列Beanstalkd2026/10/22

Beanstalkd 1.13 协议命令与工具详解

目标：Beanstalkd 没有任何管理 CLI 工具，运维与调试的全部入口就是它基于 TCP 11300 端口的 ASCII 文本协议。本文用 telnet/nc 逐条实测全部协议命令（生产、消费、peek、kick、tube 管理、统计监控、错误响应），并给出 Python 标准库的生产者/消费者实战脚本与排障技巧，命令与回显全部实测。

一、环境信息

本文实验环境（实测日期 2026-10-22）：

二、工具概览

Beanstalkd 的发行包里只有一个服务端二进制：没有 beanstalkctl、没有管理子命令、没有 HTTP 管理端口。它的管理面就是协议本身——这点在动手前必须清楚。好处是协议为纯 ASCII 文本，任何能读写 TCP 的工具都能当客户端：

工具定位适用场景
nc（netcat）通用 TCP 管道逐命令调试、脚本化巡检（本文主力工具）
telnet交互式 TCP 终端人工交互式排查，所见即所得
Python/Ruby/Go 等语言客户端库生产代码接入应用投递/消费任务
beanstalkd 自带 -VV 启动选项服务端协议抓包定位「客户端发了什么/服务端回了什么」

工具选型解读：日常运维巡检与验证用 nc 一把梭；需要看着回显一步步敲用 telnet；生产代码用语言客户端库；怀疑「任务到底有没有发出去」时用 -VV 在服务端抓协议明细。

💡 技术拓展：同为文本协议的消息队列，管理手段对比——

软件管理面管理工具
Beanstalkd纯文本协议（TCP 11300）无官方 CLI，靠 nc/telnet/客户端库
RabbitMQAMQP 协议 + HTTP 管理 APIrabbitmqctl、management UI
Redis文本协议（RESP）redis-cli 官方 CLI
NSQ二进制 TCP + HTTP APInsqadmin Web UI

Beanstalkd 是其中唯一「零配套工具」的——一切皆协议，本文就是把协议当 CLI 用。

三、前置条件

服务在跑、端口可达、客户端工具在位，三步核验：

systemctl is-active beanstalkd

实测回显：

active

ss -ltnp | grep 11300

实测回显：

LISTEN 0      1024       127.0.0.1:11300      0.0.0.0:*    users:(("beanstalkd",pid=5585,fd=3))

which nc telnet

实测回显：

/usr/bin/nc
/usr/bin/telnet

（最小化系统缺 nc/telnet 时：apt-get install -y netcat-openbsd inetutils-telnet 补装。）确认服务版本与基线状态：

printf 'stats\r\n' | nc -q 1 127.0.0.1 11300 | grep -E 'version|total-jobs|current-tubes'

实测回显：

total-jobs: 0
current-tubes: 1
version: "1.13"

四、连接与协议基础

理解四条协议规则，后面所有命令的行为才有解释依据。规则来自官方协议文档与本文实测。

4.1 文本协议与 CRLF

命令与响应都是 ASCII 文本行，行尾必须是 \r\n（CRLF）；任务 body 是原样字节流，长度由命令参数声明。这就是为什么全文都用 printf 构造以 CRLF 结尾的输入。用 telnet 交互演示一条完整命令：

printf 'use emails\r\nput 1 0 60 11\r\nhello world\r\n' | nc -q 1 127.0.0.1 11300

实测回显：

USING emails
INSERTED 1

4.2 单连接内命令按序串行处理

同一连接上的命令按发送顺序逐个处理、按相同顺序回送响应。实测一次发三个不同命令，回显严格有序：

printf 'use a1\r\nuse a2\r\nlist-tube-used\r\n' | nc -q 1 127.0.0.1 11300

实测回显：

USING a1
USING a2
USING a2

4.3 use 与 watch 是两套独立指针（最重要的坑）

每条连接维护两组 tube 指针：use 的 tube（put 投递目标，初始 default）与 watch 列表（reserve 取任务的来源集合，初始只有 default）。use 不会把 tube 加入 watch 列表。实测同连接「use 新 tube → put → reserve」：

(printf 'use demo3\r\nput 9 0 30 4\r\ntask\r\nreserve\r\n'; sleep 4) | nc -q 1 127.0.0.1 11300 &
sleep 1.5
printf 'stats\r\n' | nc -q 1 127.0.0.1 11300 | grep -E 'current-jobs-reserved|current-waiting'
wait

实测回显：

USING demo3
INSERTED 4
current-jobs-ready: 2
current-jobs-reserved: 0
current-producers: 1
current-workers: 1
current-waiting: 1
TIMED_OUT

解读：任务 4 已在 demo3 里（ready），但 reserve 等的是 watch 列表（只有 default），队头没人来取，连接一直挂到超时。正确姿势是投递前先 watch <tube>（或消费者单独开连接 watch），见第八章 watch/ignore 实测。

4.4 连接关闭会释放该连接 reserved 的任务

worker 连接断开时，它已 reserve 未 delete 的任务自动回到 ready 队列（等价于隐式 release）。实测：连接 A reserve 任务后直接退出，连接 B 立刻能再次 reserve 到同一任务：

printf 'use t1\r\nput 1 0 60 4\r\ntask\r\n' | nc -q 1 127.0.0.1 11300 | tail -1
printf 'reserve\r\n' | nc -q 1 127.0.0.1 11300 | head -1
printf 'reserve\r\n' | nc -q 1 127.0.0.1 11300 | head -1

实测回显（第三次连接拿到了已断开的第二次连接留下的任务）：

INSERTED 6
RESERVED 6 4
RESERVED 6 4

⚠️ 依赖此机制做「超时兜底」并不可靠：连接正常关闭才触发释放，进程被 kill -9 时由 TTR 机制兜底（见 6.4 节）。消费者代码必须显式处理 delete/release/bury，不要靠断连。

五、生产者命令详解

5.1 use：切换投递管道

use <tube>

响应 USING <tube>；tube 不存在则自动创建（本例 a1/a2 均为随手创建）。生产者每条连接应在一个 use 之后连续 put。

5.2 put：投递任务

put <pri> <delay> <ttr> <bytes>\r\n<data>\r\n

四个参数：pri 优先级（越小越先被消费，小于 1024 视为紧急任务计入 urgent 统计）、delay 延迟秒数（>0 时任务先进 delayed 状态）、ttr 允许 worker 处理的秒数（超时任务被回收重投）、bytes body 字节数。成功响应 INSERTED <id>；内存紧张时响应 BURIED <id>（任务直接进 buried 状态，需手工 kick）。

实测三种投递（正常、延迟、优先级组合）：

printf 'put 100 0 60 4\r\ntask\r\nput 5 0 60 4\r\ntask\r\nput 50 0 60 4\r\ntask\r\n' | nc -q 1 127.0.0.1 11300

实测回显：

INSERTED 2
INSERTED 3
INSERTED 4

printf 'put 1 5 60 4\r\ntask\r\n' | nc -q 1 127.0.0.1 11300

实测回显：

INSERTED 6

第二条 delay=5 的任务（id 6）5 秒内不出现在 ready 队列，可用 peek-delayed 看到（见第七章实测）。

优先级消费顺序实测：三个任务 pri 分别为 100、5、50，消费者 reserve 的取用顺序是 5 → 50 → 100（单连接内连续取，避免断连释放干扰）：

printf 'reserve\r\nreserve\r\nreserve\r\n' | nc -q 1 127.0.0.1 11300

实测回显：

RESERVED 3 4
task
RESERVED 4 4
task
RESERVED 2 4
task

（连接断开后任务 2/3/4 被释放回 ready，符合 4.4 节规则。）

5.3 put 的边界与报错

body 超过服务端 -z 上限（默认 65535）时拒绝：

{ printf 'put 1 0 60 65536\r\n'; head -c 65536 /dev/zero | tr '\0' 'z'; printf '\r\n'; } | nc -q 1 127.0.0.1 11300 | head -1

实测回显：

JOB_TOO_BIG

命令行格式不完整（缺 bytes 字段）时：

printf 'put 1 0 60\r\n' | nc -q 1 127.0.0.1 11300

实测回显：

BAD_FORMAT

⚠️ put 最常见的客户端 bug 是声明的 bytes 与实际 body 长度不符：服务端按声明的字节数截取 body，随后必须读到 \r\n，对不齐会导致本条命令错位甚至吞掉下一条命令。发送前务必用真实字节数（注意中文等多字节字符）。

六、消费者命令详解

6.1 reserve 与 reserve-with-timeout：认领任务

reserve
reserve-with-timeout <seconds>

无任务可取时 reserve 无限等待；reserve-with-timeout 最多等 seconds 秒，超时回 TIMED_OUT。取到任务回 RESERVED <id> <bytes> + body。空队列实测：

printf 'reserve-with-timeout 1\r\n' | nc -q 1 127.0.0.1 11300

实测回显：

TIMED_OUT

多 tube 同时监听时，reserve 从 watch 列表所有 tube 中取优先级最高的任务（实测见 8.1 节）。

6.2 delete：完成任务

delete <id>

成功 DELETED；任务不存在/不属于可删除状态回 NOT_FOUND。实测正常删除与两种失败：

printf 'reserve\r\ndelete 6\r\ndelete 6\r\ndelete 999\r\n' | nc -q 1 127.0.0.1 11300

实测回显：

RESERVED 6 4
task
DELETED
NOT_FOUND
NOT_FOUND

重复 delete 与删除不存在的 id 都是 NOT_FOUND——delete 只对 ready/reserved/delayed/buried 状态的任务有效，对已删除任务自然无效。

6.3 release：放弃并重新入队

release <id> <pri> <delay>

把 reserved 任务放回 ready 队列，可顺手改优先级与延迟。响应 RELEASED/BURIED/NOT_FOUND。实测改优先级重入队（任务 4 原 pri 50，改为 10 后立即排到队首）：

printf 'reserve\r\nrelease 4 10 0\r\npeek-ready\r\n' | nc -q 1 127.0.0.1 11300

实测回显：

RESERVED 4 4
task
RELEASED
FOUND 4 4
task

带延迟 release 实测（3 秒后再可消费，先落 delayed）：

printf 'reserve\r\nrelease 4 20 3\r\npeek-delayed\r\n' | nc -q 1 127.0.0.1 11300

实测回显：

RESERVED 4 4
task
RELEASED
FOUND 4 4
task

6.4 TTR 超时与 DEADLINE_SOON：worker 心跳机制

每个 reserved 任务有 ttr 时限。到期的任务被服务端回收重投（stats 的 job-timeouts 计数加一），实测：ttr=3 的任务被 reserve 后放置不管，约 3 秒后自动回到 ready：

(printf 'put 9 0 3 4\r\ntask\r\nreserve\r\n'; sleep 4) | nc -q 1 127.0.0.1 11300 | head -2
printf 'stats\r\n' | nc -q 1 127.0.0.1 11300 | grep -E 'job-timeouts'
printf 'stats-job 7\r\n' | nc -q 1 127.0.0.1 11300 | grep -E 'state|timeouts'

实测回显：

INSERTED 7
RESERVED 7 4
job-timeouts: 1
state: ready
timeouts: 0

worker 侧对应的心跳信号是 DEADLINE_SOON：连接手握一个即将到期的 reserved 任务时再发 reserve，服务端立即回 DEADLINE_SOON 提醒「先处理手头的，否则要被回收了」。实测（ttr=3，2.5 秒后再发 reserve）：

(printf 'put 9 0 3 4\r\ntask\r\nreserve\r\n'; sleep 2.5; printf 'reserve\r\n') | nc -q 8 127.0.0.1 11300 | head -4

实测回显：

INSERTED 9
RESERVED 9 4
task
DEADLINE_SOON

收到 DEADLINE_SOON 后 worker 应立即 finish（delete）或 release 手头任务，否则任务到期被回收、别的 worker 会重复处理。

6.5 touch：续借处理时间

touch <id>

把 reserved 任务的 ttr 时限重新计满，用于长任务续命。响应 TOUCHED/NOT_FOUND。完整实测（同连接 use+put+watch 后 reserve、touch、delete 一气呵成）：

(printf 'use demo4\r\nput 9 0 30 4\r\ntask\r\nwatch demo4\r\nignore default\r\nreserve\r\ntouch 5\r\ndelete 5\r\n'; sleep 1) | nc -q 3 127.0.0.1 11300

实测回显：

USING demo4
INSERTED 5
WATCHING 2
WATCHING 1
RESERVED 5 4
task
TOUCHED
DELETED

6.6 bury 与 kick：暂存与复活

bury <id> <pri>
kick <count>
kick-job <id>

bury 把任务埋进 buried 状态（不再被消费，通常用于「有问题待人工处理」的任务堆）；kick 从 buried 按优先级挑 count 条放回 ready；kick-job 精确复活一条。实测：

printf 'reserve\r\nbury 4 1\r\npeek-buried\r\nkick 1\r\nkick 1\r\n' | nc -q 1 127.0.0.1 11300

实测回显：

RESERVED 4 4
task
BURIED
FOUND 4 4
task
KICKED 1
KICKED 0

（第二次 kick 无物可踢，回 KICKED 0。）kick-job 实测：

printf 'reserve\r\nbury 4 1\r\nkick-job 4\r\nkick-job 999\r\n' | nc -q 1 127.0.0.1 11300

实测回显：

RESERVED 4 4
task
BURIED
KICKED
NOT_FOUND

七、peek 命令：偷看不取走

peek <id>
peek-ready
peek-delayed
peek-buried

peek 系列查看任务但不改变其状态，响应 FOUND <id> <bytes> + body，或 NOT_FOUND。注意：peek-ready/delayed/buried 作用于当前 use 的 tube。实测：

printf 'peek-ready\r\n' | nc -q 1 127.0.0.1 11300

实测回显：

FOUND 3 4
task

printf 'peek 1\r\n' | nc -q 1 127.0.0.1 11300

实测回显（peek <id> 按全局 id 查，不限 tube）：

FOUND 1 11
hello world

printf 'peek-delayed\r\npeek-buried\r\n' | nc -q 1 127.0.0.1 11300

实测回显（delayed/buried 为空时）：

NOT_FOUND
NOT_FOUND

八、tube 管理命令

8.1 watch / ignore：维护监听列表

watch <tube>
ignore <tube>

响应 WATCHING <数量>；ignore 掉列表里最后一条 tube 时拒绝并回 NOT_IGNORED。实测双 worker 监听两个 tube：

printf 'watch emails\r\nwatch sms\r\nignore default\r\nreserve\r\nreserve\r\n' | nc -q 1 127.0.0.1 11300

实测回显（emails 里有任务 10、sms 里有任务 11，reserve 依优先级跨 tube 取）：

WATCHING 2
WATCHING 3
WATCHING 2
RESERVED 10 7
mailer1
RESERVED 11 6
sms001

ignore 的两个边界实测（新连接默认 watch default）：

printf 'watch default\r\nignore default\r\nignore sms\r\n' | nc -q 1 127.0.0.1 11300

实测回显：

WATCHING 1
NOT_IGNORED
WATCHING 1

解读：ignore 最后一条被拒（NOT_IGNORED）；ignore 一个根本没 watch 的 tube（sms）不报错，计数不变——是静默成功而非错误。

8.2 pause-tube：暂停整条管道

pause-tube <tube> <delay>

让 tube 的任务在 delay 秒内不可被 reserve，响应 PAUSED/NOT_FOUND。实测暂停 blocked 管道 60 秒后，只监听它的 worker 取不到任务（尽管里面有 ready 任务）：

printf 'watch blocked\r\nignore default\r\npause-tube blocked 60\r\nreserve-with-timeout 2\r\n' | nc -q 1 127.0.0.1 11300 | grep -vE '^  '

实测回显：

WATCHING 2
WATCHING 1
PAUSED
TIMED_OUT

暂停状态可从 stats-tube 的 pause/pause-time-left 字段观察（实测 pause: 60、pause-time-left: 59）。典型用途：下游故障时暂停消费、削峰场景临时减压；到时自动恢复，无需手工解除。

8.3 list 系列：查看 tube

list-tubes
list-tube-used
list-tubes-watched

实测（list-tubes 返回 YAML，list-tube-used 返回 USING 行，list-tubes-watched 返回 YAML）：

printf 'list-tubes\r\nlist-tube-used\r\nlist-tubes-watched\r\n' | nc -q 1 127.0.0.1 11300

实测回显：

OK 29
---
- default
- emails
- sms

USING default
OK 14
---
- default

九、统计与监控命令

9.1 stats：全局统计（监控数据源）

stats

响应 OK <字节数> + 一段 YAML。全新实例的完整实测输出：

printf 'stats\r\n' | nc -q 1 127.0.0.1 11300

实测回显：

OK 1017
---
current-jobs-urgent: 0
current-jobs-ready: 0
current-jobs-reserved: 0
current-jobs-delayed: 0
current-jobs-buried: 0
cmd-put: 0
cmd-peek: 0
cmd-peek-ready: 0
cmd-peek-delayed: 0
cmd-peek-buried: 0
cmd-reserve: 0
cmd-reserve-with-timeout: 0
cmd-delete: 0
cmd-release: 0
cmd-use: 0
cmd-watch: 0
cmd-ignore: 0
cmd-bury: 0
cmd-kick: 0
cmd-touch: 0
cmd-stats: 1
cmd-stats-job: 0
cmd-stats-tube: 0
cmd-list-tubes: 0
cmd-list-tube-used: 0
cmd-list-tubes-watched: 0
cmd-pause-tube: 0
job-timeouts: 0
total-jobs: 0
max-job-size: 65535
current-tubes: 1
current-connections: 1
current-producers: 0
current-workers: 0
current-waiting: 0
total-connections: 1
pid: 4054
version: "1.13"
rusage-utime: 0.001090
rusage-stime: 0.005450
uptime: 1192
binlog-oldest-index: 1
binlog-current-index: 1
binlog-records-migrated: 0
binlog-records-written: 0
binlog-max-size: 10485760
draining: false
id: 8871d4748e040447
hostname: "ubu22-beanstalkd"
os: "#138~22.04.1-Ubuntu SMP PREEMPT_DYNAMIC Fri Aug  7 13:43:15 UTC "
platform: "x86_64"

字段分组解读：

  • current-jobs-*：五类状态的当前任务数（urgent/ready/reserved/delayed/buried），巡检核心指标；
  • cmd-*：各协议命令的累计调用次数（本次启动以来）；
  • job-timeouts/total-jobs：任务超时与累计投递数；
  • max-job-size/binlog-*：服务端配置与 WAL 状态的运行时反映；
  • current-connections/producers/workers/waiting：连接画像，waiting 长期堆积说明 worker 不够；
  • pid/version/uptime/rusage-*：进程身份与资源消耗；
  • draining：是否处于拒绝写入的排水模式（见 10.3 节）。

监控口径提示：开启 WAL 后进程重启会回放恢复任务，current-jobs-* 恢复真实存量，但 total-jobs、cmd-* 从零重计——判断存量永远用 current-jobs-*。

9.2 stats-tube：单管道统计

stats-tube <tube>

实测（emails 管道）：

printf 'stats-tube emails\r\n' | nc -q 1 127.0.0.1 11300

实测回显：

OK 266
---
name: "emails"
current-jobs-urgent: 6
current-jobs-ready: 6
current-jobs-reserved: 0
current-jobs-delayed: 0
current-jobs-buried: 0
total-jobs: 5
current-using: 0
current-watching: 1
current-waiting: 0
cmd-delete: 0
cmd-pause-tube: 0
pause: 0
pause-time-left: 0

要点解读：tube 级的 current-using/watching/waiting 是容量规划的直接依据；pause/pause-time-left 观察暂停状态。管道不存在时回 NOT_FOUND。

9.3 stats-job：单任务画像

stats-job <id>

实测：

printf 'stats-job 10\r\n' | nc -q 1 127.0.0.1 11300

实测回显：

OK 147
---
id: 10
tube: "emails"
state: ready
pri: 1
age: 29
delay: 0
ttr: 60
time-left: 0
file: 1
reserves: 1
timeouts: 0
releases: 0
buries: 0
kicks: 0

解读：age 距创建秒数、time-left reserved 状态的剩余 ttr、reserves/timeouts/releases/buries/kicks 是这条任务的生命周期履历——排查「一条任务为什么被反复重投」就盯 timeouts/releases。

十、错误响应、quit 与运维信号

10.1 错误响应总表（全部实测）

响应触发条件本文实测位置
INSERTED <id>put 成功5.2
BURIED <id>put 成功但内存紧张直接埋葬协议定义（低内存场景）
USING <tube>use 成功5.1
RESERVED <id> <bytes>reserve 成功6.1
DEADLINE_SOON手握临期 reserved 任务时发 reserve6.4
TIMED_OUTreserve-with-timeout 超时6.1
DELETED/TOUCHED/RELEASED/BURIED/KICKED对应操作成功6.2–6.6
PAUSEDpause-tube 成功8.2
WATCHING <n>/IGNORING <n>watch/ignore 成功8.1
NOT_IGNORED试图 ignore 最后一条监听8.1
NOT_FOUND任务/管道不存在或状态不符6.2/8.2/9.2
JOB_TOO_BIGbody 超过 -z 上限5.3
BAD_FORMAT命令行格式错误5.3
UNKNOWN_COMMAND未知命令10.2
DRAINING排水模式下拒绝 put10.3
OUT_OF_MEMORY/INTERNAL_ERROR服务端资源/内部错误协议定义（不可稳定复现）

10.2 UNKNOWN_COMMAND 与 quit

无法识别的命令回 UNKNOWN_COMMAND（注意不是 BAD_FORMAT——格式错误的已知命令才是 BAD_FORMAT）：

printf 'foo bar\r\n' | nc -q 1 127.0.0.1 11300

实测回显：

UNKNOWN_COMMAND

quit 结束当前连接，无任何响应、连接直接关闭：

printf 'quit\r\n' | nc -q 1 127.0.0.1 11300; echo "NC_EXIT=$?"

实测回显：

NC_EXIT=0

10.3 SIGUSR1 与排水模式（DRAINING）

向进程发 SIGUSR1 进入排水模式：停止接受新任务（put 回 DRAINING），存量消费不受影响，用于停机前优雅收尾。实测：

kill -USR1 $(ss -ltnp | grep 11300 | grep -oE 'pid=[0-9]+' | cut -d= -f2)
sleep 0.3
printf 'put 1 0 60 4\r\ntask\r\n' | nc -q 1 127.0.0.1 11300 | head -1
printf 'stats\r\n' | nc -q 1 127.0.0.1 11300 | grep draining

实测回显：

DRAINING
draining: true

排水模式只能通过重启进程退出：

systemctl restart beanstalkd
printf 'put 1 0 60 4\r\ntask\r\n' | nc -q 1 127.0.0.1 11300 | head -1
printf 'stats\r\n' | nc -q 1 127.0.0.1 11300 | grep draining

实测回显：

INSERTED 13
draining: false

10.4 -VV：服务端协议抓包排障

怀疑「客户端到底发了什么」时，用 -VV 跑前台实例即可看到每条协议收发（日志在 stdout）：

/usr/local/bin/beanstalkd -p 11408 -VV > /var/log/bs-vv.out 2>&1 &
sleep 1.5
printf 'put 1 0 60 5\r\nhello\r\n' | nc -q 1 127.0.0.1 11408 > /dev/null
sleep 1
kill %1
cat /var/log/bs-vv.out

实测回显：

pid 4696
bind 3 0.0.0.0:11408
accept 5
<5 command put
<5 job 1
>5 reply INSERTED 1
close 5

< 为客户端发来的命令、> 为服务端回执——「客户端说投了任务但队列里没有」这类扯皮一抓便知。

十一、Python 客户端实战（仅标准库）

11.1 生产者脚本

/opt/beanstalkd-demo/producer.py——向 emails tube 投递 5 个 JSON 任务：

#!/usr/bin/env python3
"""生产者示例：向 emails tube 投递 5 个任务（仅用标准库 socket）"""
import json
import socket


def recv_line(sock):
    buf = b""
    while not buf.endswith(b"\r\n"):
        buf += sock.recv(1)
    return buf


def main():
    sock = socket.create_connection(("127.0.0.1", 11301))
    f = sock.makefile("rwb")

    # 切换到 emails tube
    f.write(b"use emails\r\n")
    f.flush()
    print("use emails ->", recv_line(sock).decode().strip())

    for i in range(1, 6):
        body = json.dumps({"to": f"user{i}@example.com", "subject": f"report-{i}"}).encode()
        pri, delay, ttr = 10 * i, 0, 60
        f.write(f"put {pri} {delay} {ttr} {len(body)}\r\n".encode())
        f.write(body + b"\r\n")
        f.flush()
        print(f"put pri={pri} ->", recv_line(sock).decode().strip())

    f.write(b"quit\r\n")
    f.flush()
    sock.close()


if __name__ == "__main__":
    main()

实测运行（服务监听端口按实际环境改；本文实验服务在 11301）：

python3 producer.py

实测回显：

use emails -> USING emails
put pri=10 -> INSERTED 14
put pri=20 -> INSERTED 15
put pri=30 -> INSERTED 16
put pri=40 -> INSERTED 17
put pri=50 -> INSERTED 18

11.2 消费者脚本

/opt/beanstalkd-demo/worker.py——只监听 emails，认领 5 个任务并删除：

#!/usr/bin/env python3
"""消费者示例：从 emails tube 认领并处理 5 个任务（仅用标准库 socket）"""
import json
import socket


def main():
    sock = socket.create_connection(("127.0.0.1", 11301))
    f = sock.makefile("rwb")

    # 只监听 emails tube，不再监听 default
    f.write(b"watch emails\r\n")
    f.flush()
    print("watch emails ->", f.readline().decode().strip())
    f.write(b"ignore default\r\n")
    f.flush()
    print("ignore default ->", f.readline().decode().strip())

    for _ in range(5):
        f.write(b"reserve\r\n")                  # 认领一个任务
        f.flush()
        line = f.readline().decode().strip()     # RESERVED <id> <bytes>
        print("reserve ->", line)
        parts = line.split()
        job_id, length = int(parts[1]), int(parts[2])
        body = f.read(length + 2)[:-2]           # 读 body 及结尾 CRLF
        task = json.loads(body)
        print(f"处理任务 {job_id}: 收件人={task['to']} 主题={task['subject']}")
        f.write(f"delete {job_id}\r\n".encode())
        f.flush()
        print(f"delete {job_id} ->", f.readline().decode().strip())

    f.write(b"quit\r\n")
    f.flush()
    sock.close()


if __name__ == "__main__":
    main()

实测运行：

python3 worker.py

实测回显：

watch emails -> WATCHING 2
ignore default -> WATCHING 1
reserve -> RESERVED 14 50
处理任务 14: 收件人=user1@example.com 主题=report-1
delete 14 -> DELETED
reserve -> RESERVED 15 50
处理任务 15: 收件人=user2@example.com 主题=report-2
delete 15 -> DELETED
reserve -> RESERVED 16 50
处理任务 16: 收件人=user3@example.com 主题=report-3
delete 16 -> DELETED
reserve -> RESERVED 17 50
处理任务 17: 收件人=user4@example.com 主题=report-4
delete 17 -> DELETED
reserve -> RESERVED 18 50
处理任务 18: 收件人=user5@example.com 主题=report-5
delete 18 -> DELETED

消费完成后核对管道归零：

printf 'stats-tube emails\r\n' | nc -q 1 127.0.0.1 11301 | grep 'current-jobs-ready'

实测回显：

current-jobs-ready: 0

⚠️ 读写必须走同一个缓冲对象：示例统一用 f（makefile 对象）读写，若混用 sock.recv 与 f.readline，缓冲层数据会错位——实测混用时 body 读取错位直接导致 JSON 解析失败。

十二、注意事项

  • 协议无认证、无 TLS：安全完全依赖网络边界（绑定内网/回环 + 防火墙），暴露公网等于队列裸奔；
  • 单条连接同一时刻只应有一个挂起的 reserve，发第二个 reserve 前先收完上一个响应；
  • 命令行（含 CRLF）最长 224 字节，tube 名最长 200 字节，超限按 BAD_FORMAT 处理；
  • body 对服务端透明：多语言消费时自行约定序列化格式（本文用 JSON），版本升级注意兼容；
  • 状态统计里 current-jobs-* 是存量口径、cmd-*/total-jobs 是开机以来增量口径（WAL 回放后前者真、后者从零）；
  • 消费者务必为长任务设置合理的 ttr 并配合 touch，ttr 太短会造成任务被反复重投（stats-job 的 timeouts 暴露此问题）。

十三、常见问题（FAQ）

Q1：脚本里用 nc 测试，发完 reserve 就收 TIMED_OUT，明明队列里有任务？

现象：printf 'reserve\r\n' | nc -q 1 127.0.0.1 11300 在有 ready 任务时仍回 TIMED_OUT。定位：nc 在 stdin 关闭（printf 写完）后立即对 socket 做半关闭（FIN），服务端对「等待中且已半关闭」的连接直接回 TIMED_OUT（源码 prot.c STATE_WAIT 分支），reserve 根本没等到派发。解决：让 stdin 在等待期间保持打开——

(printf 'reserve\r\n'; sleep 2) | nc -q 3 127.0.0.1 11300

本文所有涉及等待的实测都采用这个「子 shell 拖住 stdin」的写法。

Q2：同一条连接 put 进 demo 管道后 reserve 取不到？

见 4.3 节实测：use 只切换投递管道，不加入 watch 列表；reserve 只从 watch 列表取。消费者要么 watch <tube>，要么让生产者与消费者分连接各司其职。

Q3：任务消失/重复执行怎么排查？

三步定位：① stats 看 job-timeouts 是否增长（增长=worker 处理超 ttr 被回收重投）；② stats-job <id> 看该任务的 timeouts/releases 履历；③ 复杂扯皮用 -VV 抓服务端协议明细（10.4 节）。任务「消失」多为被 worker delete 或 TTR 回收后又被别的 worker 取走，current-jobs-* 与 stats-job 状态对照即可分辨。

Q4：buried 任务堆了一堆怎么处理？

peek-buried 抽查内容，能修的 kick <数量> 放回 ready 重跑，确认废弃的逐条 delete <id>（kick 回 ready 后再 delete）。buried 数量长期增长说明业务侧把失败任务一股脑 bury 了，应在 worker 里区分「可重试」与「需人工」。

Q5：想把某条管道停掉又不删任务？

pause-tube <tube> <大延迟秒数>（如 86400 暂停一天），任务原样保留，stats-tube 的 pause-time-left 可观察剩余时间；到期自动恢复。彻底清空某管道则需逐条 peek + delete（协议没有清空管道的命令）。

十四、参考链接