本文介绍如何在多消费者场景下，通过回调机制让 beanstalkd 的发布者及时获知任务执行结果，涵盖任务结构设计、消费者逻辑处理（含非目标任务处置）、go/python 示例代码及关键注意事项。
本文介绍如何在多消费者场景下，通过回调机制让 beanstalkd 的发布者及时获知任务执行结果，涵盖任务结构设计、消费者逻辑处理（含非目标任务处置）、go/python 示例代码及关键注意事项。
Beanstalkd 本身不提供内置的“任务完成通知”机制，因此在分布式多消费者架构中（如 1 个 Go 发布者 + 2 个 Python 消费者），必须通过显式设计的回调协议实现状态回传。核心思路是：将任务元信息（含唯一 ID、业务载荷和回调地址）序列化为结构化数据（如 JSON）入队；消费者处理完毕后，主动向指定回调端点（HTTP 接口或另一 Beanstalkd 队列）上报成功/失败状态；发布者监听该回调通道，完成闭环。
✅ 正确的任务结构设计（JSON 化 Job）
推荐定义统一的 Job 请求结构，确保跨语言兼容性。例如：
// Go 端构造任务（发布前）
type JobRequest struct {
ID string `json:"id"`
RequestBody []byte `json:"request_body"`
CallbackURL string `json:"callback_url"` // 如 "http://publisher:8080/callback"
TargetMachine string `json:"target_machine"` // 显式声明目标机器标识，如 "machine-one"
}
job := JobRequest{
ID: uuid.New().String(),
RequestBody: []byte("order_id=ORD-12345"),
CallbackURL: "http://publisher-api:8080/job-status",
TargetMachine: "machine-one",
}
body, _ := json.Marshal(job)
id, _ := c.Put(body, 1, 0, 120*time.Second)
? 关键点：
CallbackURL是发布者暴露的 HTTP Webhook 地址（或另一 Beanstalkd 队列名），用于接收状态；TargetMachine字段明确指定应由哪台消费者处理，避免盲目匹配逻辑。
✅ Python 消费者：精准路由 + 合理任务处置
你的原始代码中用 job.body == "one" 判断归属，存在严重缺陷——既无法扩展，也不支持结构化数据。应改为解析 JSON 并校验 target_machine：
钉钉 AI 表格（多维表）操作技能。使用 mcporter CLI 连接钉钉官方新版 AI 表格 MCP server，基于 baseId / tableId / fieldId / recordId 体系执行 Base、Table、Field、Record 的查询与增删改。适用于创建 AI 表格、搜索表格、读取...
import beanstalkc
import json
import requests
def get_beanstalk_data():
beanstalk = beanstalkc.Connection(host='127.0.0.1', port=11300)
beanstalk.use('cloud')
beanstalk.watch('cloud')
beanstalk.ignore('default')
while True:
job = beanstalk.reserve()
try:
# 1. 解析 JSON 任务
task = json.loads(job.body)
target = task.get("target_machine")
# 2. 精准路由：仅处理本机负责的任务
if target != "machine-one": # 当前消费者为 machine-one
# ❌ 错误做法：直接 release 或 bury → 可能导致重复处理或堆积
# ✅ 正确做法：bury + 记录日志，交由运维或重试机制干预
job.bury() # 标记为"永久性不匹配"，需人工介入
print(f"[SKIP] Job {task['id']} not for this machine ({target}) → buried")
continue
# 3. 执行实际业务逻辑（此处模拟）
result = process_order(task["request_body"])
# 4. 上报状态到回调地址（HTTP）
callback_url = task.get("callback_url")
if callback_url:
requests.post(callback_url, json={
"job_id": task["id"],
"status": "success",
"result": result,
"processed_by": "machine-one"
}, timeout=5)
# 5. ✅ 仅在此处删除任务（成功后）
job.delete()
except Exception as e:
# 临时性错误？→ 不 delete，reserve 超时后自动重入 ready 队列
# 永久性错误？→ 上报失败并 delete（避免重复）
error_msg = str(e)
if callback_url:
requests.post(callback_url, json={
"job_id": task.get("id", "unknown"),
"status": "failed",
"error": error_msg,
"processed_by": "machine-one"
})
job.delete() # 永久失败，清理队列
⚠️ 关于
releasevsbury的关键决策：
- 若任务明确不属于当前消费者（如
target_machine != "machine-one"），绝不可release—— 这会使其立即被另一消费者争抢，造成无意义循环；- 应使用
bury将其移至 buried 队列，配合监控告警，由运维或专用重试服务统一处理；release仅适用于临时性失败（如依赖服务短暂不可用），且需设置合理的delay（如job.release(delay=30)），避免雪崩。
✅ Go 发布者：启动回调监听服务
发布者需提供 HTTP 接口接收状态，例如：
// 在 main.go 中添加回调服务
http.HandleFunc("/job-status", func(w http.ResponseWriter, r *http.Request) {
var status struct {
JobID string `json:"job_id"`
Status string `json:"status"`
Result string `json:"result,omitempty"`
Error string `json:"error,omitempty"`
ProcessedBy string `json:"processed_by"`
}
json.NewDecoder(r.Body).Decode(&status)
switch status.Status {
case "success":
log.Printf("✅ Job %s completed by %s: %s", status.JobID, status.ProcessedBy, status.Result)
case "failed":
log.Printf("❌ Job %s failed on %s: %s", status.JobID, status.ProcessedBy, status.Error)
}
w.WriteHeader(http.StatusOK)
})
log.Println("Callback server listening on :8080")
http.ListenAndServe(":8080", nil)
? 总结与最佳实践
- 状态驱动，而非轮询：永远不要让发布者主动查询任务状态，而是由消费者完成时主动回调；
-
结构化任务体：强制使用 JSON Schema，包含
id、callback_url、target_machine等必需字段； -
严格生命周期管理：
reserve→ 处理 →delete（成功/永久失败）；bury（明确不匹配）；release（临时失败+延迟）； -
幂等性保障：回调接口需支持重复请求（如通过
job_id去重）； -
监控必备：监控
buried队列长度、回调失败率、平均处理时长，及时发现路由或下游异常。
通过以上设计，你将构建出高可靠、可追溯、易运维的 Beanstalkd 分布式任务系统。
