消费函数不能直接接收context.Context参数，因主流MQ客户端接口不支持；应将关键值存入消息头，消费端重建context；整批消费需控制循环生命周期；Context取消需各环节显式配合。
消费函数里直接接收 context.Context 参数行不通
Go 语言标准库和主流 MQ 客户端（如 sarama、go-redismq、asynq）的消费者接口，**几乎都不允许你把 context.Context 作为消费回调的入参**。比如 sarama.ConsumerGroup 的 ConsumeClaim 方法签名是 func(context.Context, *sarama.ConsumerGroupSession, sarama.ConsumerGroupClaim) error —— 看似有 context.Context，但它是由客户端内部创建并传入的，生命周期绑定单次消息拉取，不是你外部可控的请求级或任务级上下文。
硬塞进去会破坏客户端的上下文管理逻辑，甚至导致超时不生效、取消被忽略等问题。
用 context.WithValue 包装原始 Context 并存到消息元数据里
真正可行的做法，是在生产端就把你要传递的 context.Context 中的关键值（比如 trace ID、timeout deadline、用户身份）提取出来，写入消息的 headers 或自定义字段；消费端再从消息中读出这些值，重建一个新 context.Context。
例如使用 Kafka：
立即学习“go语言免费学习笔记（深入）”；
// 生产端
ctx := context.WithTimeout(context.Background(), 30*time.Second)
ctx = context.WithValue(ctx, "user_id", "u_123")
<p>// 写入 headers（Kafka 0.11+ 支持）
headers := []sarama.RecordHeader{
{Key: []byte("trace_id"), Value: []byte(getTraceID(ctx))},
{Key: []byte("timeout_sec"), Value: []byte(strconv.FormatInt(30, 10))},
}
msg := &sarama.ProducerMessage{
Topic: "orders",
Value: sarama.StringEncoder("payload"),
Headers: headers,
}
消费端解析：
// 消费端
func (h *Handler) ConsumeClaim(sess sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error {
for msg := range claim.Messages() {
// 从 headers 提取 timeout_sec
var timeoutSec int64 = 5
if hdr := sarama.MessageHeaderValue(msg.Headers, "timeout_sec"); hdr != nil {
if sec, err := strconv.ParseInt(string(hdr), 10, 64); err == nil {
timeoutSec = sec
}
}
<pre class="brush:php;toolbar:false;"> ctx, cancel := context.WithTimeout(context.Background(), time.Second * time.Duration(timeoutSec))
defer cancel()
// 注入 trace_id
if hdr := sarama.MessageHeaderValue(msg.Headers, "trace_id"); hdr != nil {
ctx = context.WithValue(ctx, "trace_id", string(hdr))
}
// 后续业务处理用这个 ctx
h.processOrder(ctx, msg.Value)
}
return nil}
- 不要依赖
context.WithValue传复杂结构体，只传字符串/基本类型，避免内存泄漏和类型断言错误 - Kafka headers 是二进制字节，务必统一编码（推荐 UTF-8）
- RabbitMQ 可用
msg.Headers（AMQP 0.9.1），Redis 则需手动拼接 JSON 字段
用中间层封装 Consumer，支持传入外部 Context 控制整个消费生命周期
如果你需要的是“整批消息消费过程可被外部 Context 取消”，而不是单条消息，那就不能靠每条消息带 context，而要控制消费循环本身。
以 asynq 为例，它原生不支持传入 context.Context 到 ServeMux，但你可以：
- 启动消费前，用
context.WithCancel创建一个父 Context - 在
Server.Run前启动 goroutine 监听该 Context 的 Done，并调用server.Shutdown - 注意：这只能中断拉取消息，已进入 handler 的任务仍会执行完（符合 asynq 设计）
示例：
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Minute)
defer cancel()
<p>srv := asynq.NewServer(redisConnOpt, asynq.Config{...})
mux := asynq.NewServeMux()
mux.HandleFunc("order:create", h.HandleCreateOrder)</p><p>// 单独 goroutine 处理 shutdown
go func() {
<-ctx.Done()
srv.Shutdown()
}()</p><p>if err := srv.Run(mux); err != nil && !errors.Is(err, asynq.ErrServerClosed) {
log.Fatal(err)
}
别忘了 Context 的取消传播必须显式检查
即使你成功把 Context 传进了业务逻辑，如果 handler 里没主动调用 select 监听 ctx.Done()，或者调用了阻塞但不响应 cancel 的函数（比如原生 net.Conn.Read 而非 conn.SetReadDeadline），那 Context 就是摆设。
常见陷阱：
- 数据库查询没用
db.QueryContext，而是db.Query - HTTP 调用没传
ctx给http.NewRequestWithContext - 调用第三方 SDK 时忽略其 Context 参数（比如 AWS SDK 的
WithContext方法）
只要链条上任何一环没传递或没响应 Context，整个超时/取消就断了。
Context 不是自动魔法，它是靠每个函数自觉配合才能穿透到底层系统调用的。
