golang高性能推送服务器集群解决方案插件库gopush-cluster的使用

概述

gopush-cluster是一个Go语言实现的高性能推送服务器集群解决方案。

特性

• 轻量级
• 高性能
• 纯Golang实现
• 消息过期机制
• 离线消息存储
• 支持公共消息或私有消息推送
• 支持多个订阅者（可限制最大订阅数）
• 心跳检测（服务心跳或TCP keepalive）
• 认证机制（未认证的订阅者无法连接到comet节点）
• 多协议支持（WebSocket、TCP，未来计划支持HTTP长轮询）
• 统计功能
• 集群支持（可轻松添加或移除comet、web和message节点）
• 故障转移支持（使用Zookeeper）

架构图

使用示例

下面是一个使用gopush-cluster的简单示例：

package main

import (
	"log"
	"time"

	"github.com/Terry-Mao/gopush-cluster/ketama"
	"github.com/Terry-Mao/gopush-cluster/message"
)

func main() {
	// 初始化一致性哈希环
	ring := ketama.NewRing(ketama.DefaultReplicas)
	
	// 添加节点
	ring.AddNode("node1", 1)
	ring.AddNode("node2", 1)
	ring.AddNode("node3", 1)
	
	// 创建消息
	msg := &message.Message{
		Key:     "user123",  // 用户ID
		Msg:     []byte("Hello, gopush-cluster!"),  // 消息内容
		Expire:  time.Now().Add(time.Hour).Unix(),  // 过期时间
		Private: true,  // 是否为私有消息
	}
	
	// 根据key选择节点
	node := ring.Hash(msg.Key)
	log.Printf("Selected node: %s for key: %s", node, msg.Key)
	
	// 这里通常是发送消息到选中的节点
	// 实际应用中会通过网络调用将消息发送到对应的comet节点
	// sendMessageToNode(node, msg)
	
	// 接收端示例
	// 通常客户端会通过WebSocket或TCP连接到comet节点
	// 订阅自己的消息通道
}

客户端连接示例

package main

import (
	"log"
	"time"

	"golang.org/x/net/websocket"
)

func main() {
	// WebSocket连接示例
	wsURL := "ws://your-gopush-cluster-server:port/sub?key=user123&heartbeat=30"
	
	// 建立WebSocket连接
	ws, err := websocket.Dial(wsURL, "", "http://localhost/")
	if err != nil {
		log.Fatal(err)
	}
	defer ws.Close()
	
	// 心跳处理
	go func() {
		ticker := time.NewTicker(25 * time.Second) // 略小于服务器心跳间隔
		defer ticker.Stop()
		
		for range ticker.C {
			if _, err := ws.Write([]byte("heartbeat")); err != nil {
				log.Println("Heartbeat failed:", err)
				return
			}
		}
	}()
	
	// 接收消息
	for {
		var msg []byte
		if err := websocket.Message.Receive(ws, &msg); err != nil {
			log.Println("Receive error:", err)
			break
		}
		log.Printf("Received message: %s", msg)
	}
}

服务器配置示例

package main

import (
	"flag"
	"log"

	"github.com/Terry-Mao/gopush-cluster/comet"
	"github.com/Terry-Mao/gopush-cluster/web"
)

func main() {
	// 解析命令行参数
	flag.Parse()
	
	// 初始化comet节点配置
	cometConf := &comet.Config{
		TCPBind:         []string{"0.0.0.0:8080"},
		WebsocketBind:   []string{"0.0.0.0:8081"},
		RPCBind:         []string{"0.0.0.0:8082"},
		RedisSource:     "127.0.0.1:6379",
		RedisIdle:       10,
		RedisMaxActive:  100,
		RedisMaxIdle:    20,
		ZookeeperAddr:   []string{"127.0.0.1:2181"},
		ZookeeperPath:   "/gopush-cluster",
		ZookeeperTimeout: 30,
	}
	
	// 启动comet节点
	cometServer := comet.NewServer(cometConf)
	go func() {
		if err := cometServer.Init(); err != nil {
			log.Fatal(err)
		}
	}()
	
	// 初始化web节点配置
	webConf := &web.Config{
		HTTPBind:        []string{"0.0.0.0:8090"},
		RPCBind:         []string{"0.0.0.0:8091"},
		ZookeeperAddr:   []string{"127.0.0.1:2181"},
		ZookeeperPath:   "/gopush-cluster",
		ZookeeperTimeout: 30,
	}
	
	// 启动web节点
	webServer := web.NewServer(webConf)
	if err := webServer.Init(); err != nil {
		log.Fatal(err)
	}
}

许可证

gopush-cluster采用GNU通用公共许可证第3版（GPLv3）分发。