mcpField
C
centrifugo

centrifugo

github.com/pedronauck开发644252 次安装1 周前更新
代码仓库

关于 centrifugo

面向开发者的 Centrifugo 集成技能，指导将自托管实时消息服务接入应用。涵盖认证、频道、客户端连接与扩展运维。

主要功能

  1. 01选择实时消息集成模式
  2. 02配置服务与客户端认证
  3. 03设计频道并发布消息
  4. 04设置代理与水平扩展
  5. 05启用历史恢复和监控

适用场景

  1. 01构建实时聊天与通知
  2. 02为客户端连接配置 JWT
  3. 03将服务扩展为多节点集群

安装

复制下面的命令，在终端中运行。

$npx skills add https://github.com/pedronauck/skills --skill centrifugo

安装后在 Claude Code 中输入/centrifugo确认已加载。

包含的文件

除 SKILL.md 外，安装时一起下载的参考资料和脚本。

references/
authentication.md
channels.md
client-sdk.md
configuration.md
engines.md
proxy.md
server-api.md

详细介绍

Centrifugo 实时消息集成

Centrifugo 是一款可自托管、与语言无关的实时消息服务器。它负责处理持久连接（WebSocket、SSE、HTTP-streaming、WebTransport、GRPC），并通过基于频道的 PUB/SUB 模型广播消息。应用后端通过 Server API 向频道发布消息；Centrifugo 会即时将消息发送给在线订阅者。

核心架构

Backend App ──(HTTP/GRPC Server API)──> Centrifugo Cluster ──(WebSocket/SSE)──> Clients
                                              │
                                        Redis/NATS Broker
                                        (for multi-node)

核心原则：所有用户生成的数据都先经过应用后端（验证、持久化，然后发布到 Centrifugo）。Centrifugo 是传输层，而非事实来源。

操作步骤

步骤 1：确定集成模式

识别要构建的实时功能，并选择合适的模式：

  1. 单向发布（最常见）：后端在处理用户请求后向频道发布消息。客户端订阅并被动接收更新。
  2. 通过代理实现双向通信：客户端通过实时连接发送数据；Centrifugo 将事件（connect、subscribe、publish、RPC）代理给后端进行验证。
  3. 服务端订阅：后端控制用户订阅哪些频道（适用于个性化信息流、通知）。
步骤 2：配置 Centrifugo 服务器

生成最简配置：

centrifugo genconfig  # Creates config.json

也可以使用 Docker：

docker run -p 8000:8000 centrifugo/centrifugo:v6 centrifugo

基本配置结构（支持 JSON、YAML 或 TOML）：

{
  "client": {
    "token": {
      "hmac_secret_key": "<SECRET>"
    },
    "allowed_origins": ["http://localhost:3000"]
  },
  "http_api": {
    "key": "<API_KEY>"
  },
  "channel": {
    "without_namespace": {
      "allow_subscribe_for_client": true
    }
  }
}

阅读 references/configuration.md，查看完整配置参考，包括 TLS、管理界面、环境变量和高级选项。

步骤 3：设置身份验证

Centrifugo 通过 JWT 或连接代理对客户端进行身份验证。

JWT 方式（大多数情况下推荐）：

  • 后端生成一个 JWT，其中包含 sub（用户 ID）和 exp 声明，并使用配置的 HMAC/RSA/ECDSA 密钥签名。
  • 客户端连接时传递令牌。
  • 支持为长连接刷新令牌。
# Backend: generate connection JWT (Python example)
import jwt, time
token = jwt.encode(
    {"sub": "user123", "exp": int(time.time()) + 3600},
    "<HMAC_SECRET>", algorithm="HS256"
)

// Client: connect with token
const client = new Centrifuge("ws://localhost:8000/connection/websocket", {
  token: "<JWT_TOKEN>",
});
client.connect();

连接代理方式（替代方案）：

  • Centrifugo 将连接请求转发到后端端点进行身份验证。
  • 无需 JWT；后端返回用户身份信息。

阅读 references/authentication.md，了解 JWT 声明、令牌刷新、代理身份验证和频道级授权令牌。

步骤 4：设计频道结构

频道是用作消息通道的临时字符串。使用命名空间应用不同的行为。

命名约定：

  • chat:room-123 — 命名空间 chat，对应房间 123 的频道
  • notifications:user#42 — 用户受限频道（只有用户 42 可以订阅）
  • $private:secret — 私有频道前缀（需要订阅令牌）

命名空间配置：

{
  "channel": {
    "namespaces": [
      {
        "name": "chat",
        "presence": true,
        "history_size": 50,
        "history_ttl": "300s",
        "force_recovery": true,
        "join_leave": true
      },
      {
        "name": "notifications",
        "allow_user_limited_channels": true
      }
    ]
  }
}

阅读 references/channels.md，了解所有频道选项、命名空间规则和特殊频道前缀。

步骤 5：从后端发布（Server API）

使用 HTTP 或 GRPC 从应用后端发布消息。

HTTP API — 使用 X-API-Key 标头向 /api/<method> 发送 POST 请求：

# Publish to a channel
curl -X POST -H "X-API-Key: <KEY>" -H "Content-Type: application/json" \
  -d '{"channel": "chat:room-1", "data": {"text": "hello"}}' \
  http://localhost:8000/api/publish

# Broadcast to multiple channels
curl -X POST -H "X-API-Key: <KEY>" -H "Content-Type: application/json" \
  -d '{"channels": ["user:1", "user:2"], "data": {"text": "hello"}}' \
  http://localhost:8000/api/broadcast

可用的 API 方法：publish、broadcast、subscribe、unsubscribe、disconnect、refresh、presence、presence_stats、history、history_remove、channels、info、batch。

阅读 references/server-api.md，查看所有方法签名、请求/响应结构和 HTTP API 库链接。

步骤 6：使用 SDK 连接客户端

官方 SDK：centrifuge-js（浏览器/Node/React Native）、centrifuge-go、centrifuge-dart（Flutter）、centrifuge-swift（iOS）、centrifuge-java（Android）、centrifuge-python。

JavaScript 客户端模式：

import { Centrifuge } from "centrifuge";

const client = new Centrifuge("ws://localhost:8000/connection/websocket", {
  token: "<JWT>",
});

// Connection state handlers
client.on("connecting", ctx => console.log("connecting", ctx));
client.on("connected", ctx => console.log("connected", ctx));
client.on("disconnected", ctx => console.log("disconnected", ctx));

// Subscribe to channel
const sub = client.newSubscription("chat:room-1");
sub.on("publication", ctx => {
  console.log("received:", ctx.data);
});
sub.on("subscribing", ctx => console.log("subscribing", ctx));
sub.on("subscribed", ctx => console.log("subscribed", ctx));
sub.subscribe();

client.connect();

客户端状态：disconnected -> connecting -> connected（采用指数退避自动重新连接）。

订阅状态：unsubscribed -> subscribing -> subscribed（重新连接后自动重新订阅）。

阅读 references/client-sdk.md，了解所有 SDK 模式、令牌刷新、从客户端查询 presence/history、RPC 调用和 Protobuf 模式。

步骤 7：配置事件代理（可选）

将客户端事件代理到后端，以便进行验证和自定义逻辑处理。

支持代理的事件：

  • connect — 在没有 JWT 的情况下验证连接
  • refresh — 延长客户端会话
  • subscribe — 验证频道访问权限
  • publish — 验证客户端发布的内容
  • rpc — 处理自定义的客户端到服务端调用
  • sub_refresh — 延长订阅会话
{
  "client": {
    "proxy": {
      "connect": {
        "enabled": true,
        "endpoint": "http://backend:3000/centrifugo/connect"
      },
      "rpc": {
        "enabled": true,
        "endpoint": "http://backend:3000/centrifugo/rpc"
      }
    }
  },
  "channel": {
    "namespaces": [
      {
        "name": "chat",
        "proxy": {
          "subscribe": {
            "enabled": true,
            "endpoint": "http://backend:3000/centrifugo/subscribe"
          },
          "publish": {
            "enabled": true,
            "endpoint": "http://backend:3000/centrifugo/publish"
          }
        }
      }
    ]
  }
}

阅读 references/proxy.md，了解代理请求/响应架构、GRPC 代理、请求头转发和超时配置。

步骤 8：水平扩展（生产环境）

内存引擎（默认）：仅支持单节点。速度快，无依赖。适用于开发和小规模部署。

Redis 引擎：支持多节点集群。在任意节点上发布的消息都会送达所有订阅者。

{
  "engine": {
    "type": "redis",
    "redis": {
      "address": "redis://localhost:6379"
    }
  }
}

支持 Redis Sentinel、Redis Cluster、多个 Redis 实例间的一致性分片，以及兼容 Redis 的存储（Valkey、DragonflyDB、KeyDB、AWS ElastiCache）。

NATS 代理：仅支持 PUB/SUB 的替代方案（不持久化历史记录/在线状态）。

阅读 references/engines.md，了解 Redis Sentinel/Cluster 设置、分片、NATS 配置，以及独立的代理/在线状态管理器配置。

步骤 9：启用历史记录和恢复

历史记录功能可缓存每个频道最近发布的内容。恢复功能会在重新连接后自动恢复遗漏的消息。

{
  "channel": {
    "namespaces": [
      {
        "name": "chat",
        "history_size": 100,
        "history_ttl": "600s",
        "force_recovery": true,
        "force_positioning": true
      }
    ]
  }
}
  • history_size + history_ttl：定义保留窗口。
  • force_recovery：客户端在重新连接时自动恢复遗漏的消息。
  • force_positioning：检测发布内容是否存在间隙。
  • 缓存恢复模式：针对只关心最新状态的频道进行了优化。详情请阅读 references/channels.md。
步骤 10：添加可观测性

Centrifugo 在 /metrics 提供 Prometheus 指标，并支持结构化日志。

{
  "log_level": "info",
  "prometheus": {
    "enabled": true
  },
  "health": {
    "enabled": true
  }
}

指标包括：连接数、订阅数、发布数、API 调用数、传输统计信息和引擎统计信息。

反模式

不要将 Centrifugo 历史记录用作主数据库

历史记录是临时热缓存，旨在减少重新连接风暴期间的数据库负载。始终维护主应用数据库。

不要只依赖 Centrifugo 传递数据

设计时应考虑优雅降级。如果 Centrifugo 宕机，应用仍应正常运行（只是没有实时更新）。应在 API 响应中返回数据，而不只是通过频道发送。

不要在生产环境中省略 allowed_origins

allowed_origins 为空会阻止所有浏览器连接。请将其配置为前端应用的准确来源。

不要在未经代理验证的情况下直接从客户端发布

务必通过以下任一方式验证客户端发布的内容：

  • 应用后端（客户端调用后端 API，由后端向 Centrifugo 发布），或
  • 发布代理（Centrifugo 将客户端发布的内容转发到后端进行验证）
不要在生产环境中使用没有命名空间的频道

命名空间可对频道行为和权限进行精细控制。请为每项实时功能定义一个命名空间。

不要忽略令牌过期时间

在 JWT 中设置合理的 exp 声明。时间过短会导致刷新请求过多，时间过长则会延迟停用用户。请在客户端 SDK 中实现令牌刷新回调。

错误处理

  • 如果客户端收到 102: unknown channel，请确认配置中已定义该命名空间。
  • 如果客户端收到 103: permission denied，请检查频道权限（allow_subscribe_for_client、订阅 JWT 或订阅代理）。
  • 如果发布的内容未送达，请确认频道名称完全匹配（包括命名空间前缀），并确保至少有一位订阅者。
  • 如果发生重连循环，请检查 allowed_origins、JWT 有效性/过期时间和网络连接。
  • 如果 Redis 引擎连接失败，请确认地址格式（redis://、用于 TLS 的 rediss://、用于 Sentinel 的 redis+sentinel://、用于 Cluster 的 redis+cluster://）。

相关集成

Centrifugo 可与任何后端语言/框架良好集成。常见配置包括：

  • Node.js/Hono/Express：使用 HTTP API 或 centrifuge npm 包执行服务端操作。
  • Go：直接使用 centrifuge Go 库或 Centrifugo HTTP/GRPC API。
  • Python/Django/Flask：使用 pycent 库或直接调用 HTTP API。
  • PHP/Laravel：使用 phpcent 库。

要使用 JavaScript 客户端 SDK（centrifuge-js），请通过 npm 安装：

npm install centrifuge
# or
bun add centrifuge

评论

支持 Markdown