queue

package

v0.1.1 Latest Latest Go to latest Published: Jul 30, 2026 License: MIT Imports: 8 Imported by: 0

Details

Package queue 提供队列生命周期、结构化消息发布和解码能力。

This section is empty.

This section is empty.

func Decode[T any](message queueData.Message) (*T, error)

Decode 解析队列消息中的 data 字段，并兼容字符串、字节和内存对象载荷。

type Consumer struct {
	// Topic 是消费者监听的队列主题。
	Topic string
	// Handler 处理队列消息。
	Handler queueData.ConsumerFunc
}

Consumer 描述模块贡献的具名队列消费者。

type Publisher struct {
	// contains filtered or unexported fields
}

Publisher 将结构化数据发布到队列适配器。

func NewPublisher(queue kitQueue.Queue) (*Publisher, error)

NewPublisher 创建队列发布器。

func (p *Publisher) Publish(ctx context.Context, topic string, data any) error

Publish 将数据编码为 JSON 并发布到指定主题。

type Runtime struct {
	// contains filtered or unexported fields
}

Runtime 将队列消费者接入 Kratos 服务生命周期。

func NewRuntime(queue kitQueue.Queue) (*Runtime, error)

NewRuntime 创建队列运行时。

func (r *Runtime) Register(consumers ...Consumer) error

Register 注册队列消费者，并拒绝空主题、空处理器和重复主题。

func (r *Runtime) Start(context.Context) error

Start 启动队列消费循环。

func (r *Runtime) Stop(ctx context.Context) error

Stop 停止队列并等待消费循环退出。