登录
推荐 文章 Go 技术 课程 下载 专题 AI
首页 >  Golang >  Go教程

Go NATS Go 客户端怎样为消息附加请求头

来源:17golang原创

时间:2026-10-30 23:29:58 149浏览 收藏

给 NATS Go 消息附加头部,使用 nats.NewMsg(subject) 构造消息,通过 msg.Header.Set 或 Add 写入字段,再调用 nc.PublishMsg(msg)。普通 Publish(subject, data) 只有主题与正文参数,不能靠把元数据拼进 subject 来替代独立消息头。

官方地址:https://nats.io/

本文以发票事件为例:正文放发票编号,头部放追踪标识、内容类型和业务标签。这里的“请求头”指 NATS 消息头;Core NATS 发布订阅也能携带它,不要求消息一定是请求回复,更不代表 HTTP 请求。

先确定主题、正文和头部各自负责什么

Subject 决定订阅匹配,Data 承载业务正文,Header 承载附加元数据。本文约定主题为 demo.invoice.created,正文为 invoice-042,X-Trace-Id 是追踪标识,X-Label 是允许多个值的标签。消息头不会让普通主题订阅自动变成按字段筛选的订阅。

先约定必填字段、单值或多值、长度以及缺失处理,再写生产者和消费者。追踪标识只是调用方提供的数据,不应直接当作授权凭据;敏感身份信息是否允许传播,需要业务另外设计。

NATS消息主题正文和头部的静态组成
消息字段结构示意:Subject、Data 与 Header 分工不同;Header 中保存本文约定的追踪标识和多值标签。

连接准备:让服务端支持头部,让订阅先建立

示例使用已经启动的本地 NATS 服务,不包含服务部署与账户配置。在独立 Go 模块中添加依赖,并把下文代码保存为 main.go;已有工程应按自身依赖策略选择和固定版本。

# 在独立示例目录中初始化并加入官方 Go 客户端依赖
go mod init example.com/nats-header-demo
go get github.com/nats-io/nats.go
# 文件保存后运行;需要本地 NATS 服务已可连接
go run .

连接后用 HeadersSupported() 判断当前服务端能力。不支持时应明确报错,不要静默丢掉业务依赖的字段。示例用 SubscribeSync 创建订阅,并在发布前调用 FlushTimeout,把订阅注册与发布的关系写清楚。网络、权限和连接状态仍可能导致操作失败,每个错误都要向上返回。

构造消息:Set 写单值,Add 写多值

NewMsg 会准备可写的 Header,适合直接添加字段。如果自己写 &nats.Msg{Subject: ...},需要先给 Header 分配非 nil 的 map,再执行写入。不要把 nil Header 的读取和写入混为一谈。

package main

import (
    "fmt"
    "log"
    "time"

    "github.com/nats-io/nats.go"
)

func run() error {
    // 示例需要本地已启动、允许访问的 NATS 服务。
    nc, err := nats.Connect(nats.DefaultURL)
    if err != nil { return fmt.Errorf("连接: %w", err) }
    defer nc.Close()
    if !nc.HeadersSupported() {
        return fmt.Errorf("当前连接的服务端不支持消息头")
    }

    sub, err := nc.SubscribeSync("demo.invoice.created")
    if err != nil { return fmt.Errorf("订阅: %w", err) }
    defer sub.Unsubscribe()
    // 先确认服务端处理了订阅,减少示例中的订阅建立竞态。
    if err := nc.FlushTimeout(2 * time.Second); err != nil {
        return fmt.Errorf("刷新订阅: %w", err)
    }

    msg := nats.NewMsg("demo.invoice.created")
    msg.Data = []byte("invoice-042")
    msg.Header.Set("Content-Type", "text/plain; charset=utf-8")
    msg.Header.Set("X-Trace-Id", "trace-invoice-042")
    // Add 保留同名字段的多个值,Set 则替换整组值。
    msg.Header.Add("X-Label", "billing")
    msg.Header.Add("X-Label", "sandbox")
    if err := nc.PublishMsg(msg); err != nil {
        return fmt.Errorf("发布: %w", err)
    }
    if err := nc.FlushTimeout(2 * time.Second); err != nil {
        return fmt.Errorf("刷新发布: %w", err)
    }

    got, err := sub.NextMsg(2 * time.Second)
    if err != nil { return fmt.Errorf("接收: %w", err) }
    // 键名大小写必须与双方约定一致,Values 用于读取全部值。
    fmt.Println(got.Header.Get("X-Trace-Id"))
    fmt.Println(got.Header.Values("X-Label"))
    fmt.Println(string(got.Data))
    return nil
}

func main() {
    if err := run(); err != nil { log.Fatal(err) }
}

头部操作的关键是 Set 替换同名字段的整组值,Add 追加一个值,Get 取得首值,Values 取得全部值。本文两个 X-Label 使用两次 Add;改成两次 Set,最后只会留下 sandbox。NATS Go 的这些头部操作区分键名大小写,不能因为其设计参考了 http.Header,就假设会自动统一键名。

在这个例子中,若本地服务可连接、订阅和发送均成功,按代码推导应打印三行:

trace-invoice-042
[billing sandbox]
invoice-042

以上为预期结果说明,未在本文制作过程中运行。检查时应同时关注追踪字段、标签完整性和正文,而不是只看到发票编号就认为头部传递成功。

发送结果:PublishMsg 和 Flush 分别说明什么

PublishMsg 返回 nil 不能证明消费者已经完成处理;FlushTimeout 利用连接上的协议往返,等待服务端处理此前发送的数据,也不是业务完成回执。示例中的 NextMsg 只是让当前订阅收到消息,并不代表其他服务已入库、计费或发通知。

这里有两组对象:发送操作包含 PublishMsg 与 FlushTimeout,订阅读取包含 NextMsg、Header.Get、Header.Values 与 Data。PublishMsg 与 FlushTimeout 是连接操作,NextMsg 得到的消息同时包含 Header 和 Data;Get、Values 读取 Header 的不同视图。下图仅展示这些归属和静态关联,不表示业务确认顺序。

消息发送操作与订阅读取对象的静态关联
发送接口与读取对象分组示意。FlushTimeout 不等于业务确认,Header.Get 与 Header.Values 的读取范围不同。

如果后续要把示例改为请求回复,用双方定义的回复内容表达业务成功,并设置超时及幂等约定。如果需要持久化和发布确认,应独立选择 JetStream 的发布接口;仅给 Core NATS 消息加一个类似确认的自定义字段,不会改变投递保证。

集成到工程时容易遗漏的约定

场景推荐写法或处理核对点
一个追踪标识Header.Set双方使用同一键名与大小写
多个业务标签Add 写入,Values 读取Get 只给首值
手工构造 Msg初始化 Header mapnil map 不能写入
字段缺失或值为空按业务合同判断Get 返回空串不足以区分两种情况
发布返回成功继续按业务协议取得结果不能据此宣称消费者已完成
连接重建或能力变化保留发布错误处理启动时的能力判断不替代每次调用错误

需要区分字段不存在与空值时,可使用准确键名对 Header 做 map 查找,并检查返回的布尔值和切片内容。不要把空字符串当成必填字段已经通过校验。

每条消息创建自己的可写头部。若多个 goroutine 共用同一个 Header map 并修改,会引入普通 Go map 的并发访问问题;共用指针或浅拷贝并不能隔离其中的切片。需要复制时,逐个复制键和值切片,发布前完成字段构造,并避免在发布期间修改消息。

常见疑问与落地清单

能直接在 Publish 的 data 中加 JSON 字段吗?可以,那属于正文格式设计。是否放 Header 取决于元数据合同;消费者不能自动从 JSON 读取出 NATS Header。

收到没有头部的消息能读取吗?应允许 Header 为 nil,Get 会得到空串;业务若要求字段必须存在,应显式校验,不能直接写回 nil map。

最终需要保留哪些动作?先定义字段合同,再准备连接和订阅,使用 NewMsg 构造独立消息,检查 PublishMsg 错误,按需要刷新连接并读取消息,最后用业务自己的确认规则判断处理完成。关闭连接前完成所需发送与等待;本文 defer Close 负责结束示例连接,不是异步业务任务的完成保证。

技术依据:官方 Header API、官方 Go 客户端实现、NATS 客户端协议。字段命名、发票事件与业务边界示例为本文原创设计。

声明:本文转载于:17golang原创 如有侵犯,请联系study_golang@163.com删除
相关阅读
更多>
最新阅读
更多>
课程推荐
更多>