使用golongpoll实现HTTP长轮询发布订阅服务

golongpoll是一个简单高效的Go语言库，用于实现HTTP长轮询模式的发布订阅服务。下面我将详细介绍如何使用这个库。

安装

首先安装golongpoll库：

go get github.com/jcuga/golongpoll

基本使用

1. 创建长轮询管理器

package main

import (
	"fmt"
	"log"
	"net/http"
	"time"

	"github.com/jcuga/golongpoll"
)

func main() {
	// 创建长轮询管理器
	manager, err := golongpoll.StartLongpoll(golongpoll.Options{
		LoggingEnabled: true, // 启用日志
		// 消息保留1小时
		EventTimeToLiveSeconds: 60 * 60,
		// 长轮询请求最多等待90秒
		MaxLongpollTimeoutSeconds: 90,
	})
	if err != nil {
		log.Fatalf("Failed to create longpoll manager: %v", err)
	}
	defer manager.Shutdown() // 程序退出时关闭管理器

	// 设置HTTP路由
	http.HandleFunc("/events", manager.SubscriptionHandler)
	http.HandleFunc("/publish", func(w http.ResponseWriter, r *http.Request) {
		// 这里处理发布逻辑
	})

	// 启动HTTP服务器
	fmt.Println("Server starting on :8080...")
	log.Fatal(http.ListenAndServe(":8080", nil))
}

2. 发布消息

// 发布消息到特定分类
func publishHandler(w http.ResponseWriter, r *http.Request, manager *golongpoll.LongpollManager) {
	category := r.URL.Query().Get("category")
	message := r.URL.Query().Get("message")
	
	if category == "" || message == "" {
		http.Error(w, "Missing category or message", http.StatusBadRequest)
		return
	}
	
	// 发布消息
	manager.Publish(category, message)
	fmt.Fprintf(w, "Message published to category '%s'", category)
}

3. 订阅消息

客户端可以通过向/events端点发送GET请求来订阅消息：

// 前端JavaScript示例
function subscribe(category) {
    fetch(`/events?category=${category}&timeout=30`)
        .then(response => response.json())
        .then(events => {
            // 处理接收到的事件
            events.forEach(event => {
                console.log("Received:", event.data);
            });
            // 继续订阅
            subscribe(category);
        })
        .catch(error => {
            console.error("Subscription error:", error);
            // 出错后重试
            setTimeout(() => subscribe(category), 1000);
        });
}

// 开始订阅
subscribe("news");

高级功能

1. 多分类订阅

// 客户端可以订阅多个分类
http.HandleFunc("/multi-events", func(w http.ResponseWriter, r *http.Request) {
    categories := r.URL.Query()["category"]
    if len(categories) == 0 {
        http.Error(w, "No categories specified", http.StatusBadRequest)
        return
    }
    
    // 使用manager.SubscriptionHandlerWithCategories处理多分类订阅
    manager.SubscriptionHandlerWithCategories(w, r, categories)
})

2. 自定义事件数据

// 可以发布结构化的自定义数据
type CustomEvent struct {
    UserID  int    `json:"user_id"`
    Message string `json:"message"`
    Time    int64  `json:"timestamp"`
}

// 发布自定义事件
event := CustomEvent{
    UserID:  123,
    Message: "Hello world",
    Time:    time.Now().Unix(),
}
manager.Publish("user_updates", event)

3. 安全性考虑

// 添加认证中间件
func authMiddleware(next http.HandlerFunc) http.HandlerFunc {
    return func(w http.ResponseWriter, r *http.Request) {
        token := r.Header.Get("Authorization")
        if token != "valid-token" {
            http.Error(w, "Unauthorized", http.StatusUnauthorized)
            return
        }
        next(w, r)
    }
}

// 保护订阅端点
http.HandleFunc("/secure-events", authMiddleware(manager.SubscriptionHandler))

性能优化建议

1. 调整超时时间：根据业务需求调整MaxLongpollTimeoutSeconds，通常30-90秒是合理范围。

2. 分类粒度：合理设计分类系统，避免过于宽泛或过于精细的分类。

3. 消息TTL：设置合适的EventTimeToLiveSeconds，避免内存无限增长。

4. 连接限制：在生产环境中，应该限制单个客户端的连接数。

完整示例

package main

import (
	"fmt"
	"log"
	"net/http"
	"time"

	"github.com/jcuga/golongpoll"
)

func main() {
	// 初始化长轮询管理器
	manager, err := golongpoll.StartLongpoll(golongpoll.Options{
		LoggingEnabled:          true,
		EventTimeToLiveSeconds:  60 * 60,
		MaxLongpollTimeoutSeconds: 90,
	})
	if err != nil {
		log.Fatal(err)
	}
	defer manager.Shutdown()

	// 设置路由
	http.HandleFunc("/events", manager.SubscriptionHandler)
	http.HandleFunc("/publish", func(w http.ResponseWriter, r *http.Request) {
		category := r.URL.Query().Get("category")
		message := r.URL.Query().Get("message")
		
		if category == "" || message == "" {
			http.Error(w, "Missing category or message", http.StatusBadRequest)
			return
		}
		
		manager.Publish(category, message)
		fmt.Fprintf(w, "Published to %s: %s", category, message)
	})

	// 模拟后台发布
	go func() {
		ticker := time.NewTicker(5 * time.Second)
		for range ticker.C {
			manager.Publish("time_updates", fmt.Sprintf("Current time: %v", time.Now()))
		}
	}()

	// 启动服务器
	log.Println("Starting server on :8080")
	log.Fatal(http.ListenAndServe(":8080", nil))
}

golongpoll库提供了一种简单高效的方式来实现HTTP长轮询服务，适用于需要实时通知但又不适合WebSocket的场景。它的API设计简洁，性能良好，是构建实时应用的优秀选择。