lan
6bf87fec46
Introduce several performance and synchronization enhancements: - Implement `SeqBufferManager` to allow sequence number pre-allocation via Redis Lua scripts, reducing atomic increment overhead. - Add `PushWorker` to handle asynchronous message pushing using Redis Streams. - Implement incremental conversation synchronization via `ConversationVersionLog` to allow clients to fetch only recent changes. - Add support for Gzip compression in WebSocket communications to reduce bandwidth usage. - Update dependency injection and configuration to support these new components.
381 lines
9.8 KiB
Go
381 lines
9.8 KiB
Go
package service
|
||
|
||
import (
|
||
"context"
|
||
"encoding/json"
|
||
"fmt"
|
||
"time"
|
||
|
||
"with_you/internal/dto"
|
||
"with_you/internal/model"
|
||
"with_you/internal/pkg/ws"
|
||
"with_you/internal/repository"
|
||
|
||
"github.com/redis/go-redis/v9"
|
||
"go.uber.org/zap"
|
||
)
|
||
|
||
// PushWorkerConfig 推送 Worker 配置
|
||
type PushWorkerConfig struct {
|
||
Enabled bool
|
||
Stream string // Redis Stream 名称，默认 "msg_push"
|
||
Group string // Consumer Group 名称，默认 "push_worker"
|
||
BatchSize int // XREADGROUP 每次读取条数，默认 50
|
||
PollTimeoutMs int // XREADGROUP BLOCK 超时 ms，默认 2000
|
||
MaxRetries int // 消息最大重试次数，默认 3
|
||
MaxStreamLen int64 // Stream MAXLEN，默认 100000
|
||
IdleTimeoutMs int64 // XPENDING 空闲超时 ms，默认 30000
|
||
}
|
||
|
||
// PushStreamMsg 推送到 Redis Stream 的消息结构
|
||
type PushStreamMsg struct {
|
||
ConversationID string `json:"conversation_id"`
|
||
SenderID string `json:"sender_id"`
|
||
MessageID string `json:"message_id"`
|
||
ConvType string `json:"conv_type"`
|
||
Timestamp int64 `json:"timestamp"`
|
||
}
|
||
|
||
// PushWorker 从 Redis Stream 消费消息并异步推送
|
||
type PushWorker struct {
|
||
rdb *redis.Client
|
||
publisher ws.MessagePublisher
|
||
pushSvc PushService
|
||
msgRepo repository.MessageRepository
|
||
userRepo repository.UserRepository
|
||
config PushWorkerConfig
|
||
stopCh chan struct{}
|
||
doneCh chan struct{}
|
||
}
|
||
|
||
// NewPushWorker 创建推送 Worker
|
||
func NewPushWorker(
|
||
rdb *redis.Client,
|
||
publisher ws.MessagePublisher,
|
||
pushSvc PushService,
|
||
msgRepo repository.MessageRepository,
|
||
userRepo repository.UserRepository,
|
||
config PushWorkerConfig,
|
||
) *PushWorker {
|
||
if config.Stream == "" {
|
||
config.Stream = "msg_push"
|
||
}
|
||
if config.Group == "" {
|
||
config.Group = "push_worker"
|
||
}
|
||
if config.BatchSize <= 0 {
|
||
config.BatchSize = 50
|
||
}
|
||
if config.PollTimeoutMs <= 0 {
|
||
config.PollTimeoutMs = 2000
|
||
}
|
||
if config.MaxRetries <= 0 {
|
||
config.MaxRetries = 3
|
||
}
|
||
if config.MaxStreamLen <= 0 {
|
||
config.MaxStreamLen = 100000
|
||
}
|
||
if config.IdleTimeoutMs <= 0 {
|
||
config.IdleTimeoutMs = 30000
|
||
}
|
||
return &PushWorker{
|
||
rdb: rdb,
|
||
publisher: publisher,
|
||
pushSvc: pushSvc,
|
||
msgRepo: msgRepo,
|
||
userRepo: userRepo,
|
||
config: config,
|
||
stopCh: make(chan struct{}),
|
||
doneCh: make(chan struct{}),
|
||
}
|
||
}
|
||
|
||
// IsEnabled 是否启用
|
||
func (w *PushWorker) IsEnabled() bool {
|
||
return w != nil && w.config.Enabled
|
||
}
|
||
|
||
// Start 启动 Worker
|
||
func (w *PushWorker) Start(ctx context.Context) {
|
||
if !w.config.Enabled {
|
||
return
|
||
}
|
||
|
||
// 创建 Consumer Group（忽略 BUSYGROUP 错误）
|
||
err := w.rdb.XGroupCreateMkStream(ctx, w.config.Stream, w.config.Group, "0").Err()
|
||
if err != nil {
|
||
// BUSYGROUP 错误可以忽略
|
||
zap.L().Debug("XGroupCreateMkStream", zap.Error(err))
|
||
}
|
||
|
||
go w.consumeLoop()
|
||
go w.claimPendingLoop()
|
||
|
||
zap.L().Info("PushWorker started",
|
||
zap.String("stream", w.config.Stream),
|
||
zap.String("group", w.config.Group),
|
||
)
|
||
}
|
||
|
||
// Stop 停止 Worker
|
||
func (w *PushWorker) Stop() {
|
||
if !w.config.Enabled {
|
||
return
|
||
}
|
||
close(w.stopCh)
|
||
// 等待两个循环退出
|
||
<-w.doneCh
|
||
<-w.doneCh
|
||
zap.L().Info("PushWorker stopped")
|
||
}
|
||
|
||
// consumeLoop 主消费循环
|
||
func (w *PushWorker) consumeLoop() {
|
||
defer func() { w.doneCh <- struct{}{} }()
|
||
|
||
consumerName := fmt.Sprintf("pw-%d", time.Now().UnixNano())
|
||
args := &redis.XReadGroupArgs{
|
||
Group: w.config.Group,
|
||
Consumer: consumerName,
|
||
Streams: []string{w.config.Stream, ">"},
|
||
Count: int64(w.config.BatchSize),
|
||
Block: time.Duration(w.config.PollTimeoutMs) * time.Millisecond,
|
||
}
|
||
|
||
for {
|
||
select {
|
||
case <-w.stopCh:
|
||
return
|
||
default:
|
||
}
|
||
|
||
results, err := w.rdb.XReadGroup(context.Background(), args).Result()
|
||
if err != nil {
|
||
if err == redis.Nil {
|
||
continue
|
||
}
|
||
zap.L().Warn("PushWorker XReadGroup error", zap.Error(err))
|
||
time.Sleep(1 * time.Second)
|
||
continue
|
||
}
|
||
|
||
for _, stream := range results {
|
||
for _, msg := range stream.Messages {
|
||
if err := w.handleMsg(msg); err != nil {
|
||
zap.L().Warn("PushWorker handle message error",
|
||
zap.String("stream_id", msg.ID),
|
||
zap.Error(err),
|
||
)
|
||
}
|
||
// ACK 无论成功失败（失败消息由 claimPendingLoop 重试）
|
||
w.rdb.XAck(context.Background(), w.config.Stream, w.config.Group, msg.ID)
|
||
}
|
||
}
|
||
}
|
||
}
|
||
|
||
// claimPendingLoop 定期认领超时未 ACK 的消息
|
||
func (w *PushWorker) claimPendingLoop() {
|
||
defer func() { w.doneCh <- struct{}{} }()
|
||
|
||
consumerName := fmt.Sprintf("pw-claim-%d", time.Now().UnixNano())
|
||
ticker := time.NewTicker(30 * time.Second)
|
||
defer ticker.Stop()
|
||
|
||
for {
|
||
select {
|
||
case <-w.stopCh:
|
||
return
|
||
case <-ticker.C:
|
||
}
|
||
|
||
// 查询 pending 消息
|
||
pendingResult, err := w.rdb.XPendingExt(context.Background(), &redis.XPendingExtArgs{
|
||
Stream: w.config.Stream,
|
||
Group: w.config.Group,
|
||
Start: "-",
|
||
End: "+",
|
||
Count: int64(w.config.BatchSize),
|
||
Idle: time.Duration(w.config.IdleTimeoutMs) * time.Millisecond,
|
||
}).Result()
|
||
if err != nil {
|
||
if err == redis.Nil {
|
||
continue
|
||
}
|
||
zap.L().Warn("PushWorker XPENDING error", zap.Error(err))
|
||
continue
|
||
}
|
||
|
||
for _, pending := range pendingResult {
|
||
// 认领消息
|
||
claimed, err := w.rdb.XClaim(context.Background(), &redis.XClaimArgs{
|
||
Stream: w.config.Stream,
|
||
Group: w.config.Group,
|
||
Consumer: consumerName,
|
||
MinIdle: time.Duration(w.config.IdleTimeoutMs) * time.Millisecond,
|
||
Messages: []string{pending.ID},
|
||
}).Result()
|
||
if err != nil || len(claimed) == 0 {
|
||
continue
|
||
}
|
||
|
||
retryCount := int(pending.RetryCount)
|
||
for _, msg := range claimed {
|
||
if retryCount >= w.config.MaxRetries {
|
||
zap.L().Warn("PushWorker message exceeded max retries, deleting",
|
||
zap.String("stream_id", msg.ID),
|
||
zap.Int("retry_count", retryCount),
|
||
)
|
||
w.rdb.XDel(context.Background(), w.config.Stream, msg.ID)
|
||
continue
|
||
}
|
||
|
||
if err := w.handleMsg(msg); err != nil {
|
||
zap.L().Warn("PushWorker claim retry handle error",
|
||
zap.String("stream_id", msg.ID),
|
||
zap.Int("retry_count", retryCount),
|
||
zap.Error(err),
|
||
)
|
||
}
|
||
w.rdb.XAck(context.Background(), w.config.Stream, w.config.Group, msg.ID)
|
||
}
|
||
}
|
||
}
|
||
}
|
||
|
||
// handleMsg 处理一条 Stream 消息
|
||
func (w *PushWorker) handleMsg(streamMsg redis.XMessage) error {
|
||
data, ok := streamMsg.Values["data"]
|
||
if !ok {
|
||
return fmt.Errorf("missing data field in stream message %s", streamMsg.ID)
|
||
}
|
||
|
||
var pushMsg PushStreamMsg
|
||
dataStr, ok := data.(string)
|
||
if !ok {
|
||
return fmt.Errorf("data field is not string in stream message %s", streamMsg.ID)
|
||
}
|
||
if err := json.Unmarshal([]byte(dataStr), &pushMsg); err != nil {
|
||
return fmt.Errorf("unmarshal push message: %w", err)
|
||
}
|
||
|
||
ctx := context.Background()
|
||
|
||
// 加载消息
|
||
message, err := w.msgRepo.GetMessageByID(pushMsg.MessageID)
|
||
if err != nil {
|
||
return fmt.Errorf("load message %s: %w", pushMsg.MessageID, err)
|
||
}
|
||
|
||
// 加载会话
|
||
conv, err := w.msgRepo.GetConversation(pushMsg.ConversationID)
|
||
if err != nil {
|
||
return fmt.Errorf("load conversation %s: %w", pushMsg.ConversationID, err)
|
||
}
|
||
|
||
// 加载参与者
|
||
participants, err := w.msgRepo.GetConversationParticipants(pushMsg.ConversationID)
|
||
if err != nil {
|
||
return fmt.Errorf("load participants for %s: %w", pushMsg.ConversationID, err)
|
||
}
|
||
|
||
// ---- WS Push ----
|
||
targetIDs := make([]string, 0, len(participants))
|
||
for _, p := range participants {
|
||
if conv.Type == model.ConversationTypePrivate && p.UserID == pushMsg.SenderID {
|
||
continue
|
||
}
|
||
targetIDs = append(targetIDs, p.UserID)
|
||
}
|
||
|
||
detailType := "private"
|
||
if conv.Type == model.ConversationTypeGroup {
|
||
detailType = "group"
|
||
}
|
||
|
||
if len(targetIDs) > 20 && conv.Type == model.ConversationTypeGroup {
|
||
w.publisher.PublishToUsersConcurrent(targetIDs, "chat_message", map[string]any{
|
||
"detail_type": detailType,
|
||
"message": dto.ConvertMessageToResponse(message),
|
||
})
|
||
} else {
|
||
w.publisher.PublishToUsers(targetIDs, "chat_message", map[string]any{
|
||
"detail_type": detailType,
|
||
"message": dto.ConvertMessageToResponse(message),
|
||
})
|
||
}
|
||
|
||
// ---- WS unread count push ----
|
||
for _, p := range participants {
|
||
if p.UserID == pushMsg.SenderID {
|
||
continue
|
||
}
|
||
// PushWorker doesn't have conversationCache, skip unread count push here
|
||
// The inline push in chat_service still handles this when push worker is disabled
|
||
}
|
||
|
||
// ---- JPush for offline users ----
|
||
if w.pushSvc == nil {
|
||
return nil
|
||
}
|
||
|
||
sender := ChatMessageSender{ID: pushMsg.SenderID}
|
||
if senderUser, sErr := w.userRepo.GetByID(pushMsg.SenderID); sErr == nil {
|
||
sender.Name = senderUser.Nickname
|
||
sender.Avatar = senderUser.Avatar
|
||
}
|
||
|
||
convName := ""
|
||
groupID := ""
|
||
groupAvatar := ""
|
||
if conv.Type == model.ConversationTypeGroup && conv.Group != nil {
|
||
convName = conv.Group.Name
|
||
groupID = conv.Group.ID
|
||
groupAvatar = conv.Group.Avatar
|
||
}
|
||
|
||
allTargetIDs := make([]string, 0, len(participants))
|
||
for _, p := range participants {
|
||
if p.UserID == pushMsg.SenderID {
|
||
continue
|
||
}
|
||
allTargetIDs = append(allTargetIDs, p.UserID)
|
||
}
|
||
|
||
var offlineTargetIDs []string
|
||
if w.publisher != nil && len(allTargetIDs) > 0 {
|
||
_, offlineTargetIDs = w.publisher.FilterOnline(allTargetIDs)
|
||
} else {
|
||
offlineTargetIDs = allTargetIDs
|
||
}
|
||
|
||
for _, uid := range offlineTargetIDs {
|
||
if pushErr := w.pushSvc.PushChatMessage(ctx, uid, pushMsg.ConversationID, &sender, conv.Type, convName, message, groupID, groupAvatar); pushErr != nil {
|
||
zap.L().Debug("push worker: JPush chat message failed",
|
||
zap.String("userID", uid),
|
||
zap.String("conversationID", pushMsg.ConversationID),
|
||
zap.Error(pushErr),
|
||
)
|
||
}
|
||
}
|
||
|
||
return nil
|
||
}
|
||
|
||
// PublishToPush 将消息发布到推送 Stream
|
||
func PublishToPush(ctx context.Context, rdb *redis.Client, stream string, msg *PushStreamMsg) error {
|
||
data, err := json.Marshal(msg)
|
||
if err != nil {
|
||
return fmt.Errorf("marshal push message: %w", err)
|
||
}
|
||
|
||
return rdb.XAdd(ctx, &redis.XAddArgs{
|
||
Stream: stream,
|
||
MaxLen: 100000,
|
||
Approx: true,
|
||
ID: "*",
|
||
Values: map[string]any{
|
||
"data": string(data),
|
||
},
|
||
}).Err()
|
||
} |