345 lines
9.4 KiB
Go
345 lines
9.4 KiB
Go
package message
|
||
|
||
import (
|
||
"context"
|
||
"encoding/json"
|
||
"fmt"
|
||
"strings"
|
||
"time"
|
||
|
||
"github.com/gogf/gf/v2/frame/g"
|
||
"github.com/gogf/gf/v2/os/glog"
|
||
"github.com/gogf/gf/v2/util/gconv"
|
||
)
|
||
|
||
// redisMessageClient Redis 实现
|
||
type redisMessageClient struct {
|
||
clientType messageClientType
|
||
}
|
||
|
||
// RedisStreamMessage Redis Stream 消息结构
|
||
type RedisStreamMessage struct {
|
||
ID string
|
||
Values map[string]interface{}
|
||
}
|
||
|
||
// StreamGroup 创建消费组（支持单个或批量）
|
||
func (q *redisMessageClient) streamGroup(ctx context.Context, configs ...interface{}) error {
|
||
if len(configs) == 0 {
|
||
return fmt.Errorf("配置不能为空")
|
||
}
|
||
for _, config := range configs {
|
||
cfg, ok := config.(*RedisConfig)
|
||
if !ok {
|
||
return fmt.Errorf("无效的 Redis 配置类型")
|
||
}
|
||
if err := q.createStreamGroup(ctx, cfg); err != nil {
|
||
return err
|
||
}
|
||
}
|
||
return nil
|
||
}
|
||
|
||
// streamGroup 内部单个创建消费组
|
||
func (q *redisMessageClient) createStreamGroup(ctx context.Context, cfg *RedisConfig) error {
|
||
// 获取默认数据源
|
||
ds, err := GetManager().GetDefaultDataSource()
|
||
if err != nil {
|
||
return fmt.Errorf("获取默认数据源失败: %w", err)
|
||
}
|
||
|
||
// 检查连接状态,未连接则自动重连
|
||
if !ds.IsConnected() {
|
||
if err := ds.Reconnect(ctx); err != nil {
|
||
return fmt.Errorf("redis重连失败: %w", err)
|
||
}
|
||
}
|
||
|
||
_, err = ds.Redis().Do(ctx, "XGROUP", "CREATE", cfg.Stream, cfg.Group, "0", "MKSTREAM")
|
||
if err != nil {
|
||
errStr := err.Error()
|
||
if strings.Contains(errStr, "BUSYGROUP") && strings.Contains(errStr, "already exists") {
|
||
glog.Infof(ctx, "✅ Redis 消费者组已存在: %s", cfg.Group)
|
||
return nil
|
||
}
|
||
return fmt.Errorf("初始化消费者组失败: %w", err)
|
||
}
|
||
glog.Infof(ctx, "✅ Redis 消费者组创建成功: %s", cfg.Group)
|
||
return nil
|
||
}
|
||
|
||
// Publish 内部单个发布消息
|
||
func (q *redisMessageClient) publish(ctx context.Context, config interface{}, data interface{}) error {
|
||
ds, err := GetManager().GetDefaultDataSource()
|
||
if err != nil {
|
||
return fmt.Errorf("获取默认数据源失败: %w", err)
|
||
}
|
||
|
||
if !ds.IsConnected() {
|
||
if err := ds.Reconnect(ctx); err != nil {
|
||
return fmt.Errorf("redis重连失败: %w", err)
|
||
}
|
||
}
|
||
|
||
cfg, ok := config.(*RedisConfig)
|
||
if !ok {
|
||
return fmt.Errorf("无效的redis配置类型")
|
||
}
|
||
values := gconv.Map(data)
|
||
args := make([]interface{}, 0, len(values)*2+2)
|
||
args = append(args, cfg.Stream, "*")
|
||
for key, val := range values {
|
||
args = append(args, key, val)
|
||
}
|
||
result, err := ds.Redis().Do(ctx, "XADD", args...)
|
||
if err != nil {
|
||
g.Log().Errorf(ctx, "❌ Redis 发布消息失败: topic=%s, err=%v", cfg.Stream, err)
|
||
return err
|
||
}
|
||
g.Log().Infof(ctx, "✅ Redis 发布消息成功: topic=%s, messageID=%s", cfg.Stream, gconv.String(result))
|
||
return nil
|
||
}
|
||
|
||
// PublishDelayed 发布延迟消息（使用 ZSET）
|
||
func (q *redisMessageClient) publishDelayed(ctx context.Context, config interface{}, data interface{}, delay int) error {
|
||
ds, err := GetManager().GetDefaultDataSource()
|
||
if err != nil {
|
||
return fmt.Errorf("获取默认数据源失败: %w", err)
|
||
}
|
||
|
||
if !ds.IsConnected() {
|
||
if err := ds.Reconnect(ctx); err != nil {
|
||
return fmt.Errorf("redis重连失败: %w", err)
|
||
}
|
||
}
|
||
|
||
cfg, ok := config.(*RedisConfig)
|
||
if !ok {
|
||
return fmt.Errorf("无效的redis配置类型")
|
||
}
|
||
payload, err := json.Marshal(data)
|
||
if err != nil {
|
||
return fmt.Errorf("序列化数据失败: %w", err)
|
||
}
|
||
score := float64(time.Now().Add(time.Duration(delay)).UnixMilli())
|
||
delayedKey := fmt.Sprintf("delayed:%s", cfg.Stream)
|
||
|
||
// ZADD delayedKey score payload
|
||
_, err = ds.Redis().Do(ctx, "ZADD", delayedKey, score, string(payload))
|
||
if err != nil {
|
||
return err
|
||
}
|
||
|
||
g.Log().Infof(ctx, "✅ Redis 延迟消息已发布: topic=%s, delay=%v", cfg.Stream, delay)
|
||
return nil
|
||
}
|
||
|
||
// Subscribe 订阅消息（支持单个或批量）
|
||
func (q *redisMessageClient) subscribe(ctx context.Context, configs ...interface{}) error {
|
||
if len(configs) == 0 {
|
||
return fmt.Errorf("配置不能为空")
|
||
}
|
||
for _, config := range configs {
|
||
cfg, ok := config.(*RedisConfig)
|
||
if !ok {
|
||
return fmt.Errorf("无效的 Redis 配置类型")
|
||
}
|
||
handler := cfg.HandleFunc
|
||
if handler == nil {
|
||
return fmt.Errorf("必须提供处理函数")
|
||
}
|
||
if err := q.createSubscribe(ctx, cfg, handler); err != nil {
|
||
return err
|
||
}
|
||
}
|
||
return nil
|
||
}
|
||
|
||
// subscribe 内部单个订阅消息
|
||
func (q *redisMessageClient) createSubscribe(ctx context.Context, cfg *RedisConfig, handler func(ctx context.Context, message map[string]interface{}) error) error {
|
||
go func() {
|
||
defer func() {
|
||
if r := recover(); r != nil {
|
||
g.Log().Errorf(ctx, "❌ Redis 消费者 panic: %v", r)
|
||
}
|
||
}()
|
||
|
||
retryTicker := time.NewTicker(time.Second)
|
||
defer retryTicker.Stop()
|
||
|
||
// 重试计数器
|
||
var consecutiveErrors int
|
||
const maxConsecutiveErrors = 3
|
||
|
||
for {
|
||
select {
|
||
case <-ctx.Done():
|
||
g.Log().Infof(ctx, "🔕 Redis 消费者停止: topic=%s", cfg.Stream)
|
||
return
|
||
case <-retryTicker.C:
|
||
err := q.consumeMessages(ctx, cfg, handler)
|
||
if err != nil {
|
||
// 对于超时错误,返回nil继续循环,而不是返回错误
|
||
if strings.Contains(err.Error(), "i/o timeout") || strings.Contains(err.Error(), "timeout") ||
|
||
strings.Contains(err.Error(), "context deadline exceeded") || strings.Contains(err.Error(), "context canceled") {
|
||
|
||
consecutiveErrors++
|
||
if consecutiveErrors > maxConsecutiveErrors {
|
||
g.Log().Errorf(ctx, "Max retries exceeded, giving up")
|
||
return
|
||
}
|
||
backoffTime := 5 * time.Second
|
||
g.Log().Warningf(ctx, "⚠️ 等待 %v 后重试...", backoffTime)
|
||
|
||
time.Sleep(backoffTime)
|
||
} else {
|
||
// 非超时错误（严重错误）
|
||
consecutiveErrors = 0 // 重置计数
|
||
g.Log().Errorf(ctx, "严重错误，立即重试: %v", err)
|
||
|
||
// 短暂等待后重试
|
||
select {
|
||
case <-ctx.Done():
|
||
return
|
||
case <-time.After(time.Second):
|
||
// 继续循环
|
||
}
|
||
}
|
||
} else {
|
||
// 成功时重置错误计数器
|
||
consecutiveErrors = 0
|
||
}
|
||
}
|
||
}
|
||
}()
|
||
return nil
|
||
}
|
||
|
||
// consumeMessages 消费消息
|
||
func (q *redisMessageClient) consumeMessages(ctx context.Context, cfg *RedisConfig, handler func(ctx context.Context, message map[string]interface{}) error) error {
|
||
ds, err := GetManager().GetDefaultDataSource()
|
||
if err != nil {
|
||
return fmt.Errorf("获取默认数据源失败: %w", err)
|
||
}
|
||
|
||
if !ds.IsConnected() {
|
||
if err := ds.Reconnect(ctx); err != nil {
|
||
return fmt.Errorf("redis重连失败: %w", err)
|
||
}
|
||
}
|
||
|
||
// 检查消费者组是否存在
|
||
if err := q.createStreamGroup(ctx, cfg); err != nil {
|
||
return fmt.Errorf("create stream group failed: %w", err)
|
||
}
|
||
|
||
// 使用带重试的命令执行
|
||
result, err := ds.Redis().Do(ctx, "XREADGROUP", "GROUP", cfg.Group, cfg.Consumer, "COUNT", cfg.Count, "BLOCK", 0, "STREAMS", cfg.Stream, ">")
|
||
if err != nil {
|
||
if strings.Contains(err.Error(), "i/o timeout") || strings.Contains(err.Error(), "timeout") ||
|
||
strings.Contains(err.Error(), "context deadline exceeded") || strings.Contains(err.Error(), "context canceled") {
|
||
|
||
}
|
||
return err
|
||
}
|
||
messages, err := q.parseStreamResult(result)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
for _, msg := range messages {
|
||
// 处理消息
|
||
if err := handler(ctx, msg.Values); err != nil {
|
||
g.Log().Errorf(ctx, "❌ 消息处理失败: messageID=%s, err=%v", msg.ID, err)
|
||
continue
|
||
}
|
||
|
||
// ACK 消息
|
||
if cfg.AutoAck {
|
||
if err := q.ackMessage(ctx, cfg.Stream, cfg.Group, msg.ID); err != nil {
|
||
g.Log().Errorf(ctx, "❌ ACK 消息失败: messageID=%s, err=%v", msg.ID, err)
|
||
}
|
||
}
|
||
}
|
||
|
||
return nil
|
||
}
|
||
|
||
// ackMessage ACK 消息
|
||
func (q *redisMessageClient) ackMessage(ctx context.Context, streamKey, groupName string, messageIDs ...string) error {
|
||
ds, err := GetManager().GetDefaultDataSource()
|
||
if err != nil {
|
||
return fmt.Errorf("获取默认数据源失败: %w", err)
|
||
}
|
||
|
||
if !ds.IsConnected() {
|
||
if err := ds.Reconnect(ctx); err != nil {
|
||
return fmt.Errorf("redis重连失败: %w", err)
|
||
}
|
||
}
|
||
|
||
args := make([]interface{}, 0, len(messageIDs)+2)
|
||
args = append(args, streamKey, groupName)
|
||
for _, id := range messageIDs {
|
||
args = append(args, id)
|
||
}
|
||
_, err = ds.Redis().Do(ctx, "XACK", args...)
|
||
return err
|
||
}
|
||
|
||
// parseStreamResult 解析 Stream 结果
|
||
func (q *redisMessageClient) parseStreamResult(result interface{}) ([]RedisStreamMessage, error) {
|
||
if result == nil {
|
||
return []RedisStreamMessage{}, nil
|
||
}
|
||
|
||
var resultVal interface{}
|
||
|
||
// 尝试获取 Val() 方法
|
||
if valuer, ok := result.(interface{ Val() interface{} }); ok {
|
||
resultVal = valuer.Val()
|
||
} else {
|
||
resultVal = result
|
||
}
|
||
|
||
// 检查是否为空
|
||
if resultVal == nil {
|
||
return []RedisStreamMessage{}, nil
|
||
}
|
||
|
||
// 预分配切片容量,避免多次扩容
|
||
messages := make([]RedisStreamMessage, 0)
|
||
|
||
if streamsMap, ok := resultVal.(map[interface{}]interface{}); ok {
|
||
for _, streamMsg := range streamsMap {
|
||
msgArray, ok := streamMsg.([]interface{})
|
||
if !ok {
|
||
continue
|
||
}
|
||
for _, msgData := range msgArray {
|
||
msgArray, ok := msgData.([]interface{})
|
||
if !ok || len(msgArray) < 2 {
|
||
continue
|
||
}
|
||
msgID := gconv.String(msgArray[0])
|
||
fieldsArray, ok := msgArray[1].([]interface{})
|
||
if !ok {
|
||
continue
|
||
}
|
||
values := make(map[string]interface{}, len(fieldsArray)/2)
|
||
for i := 0; i < len(fieldsArray); i += 2 {
|
||
if i+1 < len(fieldsArray) {
|
||
key := gconv.String(fieldsArray[i])
|
||
values[key] = fieldsArray[i+1]
|
||
}
|
||
}
|
||
messages = append(messages, RedisStreamMessage{
|
||
ID: msgID,
|
||
Values: values,
|
||
})
|
||
}
|
||
}
|
||
}
|
||
|
||
return messages, nil
|
||
}
|