Golang 实现实时消息通信(WebSockets/SockJS)的服务器插件 Centrifugo 的使用

Centrifugo 简介

Centrifugo 是一个开源的、可扩展的实时消息服务器。它可以通过支持的传输协议（WebSocket、HTTP-streaming、Server-Sent Events、GRPC、WebTransport）即时将消息传递给在线的应用程序用户。Centrifugo 采用频道订阅的概念，是一个面向用户的 PUB/SUB 服务器。

Centrifugo 与语言无关，可以与任何后端结合使用，用于构建聊天应用、实时评论、多人在线游戏、实时数据可视化、协作工具等。它非常适合现代架构，可以将业务逻辑与实时传输层解耦。

核心概念

Centrifugo 的核心思想很简单 - 它是基于现代实时传输协议的 PUB/SUB 服务器：

Golang 使用 Centrifugo 的完整示例

下面是一个使用 Golang 与 Centrifugo 集成的完整示例：

1. 安装 Centrifugo

# 使用 Docker 运行 Centrifugo
docker run -p 8000:8000 centrifugo/centrifugo:v5 centrifugo --config /centrifugo/config.json

2. 创建 config.json 配置文件

{
  "token_hmac_secret_key": "your_secret_key",
  "admin_password": "your_admin_password",
  "admin_secret": "your_admin_secret",
  "api_key": "your_api_key",
  "allowed_origins": ["*"]
}

3. Golang 客户端代码示例

package main

import (
	"context"
	"encoding/json"
	"fmt"
	"log"
	"time"

	"github.com/centrifugal/centrifuge-go"
)

// 定义消息结构体
type Message struct {
	Text      string `json:"text"`
	Timestamp int64  `json:"timestamp"`
}

func main() {
	// 创建客户端
	client := centrifuge.NewJsonClient(
		"ws://localhost:8000/connection/websocket",
		centrifuge.Config{
			Token: generateToken("user123"), // 生成 JWT token
		},
	)

	// 设置事件处理器
	client.OnConnecting(func(e centrifuge.ConnectingEvent) {
		log.Printf("Connecting: %d (%s)", e.Code, e.Reason)
	})
	client.OnConnected(func(e centrifuge.ConnectedEvent) {
		log.Printf("Connected with ID: %s", e.ClientID)
	})
	client.OnDisconnected(func(e centrifuge.DisconnectedEvent) {
		log.Printf("Disconnected: %d (%s)", e.Code, e.Reason)
	})

	// 连接
	err := client.Connect()
	if err != nil {
		log.Fatalf("Could not connect: %v", err)
	}
	defer client.Close()

	// 订阅频道
	sub, err := client.NewSubscription("chat:general")
	if err != nil {
		log.Fatalf("Could not create subscription: %v", err)
	}

	// 设置订阅事件处理器
	sub.OnSubscribing(func(e centrifuge.SubscribingEvent) {
		log.Printf("Subscribing: %d (%s)", e.Code, e.Reason)
	})
	sub.OnSubscribed(func(e centrifuge.SubscribedEvent) {
		log.Printf("Subscribed to channel %s", sub.Channel())
	})
	sub.OnUnsubscribed(func(e centrifuge.UnsubscribedEvent) {
		log.Printf("Unsubscribed from channel %s", sub.Channel())
	})

	// 处理接收到的消息
	sub.OnPublication(func(e centrifuge.PublicationEvent) {
		var msg Message
		err := json.Unmarshal(e.Data, &msg)
		if err != nil {
			log.Printf("Error unmarshaling message: %v", err)
			return
		}
		log.Printf("Received message: %s (timestamp: %d)", msg.Text, msg.Timestamp)
	})

	// 订阅
	err = sub.Subscribe()
	if err != nil {
		log.Fatalf("Could not subscribe: %v", err)
	}
	defer sub.Unsubscribe()

	// 发布消息示例
	go func() {
		for i := 0; i < 5; i++ {
			msg := Message{
				Text:      fmt.Sprintf("Hello from Go! (%d)", i),
				Timestamp: time.Now().Unix(),
			}
			data, _ := json.Marshal(msg)
			_, err := sub.Publish(context.Background(), data)
			if err != nil {
				log.Printf("Error publishing message: %v", err)
			}
			time.Sleep(2 * time.Second)
		}
	}()

	// 保持运行
	select {}
}

// 生成 JWT token (简化版，实际应用中应该在后端生成)
func generateToken(userID string) string {
	// 实际应用中应该使用 HMAC 签名生成 JWT
	// 这里简化返回一个示例 token
	return "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJzdWIiOiJ1c2VyMTIzIn0.1234567890"
}

4. Golang 服务端 API 示例

package main

import (
	"bytes"
	"encoding/json"
	"fmt"
	"io/ioutil"
	"log"
	"net/http"
)

type PublishRequest struct {
	Channel string      `json:"channel"`
	Data    interface{} `json:"data"`
}

func main() {
	// Centrifugo API 地址
	apiURL := "http://localhost:8000/api"

	// 准备发布的消息
	message := map[string]interface{}{
		"text":      "Server message",
		"timestamp": 1234567890,
	}

	// 创建请求体
	reqBody := PublishRequest{
		Channel: "chat:general",
		Data:    message,
	}
	jsonBody, err := json.Marshal(reqBody)
	if err != nil {
		log.Fatal(err)
	}

	// 创建 HTTP 请求
	req, err := http.NewRequest("POST", apiURL+"/publish", bytes.NewBuffer(jsonBody))
	if err != nil {
		log.Fatal(err)
	}

	// 设置请求头
	req.Header.Set("Content-Type", "application/json")
	req.Header.Set("Authorization", "apikey your_api_key") // 使用配置中的 API key

	// 发送请求
	client := &http.Client{}
	resp, err := client.Do(req)
	if err != nil {
		log.Fatal(err)
	}
	defer resp.Body.Close()

	// 读取响应
	body, err := ioutil.ReadAll(resp.Body)
	if err != nil {
		log.Fatal(err)
	}

	fmt.Printf("Response: %s\n", body)
}

主要特性

• 高效的实时传输协议：WebSocket、HTTP-streaming、Server-Sent Events、GRPC、WebTransport
• 内置可扩展性，支持 Redis、Redis Cluster 或兼容 Redis 的存储
• 简单的 HTTP 和 GRPC 服务器 API，用于从应用后端与 Centrifugo 通信
• 异步 PostgreSQL 和 Kafka 消费者，支持事务性发件箱和 CDC 模式
• 灵活的连接认证机制：JWT 和代理式（通过 Centrifugo 向后端发送请求）
• 在单个连接上多路复用频道订阅
• 不同类型的订阅：客户端订阅和服务器端订阅
• 各种频道权限策略，频道命名空间概念
• 频道中的热消息历史记录，在重新连接时自动恢复消息

总结

Centrifugo 是一个功能强大的实时消息服务器，Golang 可以轻松地与其集成。上面的示例展示了如何创建客户端连接、订阅频道、发布和接收消息，以及如何从后端服务器通过 API 发布消息。