centrifugo

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 はイベント（接続、購読、パブリッシュ、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
    }
  }
}

TLS、管理 UI、環境変数、高度なオプションを含む完全な設定リファレンスについては、references/configuration.md を参照してください。

ステップ 3：認証を設定する

Centrifugo は JWT または接続プロキシを使用してクライアントを認証します。

JWT 方式（ほとんどの場合に推奨）：

• バックエンドは sub（ユーザー ID）と exp クレームを含む JWT を生成し、設定済みの 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 は不要です。バックエンドがユーザー ID を返します。

JWT クレーム、トークン更新、プロキシ認証、チャネル単位の認可トークンについては、references/authentication.md を参照してください。

ステップ 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。

すべてのメソッドのシグネチャ、リクエスト／レスポンスのスキーマ、HTTP API ライブラリーへのリンクについては、references/server-api.md を参照してください。

ステップ 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（再接続時に自動再購読）。

すべての SDK パターン、トークン更新、クライアントからの presence/history、RPC 呼び出し、Protobuf モードについては、references/client-sdk.md を参照してください。

ステップ 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"
          }
        }
      }
    ]
  }
}

プロキシのリクエスト／レスポンススキーマ、GRPC プロキシ、ヘッダー転送、タイムアウト設定については、references/proxy.md をお読みください。

ステップ 8：水平スケーリングする（本番環境）

メモリエンジン（デフォルト）：単一ノードのみです。高速で、依存関係がありません。開発や小規模なデプロイに適しています。

Redis エンジン：マルチノードクラスターに対応します。どのノードでパブリッシュされたメッセージも、すべてのサブスクライバーに届きます。

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

Redis Sentinel、Redis Cluster、複数の Redis インスタンス間での一貫性のあるシャーディング、Redis 互換ストレージ（Valkey、DragonflyDB、KeyDB、AWS ElastiCache）をサポートしています。

NATS ブローカー：PUB/SUB のみを対象とする代替手段です（履歴／プレゼンスの永続化には対応していません）。

Redis Sentinel／Cluster のセットアップ、シャーディング、NATS の設定、ブローカーとプレゼンスマネージャーの個別設定については、references/engines.md をお読みください。

ステップ 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、またはサブスクライブプロキシ）を確認してください。
• パブリケーションが配信されない場合 — チャンネル名が完全に一致していること（名前空間プレフィックスを含む）を確認してください。少なくとも 1 つのサブスクライバーが存在することを確認してください。
• 再接続ループが発生する場合 — 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