双流架构

同时将事件发布到Kafka（持久化）和Redis Pub/Sub（实时），适用于既需要保证交付又需要即时更新的系统。

安装

OpenClaw / Moltbot / Clawbot

bash
npx clawhub@latest install dual-stream-architecture

适用场景

• - 既需要持久化又需要实时性的事件驱动系统
• 推送实时更新的WebSocket/SSE后端
• 实时展示事件的仪表盘
• Kafka消费者存在延迟但用户期望即时更新

核心模式

go
type DualPublisher struct {
kafka *kafka.Writer
redis *redis.Client
logger *slog.Logger
}

func (p *DualPublisher) Publish(ctx context.Context, event Event) error {
// 1. Kafka：关键路径 - 必须成功
payload, _ := json.Marshal(event)
err := p.kafka.WriteMessages(ctx, kafka.Message{
Key: []byte(event.SourceID),
Value: payload,
})
if err != nil {
return fmt.Errorf(kafka发布失败: %w, err)
}

// 2. Redis：尽力而为 - 不中断操作
p.publishToRedis(ctx, event)

return nil
}

func (p *DualPublisher) publishToRedis(ctx context.Context, event Event) {
// 轻量级负载（完整事件在Kafka中）
notification := map[string]interface{}{
id: event.ID,
type: event.Type,
source_id: event.SourceID,
}

payload, _ := json.Marshal(notification)
channel := fmt.Sprintf(events:%s:%s, event.SourceType, event.SourceID)

// 即发即忘 - 记录错误但不传播
if err := p.redis.Publish(ctx, channel, payload).Err(); err != nil {
p.logger.Warn(redis发布失败, error, err)
}
}

架构

┌──────────────┐ ┌─────────────────┐ ┌──────────────┐
│ 数据摄取器 │────▶│ 双流发布器 │────▶│ Kafka │──▶ 事件处理器
│ │ │ │ │ （持久化） │
└──────────────┘ │ │ └──────────────┘
│ │ ┌──────────────┐
│ │────▶│ Redis PubSub │──▶ WebSocket网关
│ │ │ （实时） │
└─────────────────┘ └──────────────┘

频道命名规范

events:{sourcetype}:{sourceid}

示例：

• - events:user:octocat - 用户octocat的事件
• events:repo:owner/repo - 仓库的事件
• events:org:microsoft - 组织的事件

批量发布

适用于高吞吐量场景：

go
func (p *DualPublisher) PublishBatch(ctx context.Context, events []Event) error {
// 1. 批量发送到Kafka
messages := make([]kafka.Message, len(events))
for i, event := range events {
payload, _ := json.Marshal(event)
messages[i] = kafka.Message{
Key: []byte(event.SourceID),
Value: payload,
}
}

if err := p.kafka.WriteMessages(ctx, messages...); err != nil {
return fmt.Errorf(kafka批量发送失败: %w, err)
}

// 2. Redis：使用管道提高效率
pipe := p.redis.Pipeline()
for _, event := range events {
channel := fmt.Sprintf(events:%s:%s, event.SourceType, event.SourceID)
notification, _ := json.Marshal(map[string]interface{}{
id: event.ID,
type: event.Type,
})
pipe.Publish(ctx, channel, notification)
}

if _, err := pipe.Exec(ctx); err != nil {
p.logger.Warn(redis批量发送失败, error, err)
}

return nil
}

决策树

相关技能

• - 元技能： ai/skills/meta/realtime-dashboard/ — 完整实时仪表盘指南
• websocket-hub-patterns — WebSocket网关
• backend/service-layer-architecture — 服务集成

绝对禁止

• - 绝对不要在Redis错误时失败 — Redis是尽力而为。记录日志并继续。
• 绝对不要向Redis发送完整负载 — 只发送ID，客户端从API获取。
• 绝对不要为每个事件创建Redis频道 — 使用源级别频道。
• 绝对不要跳过Kafka处理不重要的事件 — 所有事件都进入Kafka以便重放。
• 绝对不要使用Redis Pub/Sub进行持久化 — 消息即发即忘。

边界情况