Help us shape the future of Go! Take the 2026 Go Developer Survey

centrifugeplus

package module
v0.0.0-...-4f86317 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Aug 23, 2026 License: MIT Imports: 19 Imported by: 0

README

centrifuge-plus

基于 centrifuge v0.38.0 的增强库，核心提供 DualBroker——一个实现了 centrifuge Broker 接口的路由器，根据频道类型将消息分发到不同的底层 broker。

核心概念

两种频道模式
模式 持久化 实时推送 适用场景
Live 否 是 在线状态、打字机效果等高频低延迟场景，用户离线即丢失
Topic 是 是 聊天记录、通知等需要可靠投递的场景，离线用户上线后可恢复

频道路由通过注册指定，不通过频道名前缀。接入者在 centrifuge 的 OnSubscribe 回调中调用 DualBroker.RegisterChannelType 注册频道类型：

client.OnSubscribe(func(e centrifuge.SubscribeEvent, cb centrifuge.SubscribeCallback) {
    if isTopicChannel(e.Channel) {
        broker.RegisterChannelType(e.Channel, centrifugeplus.Topic)
        cb(centrifuge.SubscribeReply{
            Options: centrifuge.SubscribeOptions{EnableRecovery: true},
        }, nil)
    } else {
        broker.RegisterChannelType(e.Channel, centrifugeplus.Live)
        cb(centrifuge.SubscribeReply{}, nil)
    }
})
设计原则

centrifuge-plus 不管理用户系统、频道订阅关系、业务逻辑。仅提供配置和接口，让接入者自行实现。

DualBroker

DualBroker 实现 centrifuge 的 Broker 接口，内部持有两个底层 broker，根据频道类型路由：

  • Live 频道 → 委托给 centrifuge 内置的 RedisBroker
  • Topic 频道 → 委托给自定义的 TopicBroker
配置
type DualBrokerConfig struct {
    Live  centrifuge.RedisBrokerConfig
    Topic TopicBrokerConfig
}
使用方式
node, err := centrifuge.New(centrifuge.Config{})
if err != nil {
    log.Fatalf("创建 node 失败: %v", err)
}

broker, err := NewDualBroker(node, DualBrokerConfig{
    Live: centrifuge.RedisBrokerConfig{
        Prefix: "live",
        Shards: []*centrifuge.RedisShard{redisShard},
    },
    Topic: TopicBrokerConfig{
        Prefix:       "topic",
        RedisAddr:    "localhost:6379",
        HistoryStore: myHistoryStore,
    },
})
if err != nil {
    log.Fatalf("创建 DualBroker 失败: %v", err)
}

node.SetBroker(broker)

Topic 模式

先落盘再推送模型

Topic 模式采用"先落盘再推送"架构，确保数据一致性：

BatchIncrby → DB 事务写入 → PublishWithOffset
(预分配offset)   (持久化)       (尽力推送)

关键原则：数据先落盘，再推送。推送是尽力而为（best-effort），失败不影响数据一致性。

接入流程

重要：Topic 消息应通过服务端 RPC handler发送，而非客户端 centrifuge publish。 客户端通过 RPC 请求发送消息，服务端 handler 执行 BatchIncrby → DB 事务 → PublishWithOffset。 不要使用 OnPublish 回调实现"先落盘再推送"—— centrifuge 在 OnPublish 后还会调用 Broker.Publish()，导致重复发布。

// === 服务端 RPC handler（正确处理 Topic 消息）===

// 客户端通过 RPC 方法（如 "send.topic"）发送消息
client.OnRPC(func(e centrifuge.RPCEvent, cb centrifuge.RPCCallback) {
    switch e.Method {
    case "send.topic":
        handleSendTopic(ctx, broker, historyStore, client, e, cb)
    default:
        cb(centrifuge.RPCReply{}, centrifuge.ErrorMethodNotFound)
    }
})

func handleSendTopic(ctx context.Context, broker *centrifugeplus.DualBroker,
    historyStore HistoryStore, client *centrifuge.Client,
    e centrifuge.RPCEvent, cb centrifuge.RPCCallback) {

    var req struct {
        Channel string `json:"channel"`
        Data    string `json:"data"`
    }
    if err := json.Unmarshal(e.Data, &req); err != nil {
        cb(centrifuge.RPCReply{}, err)
        return
    }

    // Step 1: 预分配 offset（原子 HINCRBY）
    positions, err := broker.BatchIncrby(ctx, []centrifugeplus.ChannelIncrbyRequest{
        {Channel: req.Channel},
    })
    if err != nil {
        cb(centrifuge.RPCReply{}, err)
        return
    }
    sp := positions[req.Channel]

    // Step 2: DB 事务（接入方自行实现）
    //   db.Begin() → Create(message) → Create(user_updates) → Commit()
    //   如果事务失败，预分配的 offset 产生 gap，客户端跳过该 offset 继续拉取
    if err := db.Transaction(func(tx *gorm.DB) error {
        if err := tx.Create(&Message{Channel: req.Channel, Data: req.Data, Offset: sp.Offset}).Error; err != nil {
            return err
        }
        if err := tx.Create(&UserUpdate{UserID: client.UserID(), Offset: sp.Offset}).Error; err != nil {
            return err
        }
        return nil
    }); err != nil {
        cb(centrifuge.RPCReply{}, err)
        return
    }

    // Step 3: 事务后推送（best-effort，失败不影响数据一致性）
    broker.PublishWithOffset(ctx, req.Channel, []byte(req.Data), centrifuge.PublishOptions{}, sp)

    cb(centrifuge.RPCReply{Data: []byte(fmt.Sprintf(`{"offset":%d}`, sp.Offset))}, nil)
}
TopicBroker

自定义 broker，提供以下核心方法：

方法 用途
BatchIncrby(ctx, reqs) 预分配 offset，DB 事务前调用
PublishWithOffset(ctx, ch, data, opts, sp) 使用预分配 offset 推送，DB 事务后调用
Publish(ch, data, opts) 便捷方法，内部自动 BatchIncrby + PublishWithOffset
History(ch, opts) 从 HistoryStore 读取历史，StreamPosition 从 Redis meta 读取

Publish 是便捷方法，适用于不需要 DB 事务的场景（如简单实时通知）。IM 场景应使用 BatchIncrby + DB 事务 + PublishWithOffset。

边缘场景
场景 处理方式
BatchIncrby 成功，DB 回滚 offset 产生 gap，客户端跳过该 offset 继续拉取
DB 成功，PublishWithOffset 失败 数据在 DB，客户端通过主动拉取发现
Epoch 变化 PublishWithOffset 返回 error，接入方可重试
网络重试 幂等缓存（result_key）防止重复消息
DualBroker 扩展方法

DualBroker 暴露 Topic 模式的扩展方法，路由到内部的 TopicBroker：

func (d *DualBroker) BatchIncrby(ctx context.Context, reqs []ChannelIncrbyRequest) (map[string]centrifuge.StreamPosition, error)
func (d *DualBroker) PublishWithOffset(ctx context.Context, ch string, data []byte, opts centrifuge.PublishOptions, sp centrifuge.StreamPosition) error

HistoryStore

接入者实现此接口，提供历史消息的查询能力。是历史消息的唯一数据源。

type HistoryStore interface {
    Query(ctx context.Context, channel string, sinceOffset uint32, latestOffset uint32) ([]*centrifuge.Publication, error)
}

Lua 脚本

脚本 用途
lua/incrby_offset.lua 批量预分配 offset，原子执行多个 HINCRBY
lua/publish_with_offset.lua 使用预分配 offset 发布到 PUB/SUB，校验 epoch 一致性

分布式链路追踪

centrifuge-plus 内置可选的 OpenTelemetry 分布式追踪支持，零值配置下无开销。

追踪配置
type TracingConfig struct {
    Enabled  bool
    Provider trace.TracerProvider  // nil 时使用 noop provider
}

关闭时（零值），所有埋点使用 trace.NewNoopTracerProvider()，零开销。

Span 命名
  • Span: centrifugeplus.{组件}.{operation}（如 centrifugeplus.topicbroker.publish）
  • Attribute: centrifugeplus.{属性}（如 centrifugeplus.channel）
跨边界传播

PUB/SUB 消息尾缀携带 W3C traceparent：__p1:{offset}:{epoch}:{data_len}__{data}[__tp:{traceparent}]

所有 Span 类型
Span 触发点
centrifugeplus.dualbroker.publish Publish 路由
centrifugeplus.topicbroker.publish Publish 便捷方法
centrifugeplus.topicbroker.batch_incrby BatchIncrby
centrifugeplus.topicbroker.batch_incrby.lua BatchIncrby Lua 脚本
centrifugeplus.topicbroker.publish_with_offset PublishWithOffset
centrifugeplus.topicbroker.publish_with_offset.lua PublishWithOffset Lua 脚本
centrifugeplus.topicbroker.subscribe Subscribe
centrifugeplus.topicbroker.unsubscribe Unsubscribe
centrifugeplus.topicbroker.publish_join PublishJoin
centrifugeplus.topicbroker.publish_leave PublishLeave
centrifugeplus.topicbroker.pubsub PUB/SUB 消息接收
centrifugeplus.topicbroker.history History 查询
Jaeger 验证

示例程序默认上报 span 至 Jaeger（localhost:4317 gRPC）。启动：

docker compose -f deploy/dev/docker-compose.dependencies.yml up -d jaeger
open http://localhost:16686  # Jaeger UI

完整示例

参见 example/main.go，演示了：

  1. Live 消息：实时推送，不持久化
  2. Topic 消息：先落盘再推送（BatchIncrby → Save → PublishWithOffset）
  3. 离线恢复：客户端断线重连后通过主动拉取 DB 恢复消息
  4. 流式文本：逐词发送模拟打字机效果

运行方式：

# 确保本地 Redis 运行
redis-server

# 确保 Jaeger 运行（用于分布式追踪）
docker compose -f deploy/dev/docker-compose.dependencies.yml up -d jaeger

# 启动示例
cd centrifuge-plus/example && go run .

项目结构

centrifuge-plus/
├── channel_type.go          # ChannelType 枚举
├── dual_broker.go           # DualBroker 实现
├── dual_broker_config.go    # DualBrokerConfig
├── topic_broker.go          # TopicBroker 实现
├── topic_broker_config.go   # TopicBrokerConfig
├── history_store.go         # HistoryStore 接口
├── logger.go                # Logger 接口
├── tracer.go                # TracingConfig + W3C traceparent 编解码
├── lua/
│   ├── incrby_offset.lua        # 批量预分配 offset
│   └── publish_with_offset.lua  # 使用预分配 offset 发布
└── example/                 # 完整示例程序

Documentation

Index

Constants

View Source 
const (
	AttributeChannel         = attribute.Key("centrifugeplus.channel")
	AttributeChannelType     = attribute.Key("centrifugeplus.channel_type")
	AttributeOffset          = attribute.Key("centrifugeplus.offset")
	AttributeEpoch           = attribute.Key("centrifugeplus.epoch")
	AttributeTaskID          = attribute.Key("centrifugeplus.task_id")
	AttributeEnqueued        = attribute.Key("centrifugeplus.enqueued")
	AttributeFromCache       = attribute.Key("centrifugeplus.from_cache")
	AttributeQueue           = attribute.Key("centrifugeplus.queue")
	AttributeMessageType     = attribute.Key("centrifugeplus.message_type")
	AttributeStreamMinOffset = attribute.Key("centrifugeplus.stream_min_offset")
)

Span attribute keys for centrifuge-plus tracing.

Variables

This section is empty.

Functions

This section is empty.

Types

type ChannelIncrbyRequest 

type ChannelIncrbyRequest struct {
	Channel string
	Count   int
}

ChannelIncrbyRequest represents a channel to pre-allocate offsets for. Count specifies how many consecutive offsets to reserve (default 1 if zero). The returned StreamPosition for the channel carries the HIGHEST allocated offset; callers derive the full range as [final - Count + 1, ..., final]. Each channel must appear at most once per BatchIncrby call.

type ChannelType 

type ChannelType int

ChannelType defines the type of a channel. 

const (
	// Live channels are not persisted, messages are only delivered to online subscribers.
	// Uses centrifuge's built-in RedisBroker.
	Live ChannelType = iota

	// Topic channels are persisted and support real-time push.
	// Messages are stored in history stream and also pushed to asynq for offline recovery.
	Topic
)

func (ChannelType) String 

func (ct ChannelType) String() string

String returns the string representation of the ChannelType.

type DualBroker 

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

DualBroker implements centrifuge.Broker interface. It routes messages to either RedisBroker (Live) or TopicBroker (Topic) based on channel type.

func NewDualBroker 

func NewDualBroker(node *centrifuge.Node, config DualBrokerConfig) (*DualBroker, error)

NewDualBroker creates a new DualBroker instance. The node parameter is required for RedisBroker initialization.

func (*DualBroker) BatchIncrby 

func (d *DualBroker) BatchIncrby(ctx context.Context, reqs []ChannelIncrbyRequest) (map[string]centrifuge.StreamPosition, error)

BatchIncrby batch pre-allocates offsets for multiple channels. Routes to TopicBroker.

func (*DualBroker) Close 

func (d *DualBroker) Close(ctx context.Context) error

Close closes both brokers, aggregating any errors.

func (*DualBroker) History 

func (d *DualBroker) History(ch string, opts centrifuge.HistoryOptions) (pubs []*centrifuge.Publication, sp centrifuge.StreamPosition, err error)

History returns publications for channel.

func (*DualBroker) IncrConversationOffset 

func (d *DualBroker) IncrConversationOffset(ctx context.Context, conversationID string) (uint32, error)

IncrConversationOffset 为指定会话分配下一个 conversation offset

func (*DualBroker) Publish 

func (d *DualBroker) Publish(ch string, data []byte, opts centrifuge.PublishOptions) (centrifuge.StreamPosition, bool, error)

Publish publishes data to channel.

func (*DualBroker) PublishEphemeral 

func (d *DualBroker) PublishEphemeral(ctx context.Context, ch string, data []byte, opts centrifuge.PublishOptions) error

PublishEphemeral 强制走 liveBroker，纯 PUB/SUB，不持久化到 stream。 用于 typing 等瞬态事件：丢失可接受，不需要 offset 和 recovery。

func (*DualBroker) PublishJoin 

func (d *DualBroker) PublishJoin(ch string, info *centrifuge.ClientInfo) error

PublishJoin publishes Join message to channel.

func (*DualBroker) PublishLeave 

func (d *DualBroker) PublishLeave(ch string, info *centrifuge.ClientInfo) error

PublishLeave publishes Leave message to channel.

func (*DualBroker) PublishWithContext 

func (d *DualBroker) PublishWithContext(ctx context.Context, ch string, data []byte, opts centrifuge.PublishOptions) (sp centrifuge.StreamPosition, fromCache bool, err error)

PublishWithContext is like Publish but accepts a context for distributed tracing.

func (*DualBroker) PublishWithOffset 

func (d *DualBroker) PublishWithOffset(ctx context.Context, ch string, data []byte, opts centrifuge.PublishOptions, sp centrifuge.StreamPosition) error

PublishWithOffset publishes a message using a pre-allocated offset. Routes to TopicBroker.

func (*DualBroker) RegisterBrokerEventHandler 

func (d *DualBroker) RegisterBrokerEventHandler(handler centrifuge.BrokerEventHandler) error

RegisterBrokerEventHandler is called once when Broker is set to Node.

func (*DualBroker) RegisterChannelType 

func (d *DualBroker) RegisterChannelType(ch string, ct ChannelType)

RegisterChannelType registers the channel type for a given channel.

func (*DualBroker) RemoveHistory 

func (d *DualBroker) RemoveHistory(ch string) error

RemoveHistory removes history from channel.

func (*DualBroker) SetConversationOffset 

func (d *DualBroker) SetConversationOffset(ctx context.Context, conversationID string, value uint32) error

SetConversationOffset 设置指定会话的 offset 值，用于 fork 等场景初始化计数器

func (*DualBroker) Subscribe 

func (d *DualBroker) Subscribe(ch string) error

Subscribe subscribes node to channel. 注意：接口签名不含 context 参数（与 centrifuge.NodeBroker 保持一致）， 此处 context.Background() 仅用于 tracing span 创建，不影响业务逻辑。

func (*DualBroker) TopicBroker 

func (d *DualBroker) TopicBroker() *TopicBroker

TopicBroker returns the underlying TopicBroker used for Topic mode channels.

func (*DualBroker) Unsubscribe 

func (d *DualBroker) Unsubscribe(ch string) error

Unsubscribe unsubscribes node from channel. 注意：接口签名不含 context 参数（与 centrifuge.NodeBroker 保持一致）， 此处 context.Background() 仅用于 tracing span 创建，不影响业务逻辑。

type DualBrokerConfig 

type DualBrokerConfig struct {
	// Live is the configuration for Live mode channels (using RedisBroker).
	Live centrifuge.RedisBrokerConfig

	// Topic is the configuration for Topic mode channels (using TopicBroker).
	Topic TopicBrokerConfig
}

DualBrokerConfig is the configuration for DualBroker.

type HistoryStore 

type HistoryStore interface {
	// Query retrieves publications from the given channel since the specified offset.
	// latestOffset 是当前 stream 的最新 offset，用于 fillGapPublications 补齐尾部缺口，
	// 保证恢复结果满足 centrifuge 的连续性检查（末条 pub.Offset == latestOffset）。
	Query(ctx context.Context, channel string, sinceOffset uint32, latestOffset uint32) ([]*centrifuge.Publication, error)
}

HistoryStore is the interface for querying message history. Implementations are provided by the integrator.

type HistoryStoreRemover 

type HistoryStoreRemover interface {
	// RemoveHistory removes all history for the given channel.
	RemoveHistory(channel string) error
}

HistoryStoreRemover is an optional interface that HistoryStore implementations can implement to support removing history for a channel.

type Logger 

type Logger interface {
	Info(msg string, args ...any)
	Warn(msg string, args ...any)
	Error(msg string, args ...any)
}

Logger defines the logging interface used by AsynqBroker.

type TopicBroker 

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

TopicBroker implements centrifuge.Broker interface for Topic mode channels. It uses a "persist first, then push" model: BatchIncrby → DB transaction → PublishWithOffset.

func NewTopicBroker 

func NewTopicBroker(config TopicBrokerConfig) (*TopicBroker, error)

NewTopicBroker creates a new TopicBroker instance.

func (*TopicBroker) BatchIncrby 

func (b *TopicBroker) BatchIncrby(ctx context.Context, reqs []ChannelIncrbyRequest) (map[string]centrifuge.StreamPosition, error)

BatchIncrby batch pre-allocates offsets for multiple channels. Call this before DB transaction. Returns map[channel]StreamPosition where the offset is the HIGHEST allocated offset for that channel; the full reserved range is [final - Count + 1, ..., final] (Count defaults to 1 if zero). Each channel must appear at most once in reqs.

func (*TopicBroker) Close 

func (b *TopicBroker) Close(_ context.Context) error

Close closes the broker and releases resources.

func (*TopicBroker) History 

func (b *TopicBroker) History(ch string, opts centrifuge.HistoryOptions) (pubs []*centrifuge.Publication, sp centrifuge.StreamPosition, err error)

History returns publications for channel from HistoryStore. StreamPosition is read from Redis meta key.

func (*TopicBroker) IncrConversationOffset 

func (b *TopicBroker) IncrConversationOffset(ctx context.Context, conversationID string) (uint32, error)

IncrConversationOffset 为指定会话分配下一个 offset（Redis INCR） key: {prefix}:conv:offset:{conversationID}

func (*TopicBroker) Publish 

func (b *TopicBroker) Publish(ch string, data []byte, opts centrifuge.PublishOptions) (centrifuge.StreamPosition, bool, error)

Publish is a convenience method that internally calls BatchIncrby + PublishWithOffset. For IM scenarios, use BatchIncrby → DB transaction → PublishWithOffset instead.

func (*TopicBroker) PublishJoin 

func (b *TopicBroker) PublishJoin(ch string, info *centrifuge.ClientInfo) error

PublishJoin publishes Join message to channel.

func (*TopicBroker) PublishLeave 

func (b *TopicBroker) PublishLeave(ch string, info *centrifuge.ClientInfo) error

PublishLeave publishes Leave message to channel.

func (*TopicBroker) PublishWithContext 

func (b *TopicBroker) PublishWithContext(ctx context.Context, ch string, data []byte, opts centrifuge.PublishOptions) (sp centrifuge.StreamPosition, fromCache bool, err error)

PublishWithContext is like Publish but accepts a context for distributed tracing.

func (*TopicBroker) PublishWithOffset 

func (b *TopicBroker) PublishWithOffset(ctx context.Context, ch string, data []byte, opts centrifuge.PublishOptions, sp centrifuge.StreamPosition) error

PublishWithOffset publishes a message using a pre-allocated offset. Call this after DB transaction commits. No HINCRBY, no XADD, no Asynq. Returns error if epoch mismatch occurs.

func (*TopicBroker) RegisterBrokerEventHandler 

func (b *TopicBroker) RegisterBrokerEventHandler(handler centrifuge.BrokerEventHandler) error

RegisterBrokerEventHandler is called once when Broker is set to Node.

func (*TopicBroker) RemoveHistory 

func (b *TopicBroker) RemoveHistory(ch string) error

RemoveHistory removes history from channel.

func (*TopicBroker) SetConversationOffset 

func (b *TopicBroker) SetConversationOffset(ctx context.Context, conversationID string, value uint32) error

SetConversationOffset 设置指定会话的 offset 值（Redis SET），用于 fork 等场景初始化计数器

func (*TopicBroker) Subscribe 

func (b *TopicBroker) Subscribe(ch string) error

Subscribe subscribes node to channel.

func (*TopicBroker) Unsubscribe 

func (b *TopicBroker) Unsubscribe(ch string) error

Unsubscribe unsubscribes node from channel.

type TopicBrokerConfig 

type TopicBrokerConfig struct {
	// Prefix is the Redis key prefix.
	Prefix string

	// RedisAddr is the Redis server address (host:port).
	RedisAddr string

	// RedisPassword is the Redis password (optional).
	RedisPassword string

	// RedisDB is the Redis database number.
	RedisDB int

	// HistoryStore is the implementation for querying message history.
	HistoryStore HistoryStore

	// HistoryMetaTTL is the TTL for the meta key (epoch and top_offset).
	// If 0, no TTL is set.
	HistoryMetaTTL time.Duration

	// Logger is the logger used by the broker.
	// If nil, a default logger backed by log.Printf is used.
	Logger Logger

	// Tracing configures distributed tracing.
	// When disabled, all tracing operations are no-ops with zero overhead.
	Tracing TracingConfig
}

TopicBrokerConfig is the configuration for TopicBroker.

type TracingConfig 

type TracingConfig struct {
	// Enabled controls whether tracing is active.
	Enabled bool

	// Provider is the OpenTelemetry TracerProvider.
	// If nil and Enabled is true, a no-op fallback is used.
	Provider trace.TracerProvider
}

TracingConfig configures distributed tracing for centrifuge-plus. When disabled, all tracing operations use a no-op tracer with zero overhead.

Source Files

View all Source files

Directories

Path Synopsis
Package main 提供 centrifuge-plus 的示例应用，演示 TopicBroker 的使用方式。
 Package main 提供 centrifuge-plus 的示例应用，演示 TopicBroker 的使用方式。

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.