Documentation
¶
Overview ¶
Package nats 提供 NATS 消息队列客户端封装。
基于 nats-io/nats.go，为 ekit 微服务提供统一的 NATS 接入。
核心特性 ¶
- 自动重连 + 重连回调
- Publish / Subscribe / QueueSubscribe
- Request / Reply（同步 RPC 模式）
- JSON 序列化/反序列化
- 优雅关闭（drain）
- JetStream：持久化消息 + 声明式流装配（EnsureStream，#122——流名/subjects/ 存储/容量缺省单源，CreateOrUpdateStream 语义 + 存量流冲突修复文案）
使用方式 ¶
// 连接
client, err := nats.NewClient("nats://localhost:4222",
nats.WithName("my-service"),
nats.WithReconnectWait(2*time.Second),
)
defer client.Close(context.Background()) // 0.x BREAKING（#83/#86）：原无参 Close() 移除
// 发布
err := client.Publish(ctx, "order.created", []byte(`{"id":1}`))
// 订阅（#167：Subscribe 无 ctx 参数，原文档示例多传了 ctx 照抄编译不过）
sub, err := client.Subscribe("order.created", func(msg *nats.Msg) {
log.Printf("received: %s", string(msg.Data))
})
defer sub.Unsubscribe()
// Request/Reply
resp, err := client.Request(ctx, "user.get", []byte(`{"id":1}`), 5*time.Second)
Index ¶
- Constants
- func QueueSubscribeJSON[T any](c *Client, subject, queue string, handler func(ctx context.Context, msg *T)) (*nats.Subscription, error)
- func RequestJSON[Req, Resp any](ctx context.Context, c *Client, subject string, req *Req, ...) (*Resp, error)
- func SubscribeJSON[T any](c *Client, subject string, handler func(ctx context.Context, msg *T)) (*nats.Subscription, error)
- type Client
- func (c *Client) Close(ctx context.Context) error
- func (c *Client) Conn() *nats.Conn
- func (c *Client) ConnectedUrl() string
- func (c *Client) Drain() error
- func (c *Client) IsConnected() bool
- func (c *Client) NewJetStream() (*JetStream, error)
- func (c *Client) Publish(ctx context.Context, subject string, data []byte) error
- func (c *Client) PublishJSON(ctx context.Context, subject string, v any) error
- func (c *Client) PublishJSONWithTraceID(ctx context.Context, subject string, v any) error
- func (c *Client) PublishWithTraceID(ctx context.Context, subject string, data []byte) error
- func (c *Client) QueueSubscribe(subject, queue string, handler func(*nats.Msg)) (*nats.Subscription, error)
- func (c *Client) Reply(subject string, handler func(msg *nats.Msg) []byte) (*nats.Subscription, error)
- func (c *Client) Request(ctx context.Context, subject string, data []byte, timeout time.Duration) (*nats.Msg, error)
- func (c *Client) Subscribe(subject string, handler func(*nats.Msg)) (*nats.Subscription, error)
- type JetStream
- func (js *JetStream) Consume(ctx context.Context, stream, consumer string, handler jetstream.MessageHandler, ...) (jetstream.ConsumeContext, error)
- func (js *JetStream) ConsumerInfo(ctx context.Context, stream, consumer string) (*jetstream.ConsumerInfo, error)
- func (js *JetStream) CreateOrUpdateConsumer(ctx context.Context, stream string, cfg jetstream.ConsumerConfig) (jetstream.Consumer, error)
- func (js *JetStream) CreateStream(ctx context.Context, name string, subjects ...string) error
- func (js *JetStream) DeleteStream(ctx context.Context, name string) error
- func (js *JetStream) EnsureStream(ctx context.Context, spec StreamSpec) (jetstream.Stream, error)
- func (js *JetStream) Fetch(ctx context.Context, stream, consumer string, batchSize int, ...) (jetstream.MessageBatch, error)
- func (js *JetStream) Messages(ctx context.Context, stream, consumer string) (jetstream.MessagesContext, error)
- func (js *JetStream) Next(ctx context.Context, stream, consumer string, opts ...jetstream.FetchOpt) (jetstream.Msg, error)
- func (js *JetStream) Publish(ctx context.Context, subject string, data []byte, opts ...jetstream.PublishOpt) (*jetstream.PubAck, error)
- func (js *JetStream) PublishAsync(subject string, data []byte, opts ...jetstream.PublishOpt) (jetstream.PubAckFuture, error)
- func (js *JetStream) PublishAsyncComplete() <-chan struct{}
- func (js *JetStream) PublishJSON(ctx context.Context, subject string, v any, opts ...jetstream.PublishOpt) (*jetstream.PubAck, error)
- func (js *JetStream) Stream(ctx context.Context, name string) (jetstream.Stream, error)
- func (js *JetStream) StreamInfo(ctx context.Context, stream string) (*jetstream.StreamInfo, error)
- type NatsOption
- func WithCloseHandler(fn func(conn *nats.Conn)) NatsOption
- func WithDisconnectHandler(fn func(conn *nats.Conn, err error)) NatsOption
- func WithMaxReconnects(n int) NatsOption
- func WithName(name string) NatsOption
- func WithReconnectHandler(fn func(conn *nats.Conn)) NatsOption
- func WithReconnectWait(d time.Duration) NatsOption
- type StreamSpec
Constants ¶
const ( // DefaultStreamMaxAge 是 EnsureStream 的缺省消息保留时长。 DefaultStreamMaxAge = 72 * time.Hour // DefaultStreamMaxBytes 是 EnsureStream 的缺省流字节上限（512 MB）。 DefaultStreamMaxBytes = 512 << 20 )
家族缺省常量（#122）：fuxi 三仓 310 行近同手搓的 stream 装配缺省值在此 单源化——某仓改 MaxAge 其余仓不跟随的漂移面就此收口。
Variables ¶
This section is empty.
Functions ¶
func QueueSubscribeJSON ¶
func QueueSubscribeJSON[T any](c *Client, subject, queue string, handler func(ctx context.Context, msg *T)) (*nats.Subscription, error)
QueueSubscribeJSON 订阅主题（队列模式）并自动反序列化 JSON 消息。
同 SubscribeJSON，从 NATS message header 中提取 trace_id 注入 ctx。
func RequestJSON ¶
func RequestJSON[Req, Resp any](ctx context.Context, c *Client, subject string, req *Req, timeout time.Duration) (*Resp, error)
RequestJSON 发送 JSON 请求并反序列化响应。 使用 context 控制超时和取消。 nil ctx 按 context.Background 处理（#101，与同文件 Publish 系列口径一致）。
⚠️ BREAKING（#88，0.x 阶段允许）：ctx 由第二位移至首位，与 Request 及 Go「ctx 首参」惯例对齐——旧形态 RequestJSON[Req, Resp](c, ctx, ...) 与 同文件 Request(ctx, ...) 形态不一致，IDE 补全/人工誊写极易把 subject 传进 ctx 位。亦与 mq.Requester.RequestJSON 的接口形态一致。
func SubscribeJSON ¶
func SubscribeJSON[T any](c *Client, subject string, handler func(ctx context.Context, msg *T)) (*nats.Subscription, error)
SubscribeJSON 订阅主题并自动反序列化 JSON 消息。
从 NATS message header 中提取 trace_id（如果存在）并注入到 handler 的 ctx， 保持分布式追踪在异步消息边界不断裂。Publisher 端需要在 Publish 时通过 PublishWithTraceID 注入 trace_id 才能形成完整链路。
Types ¶
type Client ¶
type Client struct {
// contains filtered or unexported fields
}
Client 是 NATS 客户端封装。
func NewClient ¶
func NewClient(urls string, opts ...NatsOption) (*Client, error)
NewClient 创建 NATS 客户端并连接。
urls 示例: "nats://localhost:4222" 或 "nats://host1:4222,nats://host2:4222"
func (*Client) Close ¶
Close 关闭连接：发起 Drain 并等待「真正排空完成、连接进入 CLOSED」—— nats.go 的 Drain() 是异步的（立即返回，排空在内部 goroutine 里等所有订阅 的在途消息处理完），一个 handler 阻塞时排空会挂到库的 DrainTimeout（默认 30s）甚至更久（#83/#86），关停随之无界。现在 ctx 到期仍未排空就放弃等待、 回退 conn.Close() 强制关闭（连接关闭会让阻塞中的读/写尽快以错误收敛）， 关停有界。Drain 被拒绝启动、而连接已被强制关闭（网络侧关闭、或重连窗口里 上游 Drain 自己关掉的）视为已达目标态，不算错误（#147）。
签名遵循矩阵硬规则「组件 Close 统一 Close(ctx) error + sync.Once 幂等」： 仅首次调用执行关闭动作并返回其结果，后续调用直接返回 nil。 nil ctx 按 context.Background() 处理（#79 口径）。
⚠️ BREAKING（0.x 允许）：原无参 Close() 移除——Go 不支持同名方法重载， 无法按 issue #83 的设想保留无参兼容别名；调用点改为 c.Close(context.Background())。需要不等待排空的立即关闭，用 c.Conn().Close()。
func (*Client) Drain ¶
Drain 发起排空并立即返回（异步）：nats.go 的 Drain() 只启动排空，不等待 完成（上游实现的考证见 Close 文档）。需要「等真正排空完成、连接进入 CLOSED」的同步语义（含超时兜底与强制关闭）用 Close(ctx)。 返回值仅反映排空是否成功启动（#168：原文档「等待所有消息处理完成后关闭」 与实现不符，已按真实行为修正）。
func (*Client) NewJetStream ¶
NewJetStream 创建 JetStream v2 客户端。
func (*Client) Publish ¶
Publish 发布消息到指定主题。 nil ctx 按 context.Background 处理（#79，与同仓 WaitN/Acquire/DoValue 的兜底口径一致）。守卫收敛见 publish（#162）。
func (*Client) PublishJSON ¶
PublishJSON 发布 JSON 编码的消息。 nil ctx 按 context.Background 处理（#79）。守卫收敛见 publish（#162）。
func (*Client) PublishJSONWithTraceID ¶
PublishJSONWithTraceID 发布 JSON 消息并从 ctx 提取 trace_id 注入 NATS message header。 配合 SubscribeJSON 自动恢复 trace_id 形成跨消息链路的完整追踪。 nil ctx 按 context.Background 处理（#79）。守卫收敛见 publish（#162）。
func (*Client) PublishWithTraceID ¶
PublishWithTraceID 发布消息并从 ctx 提取 trace_id 注入 NATS message header。 配合 SubscribeJSON 自动恢复 trace_id 形成跨消息链路的完整追踪。 ctx 中无 trace_id 时退化为普通 Publish，行为向后兼容。 nil ctx 按 context.Background 处理（#79）。守卫收敛见 publish（#162）。
func (*Client) QueueSubscribe ¶
func (c *Client) QueueSubscribe(subject, queue string, handler func(*nats.Msg)) (*nats.Subscription, error)
QueueSubscribe 订阅主题（队列模式），同一队列组内只有一个消费者收到消息。 适用于负载均衡场景：多个服务实例订阅同一主题，消息只投递给其中一个。
handler 的 panic 同样就地收编（#145，见 safeMsgHandler）。
func (*Client) Reply ¶
func (c *Client) Reply(subject string, handler func(msg *nats.Msg) []byte) (*nats.Subscription, error)
Reply 订阅主题并自动回复（用于实现 RPC 服务端）。 handler 的返回值会作为回复消息发送。
type JetStream ¶
type JetStream struct {
// contains filtered or unexported fields
}
JetStream 是 NATS JetStream v2 API 客户端封装。 使用新的 jetstream 包，提供持久化消息、至少一次投递保证、流和消费者管理。
func (*JetStream) Consume ¶
func (js *JetStream) Consume(ctx context.Context, stream, consumer string, handler jetstream.MessageHandler, opts ...jetstream.PullConsumeOpt) (jetstream.ConsumeContext, error)
Consume 消费消息（推模式，自动管理 ACK）。 返回 ConsumeContext，调用 Stop() 停止消费。
handler 的 panic 被就地收编（#166，口径同 Subscribe/#145）：Error 留痕， 消费者与进程存活；消息 fate 跟随消费者的 AckPolicy（见 safeJetMsgHandler）。
⚠️ ctx 只界「查/建消费者」阶段：上游 Consume 不收 ctx，消费一旦启动 **只由返回的 ConsumeContext 的 Stop()/Drain() 停止**（#103 文档化口径， SDK 限制）。nil ctx 按 context.Background 处理（#101）。
func (*JetStream) ConsumerInfo ¶
func (js *JetStream) ConsumerInfo(ctx context.Context, stream, consumer string) (*jetstream.ConsumerInfo, error)
ConsumerInfo 获取消费者信息。 nil ctx 按 context.Background 处理（#101）。
func (*JetStream) CreateOrUpdateConsumer ¶
func (js *JetStream) CreateOrUpdateConsumer(ctx context.Context, stream string, cfg jetstream.ConsumerConfig) (jetstream.Consumer, error)
CreateOrUpdateConsumer 创建或更新消费者。 nil ctx 按 context.Background 处理（#101）。
func (*JetStream) CreateStream ¶
CreateStream 创建流（如果不存在）。 仅在流确实不存在（ErrStreamNotFound）时创建； 瞬时错误原样返回，避免误判掩盖真实故障。 nil ctx 按 context.Background 处理（#101，与 Client 系列兜底口径一致）。
func (*JetStream) DeleteStream ¶
DeleteStream 删除流。 nil ctx 按 context.Background 处理（#101）。
func (*JetStream) EnsureStream ¶ added in v0.8.0
EnsureStream 声明式流装配（#122）：确保一条业务流按 StreamSpec 存在—— 不存在则创建，已存在则按 spec 原地更新（上游 CreateOrUpdateStream 语义）。
与 CreateStream 的关系：CreateStream 是「仅当不存在时创建」的窄原语（流已 存在时不动配置、无缺省派生）；EnsureStream 是声明式组合口——缺省派生 （流名/subjects/Storage/Retention/MaxAge/MaxBytes，见 StreamSpec）+ CreateOrUpdateStream + 存量流冲突修复文案，fuxi 三仓 310 行近同手搓的装配 知识在此单源。
WorkQueue 存量冲突：服务端对存量流的若干维度不可原地变更（retention 到/ 从 workqueue、storage 类型变更、sealed 流等）。此类错误会附「需手动删除后 重启」的修复文案——删除会丢弃未消费消息，先确认消费水位。新增/删除 subject 对 WorkQueue 存量流同样受限（部分版本服务端直接拒绝），冲突时同样按文案处置。
nil ctx 按 context.Background 处理（#101，与 JetStream 系列兜底口径一致）。
func (*JetStream) Fetch ¶
func (js *JetStream) Fetch(ctx context.Context, stream, consumer string, batchSize int, opts ...jetstream.FetchOpt) (jetstream.MessageBatch, error)
Fetch 获取指定数量的消息（拉模式）。
ctx 界住**整个调用**（#103）：既界消费者查找，也经 jetstream.FetchContext 注入取消息阶段——ctx 取消/到期时取消息立即终止，不再挂在 SDK 的 DefaultExpires（30s）上。delivery 阶段的取消原因经 batch.Error() 返回。
例外：调用方显式给 FetchMaxWait 时，上游禁止与 FetchContext 同用 （ErrInvalidOption），此时以显式超时为准回退无 ctx 形态（SDK 限制， 取消息阶段最多等 FetchMaxWait）——见 fetchWithContext。 nil ctx 按 context.Background 处理（#101）。
func (*JetStream) Messages ¶
func (js *JetStream) Messages(ctx context.Context, stream, consumer string) (jetstream.MessagesContext, error)
Messages 创建消息迭代器（拉模式，持续消费）。
⚠️ ctx 只界「查/建消费者」阶段：上游 Messages 不收 ctx，迭代器一旦创建 只由返回的 MessagesContext 的 Stop()/Drain() 停止（#103 文档化口径， SDK 限制）。nil ctx 按 context.Background 处理（#101）。
func (*JetStream) Next ¶
func (js *JetStream) Next(ctx context.Context, stream, consumer string, opts ...jetstream.FetchOpt) (jetstream.Msg, error)
Next 获取下一条消息（拉模式，阻塞直到超时）。
ctx 界住**整个调用**（#103）：同 Fetch，经 jetstream.FetchContext 注入 取消息阶段，取消不再等 30s 过期。例外口径同 Fetch（FetchMaxWait 优先）。 nil ctx 按 context.Background 处理（#101）。
func (*JetStream) Publish ¶
func (js *JetStream) Publish(ctx context.Context, subject string, data []byte, opts ...jetstream.PublishOpt) (*jetstream.PubAck, error)
Publish 发布消息到 JetStream 流。 nil ctx 按 context.Background 处理（#79，与 Client.Publish 口径一致）。
func (*JetStream) PublishAsync ¶
func (js *JetStream) PublishAsync(subject string, data []byte, opts ...jetstream.PublishOpt) (jetstream.PubAckFuture, error)
PublishAsync 异步发布消息（批量发布，提高吞吐量）。
func (*JetStream) PublishAsyncComplete ¶
func (js *JetStream) PublishAsyncComplete() <-chan struct{}
PublishAsyncComplete 等待所有异步发布完成。
func (*JetStream) PublishJSON ¶
func (js *JetStream) PublishJSON(ctx context.Context, subject string, v any, opts ...jetstream.PublishOpt) (*jetstream.PubAck, error)
PublishJSON 发布 JSON 消息到 JetStream 流。 nil ctx 按 context.Background 处理（#79）。
func (*JetStream) StreamInfo ¶
StreamInfo 获取流信息。 nil ctx 按 context.Background 处理（#101）。
type NatsOption ¶
type NatsOption func(*options)
NatsOption 配置 Client 的可选参数。
func WithCloseHandler ¶
func WithCloseHandler(fn func(conn *nats.Conn)) NatsOption
WithCloseHandler 设置连接关闭回调。
func WithDisconnectHandler ¶
func WithDisconnectHandler(fn func(conn *nats.Conn, err error)) NatsOption
WithDisconnectHandler 设置断开连接回调。
func WithMaxReconnects ¶
func WithMaxReconnects(n int) NatsOption
WithMaxReconnects 设置最大重连次数，-1 表示无限重连（默认）。
func WithReconnectHandler ¶
func WithReconnectHandler(fn func(conn *nats.Conn)) NatsOption
WithReconnectHandler 设置重连成功回调。
func WithReconnectWait ¶
func WithReconnectWait(d time.Duration) NatsOption
WithReconnectWait 设置重连等待时间，默认 2 秒。
type StreamSpec ¶ added in v0.8.0
type StreamSpec struct {
// Name 流名。空值时从客户端名（WithName）派生：非字母数字折叠为 '_'、
// 统一大写、补 "_EVENTS" 后缀（fuxi-engine → FUXI_ENGINE_EVENTS，与
// 三仓既成缺省同形）。客户端也未设 WithName 时显式报错——流名是持久化
// 对象的主键，不静默兜底。
Name string
// Subjects 流的 subject 通配集。空值时取 Name+".>"（三仓一致的缺省形态）。
Subjects []string
// Storage 存储策略。零值取 FileStorage（三仓一致）。⚠ nats.go 的
// MemoryStorage 与零值同为 0、无法与「未设置」区分——内存流不是本原语
// 的目标场景，需要时直接用 CreateOrUpdateStream 传完整 StreamConfig。
Storage jetstream.StorageType
// Retention 保留策略。零值即 LimitsPolicy（与库缺省一致）；显式
// InterestPolicy / WorkQueuePolicy 直接生效。WorkQueue 存量流的不可原地
// 更新冲突见 EnsureStream 的处置文案。
Retention jetstream.RetentionPolicy
// MaxAge 消息保留时长。零值取 DefaultStreamMaxAge。
MaxAge time.Duration
// MaxBytes 流字节上限。零值取 DefaultStreamMaxBytes；服务器原生「0=不限」
// 的语义在本原语不可达（0 被解释为未设置），需要时用 CreateOrUpdateStream。
MaxBytes int64
}
StreamSpec 是 EnsureStream 的声明式流规格（#122）。
零值字段一律取库缺省——消费方只写与缺省不同的字段；fuxi 三仓的近同手搓 （stream 名缺省、subjects 通配格式、Storage/Retention、MaxAge/MaxBytes） 逐字段收敛到 StreamConfig 的映射规则见各字段文档。
Source Files