miniblog

微型博客实时流与 WebSocket 技术

深入微型博客实时推送技术栈，对比 WebSocket、Server-Sent Events 与长轮询方案，实现消息通知、时间线增量更新与在线状态系统。

bingrong.yan 5 分钟阅读 2498 字

实时性是社交媒体体验的关键差异化因素。用户发布一条短文，期待关注者能即刻感知；收到一条回复，期待立即获得通知。本文将深入微型博客实时推送系统的技术实现，对比 WebSocket、Server-Sent Events 和长轮询三种方案的适用场景，并分别实现消息通知送达、时间线增量更新和在线状态查询三大核心实时功能。

一、实时推送方案对比

1.1 三种主流方案

维度长轮询 (Long Polling)Server-Sent Events (SSE)WebSocket
通信方向客户端 → 服务端服务端 → 客户端全双工
连接数每次请求新建单个 HTTP 连接单个 TCP 连接
协议开销高（HTTP 头重复）中（HTTP/1.1 或 HTTP/2）低（帧头极小）
浏览器支持全平台现代浏览器（IE 不支持）现代浏览器
适用场景兼容旧客户端单向推送（通知、日志）双向实时通信
复杂度低低中

微型博客的实时需求可以按方向拆解：服务端 → 客户端 的通知推送（有新消息、时间线更新）适合 SSE；客户端 → 服务端 的输入（正在输入、心跳）和 双向 的即时通讯（私信聊天）需要 WebSocket。

1.2 方案选型策略

通知系统（单向推送） → SSE
    ↓
时间线增量更新 → SSE（或 HTTP/2 Push）
    ↓
在线状态 + 私信聊天 → WebSocket
    ↓
客户端心跳 + 输入提示 → WebSocket

在典型实现中，SSE 负责主要的推送负载，WebSocket 负责需要双向交互的场景。两者可以共存，SSE 连接甚至可以降级到长轮询以兼容旧环境。

二、SSE 通知系统实现

2.1 Go SSE 服务端

package sse

import (
	"fmt"
	"net/http"
	"time"
	
	"github.com/gin-gonic/gin"
)

// Event 定义 SSE 事件结构
type Event struct {
	ID    string `json:"id"`
	Type  string `json:"type"`
	Data  []byte `json:"data"`
}

// Client 表示一个 SSE 连接
type Client struct {
	UserID int64
	Chan   chan Event
}

// Hub 管理所有 SSE 连接
type Hub struct {
	clients    map[int64]*Client  // user_id -> client
	register   chan *Client
	unregister chan *Client
	broadcast  chan Event
}

func NewHub() *Hub {
	return &Hub{
		clients:    make(map[int64]*Client),
		register:   make(chan *Client),
		unregister: make(chan *Client),
		broadcast:  make(chan Event, 100),
	}
}

func (h *Hub) Run() {
	for {
		select {
		case client := <-h.register:
			h.clients[client.UserID] = client
			
		case client := <-h.unregister:
			if _, ok := h.clients[client.UserID]; ok {
				delete(h.clients, client.UserID)
				close(client.Chan)
			}
			
		case event := <-h.broadcast:
			// 定向推送或广播
			for _, client := range h.clients {
				select {
				case client.Chan <- event:
				default:
					// 客户端缓冲区满，丢弃事件
				}
			}
		}
	}
}

func (h *Hub) SendToUser(userID int64, event Event) {
	if client, ok := h.clients[userID]; ok {
		select {
		case client.Chan <- event:
		default:
		}
	}
}

func (h *Hub) SendToUsers(userIDs []int64, event Event) {
	for _, uid := range userIDs {
		h.SendToUser(uid, event)
	}
}

// Handler SSE 连接处理器
func (h *Hub) Handler() gin.HandlerFunc {
	return func(c *gin.Context) {
		userID := c.GetInt64("user_id")
		if userID == 0 {
			c.AbortWithStatus(401)
			return
		}
		
		c.Header("Content-Type", "text/event-stream")
		c.Header("Cache-Control", "no-cache")
		c.Header("Connection", "keep-alive")
		c.Header("Access-Control-Allow-Origin", "*")
		
		client := &Client{
			UserID: userID,
			Chan:   make(chan Event, 10),
		}
		
		h.register <- client
		defer func() { h.unregister <- client }()
		
		// 发送初始连接确认
		c.SSEvent("connected", fmt.Sprintf(`{"user_id":%d,"time":"%s"}`, userID, time.Now().Format(time.RFC3339)))
		
		// 设置刷新器防止连接超时
		ticker := time.NewTicker(30 * time.Second)
		defer ticker.Stop()
		
		for {
			select {
			case event, ok := <-client.Chan:
				if !ok {
					return
				}
				c.SSEvent(event.Type, string(event.Data))
				
			case <-ticker.C:
				// 发送心跳注释保持连接
				c.Writer.Write([]byte(":heartbeat\n\n"))
				c.Writer.Flush()
				
			case <-c.Request.Context().Done():
				return
			}
		}
	}
}

2.2 事件类型定义

package domain

import (
	"encoding/json"
	"time"
)

type NotificationType string

const (
	NotificationNewPost     NotificationType = "new_post"
	NotificationNewReply    NotificationType = "new_reply"
	NotificationNewLike     NotificationType = "new_like"
	NotificationNewFollow   NotificationType = "new_follow"
	NotificationNewRepost   NotificationType = "new_repost"
	NotificationMention     NotificationType = "mention"
	NotificationSystem      NotificationType = "system"
)

type Notification struct {
	ID        string           `json:"id"`
	Type      NotificationType `json:"type"`
	UserID    int64            `json:"user_id"`    // 接收者
	ActorID   int64            `json:"actor_id"`   // 触发者
	ActorName string           `json:"actor_name"`
	PostID    *int64           `json:"post_id,omitempty"`
	Content   string           `json:"content,omitempty"`
	Read      bool             `json:"read"`
	CreatedAt time.Time        `json:"created_at"`
}

func (n *Notification) ToEvent() sse.Event {
	data, _ := json.Marshal(n)
	return sse.Event{
		ID:   n.ID,
		Type: string(n.Type),
		Data: data,
	}
}

2.3 业务层集成

package service

import (
	"context"
	"fmt"
	"time"
	
	"miniblog/internal/domain"
	"miniblog/internal/repository"
	"miniblog/pkg/sse"
)

type NotificationService struct {
	repo    *repository.NotificationRepository
	hub     *sse.Hub
}

func (s *NotificationService) CreateNotification(ctx context.Context, notif *domain.Notification) error {
	// 1. 持久化到数据库
	if err := s.repo.Create(ctx, notif); err != nil {
		return fmt.Errorf("persist notification: %w", err)
	}
	
	// 2. 通过 SSE 推送（如果用户在线）
	s.hub.SendToUser(notif.UserID, notif.ToEvent())
	
	// 3. 更新用户未读计数（Redis）
	if err := s.repo.IncrUnreadCount(ctx, notif.UserID); err != nil {
		// 非致命，记录日志
		fmt.Printf("incr unread count failed: %v\n", err)
	}
	
	return nil
}

// 当用户 A 点赞用户 B 的帖子时触发
func (s *NotificationService) OnPostLiked(ctx context.Context, post *domain.Post, likerID int64) error {
	if post.UserID == likerID {
		return nil  // 自点赞不通知
	}
	
	liker, err := s.userRepo.GetByID(ctx, likerID)
	if err != nil {
		return err
	}
	
	notif := &domain.Notification{
		ID:        generateID(),
		Type:      domain.NotificationNewLike,
		UserID:    post.UserID,
		ActorID:   likerID,
		ActorName: liker.DisplayName,
		PostID:    &post.ID,
		Read:      false,
		CreatedAt: time.Now().UTC(),
	}
	
	return s.CreateNotification(ctx, notif)
}

三、WebSocket 在线状态与私信

3.1 WebSocket 管理器

package ws

import (
	"sync"
	"time"
	
	"github.com/gorilla/websocket"
)

var upgrader = websocket.Upgrader{
	ReadBufferSize:  1024,
	WriteBufferSize: 1024,
	CheckOrigin: func(r *http.Request) bool {
		return true // 生产环境应校验 Origin
	},
}

type Message struct {
	Type    string          `json:"type"`    // message | typing | presence | heartbeat
	From    int64           `json:"from"`
	To      int64           `json:"to,omitempty"`
	Payload json.RawMessage `json:"payload"`
}

type Connection struct {
	UserID int64
	Conn   *websocket.Conn
	Hub    *WHub
	Send   chan []byte
}

func (c *Connection) readPump() {
	defer func() {
		c.Hub.unregister <- c
		c.Conn.Close()
	}()
	
	c.Conn.SetReadLimit(512 * 1024) // 512KB
	c.Conn.SetReadDeadline(time.Now().Add(60 * time.Second))
	c.Conn.SetPongHandler(func(string) error {
		c.Conn.SetReadDeadline(time.Now().Add(60 * time.Second))
		return nil
	})
	
	for {
		_, message, err := c.Conn.ReadMessage()
		if err != nil {
			if websocket.IsUnexpectedCloseError(err, websocket.CloseGoingAway, websocket.CloseAbnormalClosure) {
				log.Printf("websocket error: %v", err)
			}
			break
		}
		
		var msg Message
		if err := json.Unmarshal(message, &msg); err != nil {
			continue
		}
		
		msg.From = c.UserID
		c.Hub.handleMessage(msg)
	}
}

func (c *Connection) writePump() {
	ticker := time.NewTicker(30 * time.Second)
	defer func() {
		ticker.Stop()
		c.Conn.Close()
	}()
	
	for {
		select {
		case message, ok := <-c.Send:
			c.Conn.SetWriteDeadline(time.Now().Add(10 * time.Second))
			if !ok {
				c.Conn.WriteMessage(websocket.CloseMessage, []byte{})
				return
			}
			c.Conn.WriteMessage(websocket.TextMessage, message)
			
		case <-ticker.C:
			c.Conn.SetWriteDeadline(time.Now().Add(10 * time.Second))
			if err := c.Conn.WriteMessage(websocket.PingMessage, nil); err != nil {
				return
			}
		}
	}
}

type WHub struct {
	connections map[int64]*Connection
	mu          sync.RWMutex
	register    chan *Connection
	unregister  chan *Connection
	broadcast   chan Message
}

func NewWHub() *WHub {
	return &WHub{
		connections: make(map[int64]*Connection),
		register:    make(chan *Connection),
		unregister:  make(chan *Connection),
		broadcast:   make(chan Message, 100),
	}
}

func (h *WHub) Run() {
	for {
		select {
		case conn := <-h.register:
			h.mu.Lock()
			h.connections[conn.UserID] = conn
			h.mu.Unlock()
			h.broadcastPresence(conn.UserID, "online")
			
		case conn := <-h.unregister:
			h.mu.Lock()
			if _, ok := h.connections[conn.UserID]; ok {
				delete(h.connections, conn.UserID)
				close(conn.Send)
			}
			h.mu.Unlock()
			h.broadcastPresence(conn.UserID, "offline")
			
		case msg := <-h.broadcast:
			h.handleMessage(msg)
		}
	}
}

func (h *WHub) handleMessage(msg Message) {
	switch msg.Type {
	case "message":
		// 私信转发
		h.sendToUser(msg.To, msg)
		
	case "typing":
		// 输入状态转发
		h.sendToUser(msg.To, msg)
		
	case "heartbeat":
		// 心跳响应
		h.sendToUser(msg.From, Message{Type: "heartbeat_ack", From: 0})
	}
}

func (h *WHub) sendToUser(userID int64, msg Message) {
	h.mu.RLock()
	conn, ok := h.connections[userID]
	h.mu.RUnlock()
	
	if !ok {
		// 用户离线，存入离线队列
		return
	}
	
	data, _ := json.Marshal(msg)
	select {
	case conn.Send <- data:
	default:
		// 发送缓冲区满，关闭连接
		close(conn.Send)
		h.unregister <- conn
	}
}

func (h *WHub) broadcastPresence(userID int64, status string) {
	// 获取该用户的关注者
	// 向在线的关注者广播状态变更
	followerIDs := h.getFollowerIDs(userID)
	
	msg := Message{
		Type: "presence",
		From: userID,
		Payload: mustMarshal(map[string]interface{}{
			"user_id": userID,
			"status":  status,
			"time":    time.Now().Unix(),
		}),
	}
	
	for _, fid := range followerIDs {
		if fid != userID {
			h.sendToUser(fid, msg)
		}
	}
}

func mustMarshal(v interface{}) json.RawMessage {
	b, _ := json.Marshal(v)
	return b
}

3.2 在线状态系统

package service

type PresenceService struct {
	wsHub   *ws.WHub
	redis   *redis.Client
}

func (s *PresenceService) UpdateStatus(userID int64, status string) {
	// 更新 Redis 中的在线状态
	key := fmt.Sprintf("presence:%d", userID)
	if status == "online" {
		s.redis.SetEx(context.Background(), key, "online", 2*time.Minute)
	} else {
		s.redis.Del(context.Background(), key)
	}
}

func (s *PresenceService) GetOnlineStatus(userIDs []int64) map[int64]bool {
	// 批量查询在线状态
	keys := make([]string, len(userIDs))
	for i, id := range userIDs {
		keys[i] = fmt.Sprintf("presence:%d", id)
	}
	
	results := s.redis.MGet(context.Background(), keys...).Val()
	status := make(map[int64]bool)
	for i, r := range results {
		status[userIDs[i]] = r != nil
	}
	return status
}

四、时间线增量更新

传统方案中客户端每次下拉刷新都需要重新获取整页时间线。更优的方案是服务端推送增量更新，客户端仅请求变更部分。

// 增量更新消息格式
type TimelineDelta struct {
	Type      string    `json:"type"`      // insert | update | delete
	Position  int       `json:"position"`  // 插入位置（0 = 顶部）
	Post      *Post     `json:"post,omitempty"`
	PostID    int64     `json:"post_id,omitempty"`
}

// 当有新帖发布时，推送到关注者
func (s *FeedService) PushTimelineDelta(followerIDs []int64, post *domain.Post) {
	delta := TimelineDelta{
		Type:     "insert",
		Position: 0,
		Post:     post,
	}
	
	data, _ := json.Marshal(delta)
	event := sse.Event{
		Type: "timeline_delta",
		Data: data,
	}
	
	// 向在线用户推送
	for _, uid := range followerIDs {
		s.sseHub.SendToUser(uid, event)
	}
}

五、性能考量

5.1 连接管理

指标建议值说明
SSE 心跳间隔30 秒防止 NAT/防火墙断开空闲连接
WebSocket PING30 秒Gorilla 默认心跳机制
单用户最大连接3 个多设备同时登录
服务端缓冲区10 条/连接防止慢客户端拖垮系统
连接超时2 分钟无心跳清理僵尸连接

5.2 扩容策略

实时推送服务是有状态服务（每个连接绑定到特定服务器），水平扩容需要特殊处理：

  • 粘性会话 (Sticky Session)：负载均衡器将同一用户始终路由到同一服务器
  • 共享状态：使用 Redis Pub/Sub 在各服务器间转发消息
  • 无状态重构：将所有实时逻辑下沉到独立的 Gateway 服务
// Redis Pub/Sub 跨实例消息转发
func (h *Hub) subscribeRedis() {
	pubsub := redis.Subscribe(context.Background(), "notifications")
	defer pubsub.Close()
	
	for msg := range pubsub.Channel() {
		var event sse.Event
		json.Unmarshal([]byte(msg.Payload), &event)
		
		// 广播到本实例的所有客户端
		for _, client := range h.clients {
			select {
			case client.Chan <- event:
			default:
			}
		}
	}
}

六、总结

微型博客的实时推送系统需要根据业务场景选择合适的底层技术：SSE 轻量高效，适合单向通知和增量更新；WebSocket 功能完备，适合双向通信和在线状态。两者可以共存，SSE 承载主要的推送流量，WebSocket 处理高交互场景。

生产环境部署时，必须考虑连接管理的边界条件：缓冲区溢出、僵尸连接清理、多实例消息同步。通过 Redis Pub/Sub 和合理的超时配置，可以将有状态的推送服务扩展到支撑百万级并发连接的规模。实时推送不是锦上添花，而是现代社交媒体体验的基础设施。

继续阅读

探索更多技术文章

浏览归档，发现更多关于系统设计、工具链和工程实践的内容。

全部文章 返回首页

「miniblog」更多文章

  1. 内容审核与个性化推荐系统设计
  2. 轻社交媒体产品设计方法论
  3. 去中心化笔记与 ActivityPub 联邦协议