联犀开发者文档

消息队列与事件总线

约 927 字大约 3 分钟

2025-03-03

平台通过 share/eventBus.FastEvent 统一发布和订阅事件，后端可按部署配置使用 NATS 等传输。服务间事件与面向浏览器的 WebSocket 是不同层，不能混用主题格式。

广播还是队列

API适合场景多实例行为
Subscribe缓存失效、各实例均需处理的通知各订阅实例分别接收
QueueSubscribe任务处理、同业务工作池分担同队列组由一个订阅者接收
带 GroupID 的订阅方法需要明确区分消费组按指定组组织

默认队列组与 serverName 有关。两个不同业务若需要各自处理一次，不能误用同一组；多个副本希望分担工作时，也不能给每个副本随机生成不同组。

初始化、订阅、启动

以下函数展示生命周期，传入已解析配置、稳定服务名及节点编号：

// 本文件装配演示事件订阅与启动流程。
package startup

import (
    "context"
    "encoding/json"
    "time"
    "gitee.com/unitedrhino/share/conf"
    "gitee.com/unitedrhino/share/eventBus"
)

// Changed 是演示业务事件，EventID 供业务侧去重。
type Changed struct {
    EventID string `json:"eventID"` // 同一业务事件的稳定编号。
    ID      string `json:"id"`      // 业务对象编号，字符串保留精度。
}

// StartEvents 在注册处理器后启动消费；handle 负责业务执行和幂等。
func StartEvents(c conf.EventConf, serverName string, nodeID int64,
    handle func(context.Context, Changed) error) (*eventBus.FastEvent, error) {
    bus, err := eventBus.NewFastEvent(c, serverName, nodeID)
    if err != nil { return nil, err }
    err = bus.QueueSubscribe("demo.record.changed", func(ctx context.Context, _ time.Time, body []byte) error {
        var msg Changed // FastEvent 回调收到的是已解封的业务 JSON。
        if err := json.Unmarshal(body, &msg); err != nil { return err }
        return handle(ctx, msg)
    })
    if err != nil { return nil, err }
    if err := bus.Start(); err != nil { return nil, err }
    return bus, nil
}

生产进程中 NewFastEvent 是单例初始化。第一次失败会保留错误，后续调用不会自动重新连接初始化；必须检查错误并按启动策略退出/恢复，不能忽略错误继续使用。

Publish(ctx, topic, event) 负责序列化业务对象和附加事件信封。发布之前应确认订阅端已就绪；启动入口复用全局总线时，不要重复装配彼此冲突的服务配置。

消息信封与跨语言

底层 events.MsgHead 当前结构为：

{
  "trace": "<序列化追踪上下文>",
  "timestamp": 1792112400000,
  "data": "{\"eventID\":\"sample-1\",\"id\":\"123\"}",
  "userCtx": {}
}

外层 timestamp 是数字毫秒；data 是 JSON 文本字符串。使用 FastEvent 回调时已经解封，不能再按旧文档做 base64 解码。只有直接接底层传输的受信服务才需实现信封，内部 userCtx 不作为外部认证凭证。

投递与失败边界

普通事件发布成功不等于业务已执行，也不等于消息已持久化。生产回调异步执行，不能一概认为 handler 返回错误就能自动触发重投。

需要可靠任务时，明确选择部署支持的持久消费机制，并核对确认、重投、超时及死信策略。业务提供稳定事件 ID、去重记录和可重入处理；数据库提交后需要必达通知时使用持久发件箱，不能只靠一次 Publish。

验证与排查

验证一个实例广播、两个实例同组分担、不同组各自消费、启动顺序、重复消息及依赖中断。排查按配置模式 → broker 权限 → 完整主题 → 队列组 → Start 是否调用 → 解码与处理错误定位。

设备消息的具体主题及载荷见设备消息流转，缓存通知见缓存。

更新日志

2026/10/16 19:46
查看所有更新日志
  • b3b4d-docs(平台开发): 恢复有效专题并更新当前实现示例于