Documentation
¶
Overview ¶
Package nats 提供 NATS 消息队列客户端封装。
基于 nats-io/nats.go，为 ekit 微服务提供统一的 NATS 接入。
核心特性 ¶
- 自动重连 + 重连回调
- Publish / Subscribe / QueueSubscribe
- Request / Reply（同步 RPC 模式）
- JSON 序列化/反序列化
- 优雅关闭（drain）
使用方式 ¶
// 连接
client, err := nats.NewClient("nats://localhost:4222",
nats.WithName("my-service"),
nats.WithReconnectWait(2*time.Second),
)
defer client.Close()
// 发布
err := client.Publish(ctx, "order.created", []byte(`{"id":1}`))
// 订阅
sub, err := client.Subscribe(ctx, "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 ¶
- func QueueSubscribeJSON[T any](c *Client, subject, queue string, handler func(ctx context.Context, msg *T)) (*nats.Subscription, error)
- func RequestJSON[Req, Resp any](c *Client, ctx context.Context, 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()
- 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) 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
Constants ¶
This section is empty.
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](c *Client, ctx context.Context, subject string, req *Req, timeout time.Duration) (*Resp, error)
RequestJSON 发送 JSON 请求并反序列化响应。 使用 context 控制超时和取消。
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 ¶
func (c *Client) Close()
Close 关闭连接（先排空订阅，再关闭）。 如果需要立即关闭（不等待消息处理完成），直接调用 c.conn.Close()。 多次调用是安全的（sync.Once 保证只执行一次）。
func (*Client) NewJetStream ¶
NewJetStream 创建 JetStream v2 客户端。
func (*Client) Publish ¶
Publish 发布消息到指定主题。 nil ctx 按 context.Background 处理（#79，与同仓 WaitN/Acquire/DoValue 的兜底口径一致）。
func (*Client) PublishJSON ¶
PublishJSON 发布 JSON 编码的消息。 nil ctx 按 context.Background 处理（#79）。
func (*Client) PublishJSONWithTraceID ¶
PublishJSONWithTraceID 发布 JSON 消息并从 ctx 提取 trace_id 注入 NATS message header。 配合 SubscribeJSON 自动恢复 trace_id 形成跨消息链路的完整追踪。 nil ctx 按 context.Background 处理（#79）。
func (*Client) PublishWithTraceID ¶
PublishWithTraceID 发布消息并从 ctx 提取 trace_id 注入 NATS message header。 配合 SubscribeJSON 自动恢复 trace_id 形成跨消息链路的完整追踪。 ctx 中无 trace_id 时退化为普通 Publish，行为向后兼容。 nil ctx 按 context.Background 处理（#79）。
func (*Client) QueueSubscribe ¶
func (c *Client) QueueSubscribe(subject, queue string, handler func(*nats.Msg)) (*nats.Subscription, error)
QueueSubscribe 订阅主题（队列模式），同一队列组内只有一个消费者收到消息。 适用于负载均衡场景：多个服务实例订阅同一主题，消息只投递给其中一个。
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() 停止消费。
func (*JetStream) ConsumerInfo ¶
func (js *JetStream) ConsumerInfo(ctx context.Context, stream, consumer string) (*jetstream.ConsumerInfo, error)
ConsumerInfo 获取消费者信息。
func (*JetStream) CreateOrUpdateConsumer ¶
func (js *JetStream) CreateOrUpdateConsumer(ctx context.Context, stream string, cfg jetstream.ConsumerConfig) (jetstream.Consumer, error)
CreateOrUpdateConsumer 创建或更新消费者。
func (*JetStream) CreateStream ¶
CreateStream 创建流（如果不存在）。 仅在流确实不存在（ErrStreamNotFound）时创建； 瞬时错误原样返回，避免误判掩盖真实故障。
func (*JetStream) DeleteStream ¶
DeleteStream 删除流。
func (*JetStream) Fetch ¶
func (js *JetStream) Fetch(ctx context.Context, stream, consumer string, batchSize int, opts ...jetstream.FetchOpt) (jetstream.MessageBatch, error)
Fetch 获取指定数量的消息（拉模式）。
func (*JetStream) Messages ¶
func (js *JetStream) Messages(ctx context.Context, stream, consumer string) (jetstream.MessagesContext, error)
Messages 创建消息迭代器（拉模式，持续消费）。
func (*JetStream) Next ¶
func (js *JetStream) Next(ctx context.Context, stream, consumer string, opts ...jetstream.FetchOpt) (jetstream.Msg, error)
Next 获取下一条消息（拉模式，阻塞直到超时）。
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 获取流信息。
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 秒。
Source Files