feat: 重新设计 delay consumer producer
This commit is contained in:
211
delay/redis_consumer.go
Normal file
211
delay/redis_consumer.go
Normal file
@@ -0,0 +1,211 @@
|
||||
// Package delay ...
|
||||
//
|
||||
// Description : event ...
|
||||
//
|
||||
// Author : go_developer@163.com<白茶清欢>
|
||||
//
|
||||
// Date : 2025-08-22 18:55
|
||||
package delay
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"os"
|
||||
"strings"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"git.zhangdeman.cn/zhangdeman/network/util"
|
||||
"git.zhangdeman.cn/zhangdeman/queue/abstract"
|
||||
"git.zhangdeman.cn/zhangdeman/queue/define"
|
||||
"git.zhangdeman.cn/zhangdeman/redis"
|
||||
"git.zhangdeman.cn/zhangdeman/serialize"
|
||||
redisV9 "github.com/redis/go-redis/v9"
|
||||
"github.com/tidwall/gjson"
|
||||
)
|
||||
|
||||
// NewRedisConsumer 启动一个消费者实例
|
||||
func NewRedisConsumer(
|
||||
ctx context.Context,
|
||||
queueName string, queueCnt int, redisFlag string,
|
||||
consumerHandler abstract.IConsumerHandler,
|
||||
) abstract.IConsumer {
|
||||
if nil == consumerHandler {
|
||||
panic("consumer handler instance is nil")
|
||||
}
|
||||
if queueName == "" || redisFlag == "" {
|
||||
panic("init redis producer: queue name or redis flag is empty")
|
||||
}
|
||||
// 验证redis实例是否存在
|
||||
if _, err := redis.Client.GetRealClientWithError(redisFlag); nil != err {
|
||||
panic(err.Error())
|
||||
}
|
||||
if queueCnt <= 0 {
|
||||
queueCnt = 1 // 默认单队列
|
||||
}
|
||||
|
||||
if nil == ctx {
|
||||
ctx = context.Background()
|
||||
}
|
||||
c := &redisConsumer{
|
||||
lock: &sync.RWMutex{},
|
||||
hasStop: false,
|
||||
ctx: ctx,
|
||||
stopChan: make(chan bool),
|
||||
redisFlag: redisFlag,
|
||||
queueName: queueName,
|
||||
queueCnt: queueCnt,
|
||||
consumerHandler: consumerHandler,
|
||||
}
|
||||
return c
|
||||
}
|
||||
|
||||
type redisConsumer struct {
|
||||
lock *sync.RWMutex
|
||||
hasStop bool
|
||||
stopChan chan bool // 停止请求的chan
|
||||
ctx context.Context // 请求处理上下文
|
||||
redisFlag string // redis 标识, 必须是使用统一的 pkg/redis 进行管理的
|
||||
queueName string // 队列名称,基础公共前缀,尾部后缀会自动根据 shard cnt 进行哈希
|
||||
queueCnt int // 队列数量
|
||||
consumerHandler abstract.IConsumerHandler // 消息处理实例
|
||||
}
|
||||
|
||||
// Start 启动消费者
|
||||
func (r *redisConsumer) Start() error {
|
||||
var realQueueList []string
|
||||
for i := 0; i < r.queueCnt; i++ {
|
||||
realQueueList = append(realQueueList, fmt.Sprintf("%v_%v", r.queueName, i))
|
||||
}
|
||||
|
||||
hostname, _ := os.Hostname()
|
||||
serverInfo := &define.EventConsumerServerInfo{
|
||||
SystemTimestamp: time.Now().UnixMilli(),
|
||||
Hostname: hostname,
|
||||
HostIp: util.IP.GetHostIP(),
|
||||
Queue: strings.Join(realQueueList, ","),
|
||||
}
|
||||
for {
|
||||
isFinish := false
|
||||
select {
|
||||
case <-r.ctx.Done(): // context被取消(对应程序系统默认行为,一般对应程序停止运行)
|
||||
if nil != r.lock {
|
||||
r.lock.Lock()
|
||||
r.hasStop = true
|
||||
r.lock.Unlock()
|
||||
} else {
|
||||
r.hasStop = true
|
||||
}
|
||||
isFinish = true
|
||||
case <-r.stopChan: // 对应开发者介入处理行为, 主进程不停止运行, 但是消费者要退出
|
||||
isFinish = true
|
||||
default: // 60s 无数据进入下一轮等待, 或者 消费到数据正常处理
|
||||
if r.hasStop {
|
||||
// 已停止
|
||||
return nil
|
||||
}
|
||||
res := redis.Wrapper.BRPop(r.ctx, r.redisFlag, realQueueList, 60)
|
||||
if nil != res.Err {
|
||||
if errors.Is(res.Err, redisV9.Nil) {
|
||||
// 判断是否为未订阅到消息
|
||||
continue
|
||||
}
|
||||
r.notifyMessageSubscribeFailure(serverInfo, res.Err)
|
||||
time.Sleep(time.Second * 10)
|
||||
continue
|
||||
}
|
||||
|
||||
// 订阅成功, 判断是否无数据
|
||||
if res.Result == "" {
|
||||
// 直接进入下一轮阻塞等待
|
||||
continue
|
||||
}
|
||||
|
||||
// 订阅到数据, 数据解析
|
||||
// BRPOP 返回数据是数组 0: 读取到数据的队列名 1: 读取到的数据
|
||||
// 前置redis操作统一序列化, 此处读取到的是一个序列化后的字符串 `["queue_name", "value"]`
|
||||
serverInfo.Queue = gjson.Get(res.Result, "0").String()
|
||||
var (
|
||||
err error
|
||||
formatData define.EventData
|
||||
)
|
||||
if err = serialize.JSON.UnmarshalWithNumberForString(gjson.Get(res.Result, "1").String(), &formatData); nil != err {
|
||||
// 回调数据解析失败的处理函数
|
||||
r.notifyMessageUnmarshalFailure(serverInfo, res.Result, err)
|
||||
continue
|
||||
}
|
||||
|
||||
// 处理消息(同步或异步)
|
||||
r.handler(serverInfo, &formatData)
|
||||
}
|
||||
if isFinish {
|
||||
break
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// notifyMessageSubscribeFailure 通知订阅失败
|
||||
func (r *redisConsumer) notifyMessageSubscribeFailure(serverInfo *define.EventConsumerServerInfo, err error) {
|
||||
defer func() {
|
||||
// 防一手 panic, 订阅失败回调出现panic, 不回调panic处理方法
|
||||
recover()
|
||||
}()
|
||||
r.consumerHandler.SubscribeFailureCallback(r.ctx, serverInfo, err)
|
||||
}
|
||||
|
||||
// 回调消息返序列化失败的处理逻辑
|
||||
func (r *redisConsumer) notifyMessageUnmarshalFailure(serverInfo *define.EventConsumerServerInfo, res string, err error) {
|
||||
defer func() {
|
||||
// 防一手 panic, 返回劣化失败回调出现panic, 不回调panic处理方法
|
||||
recover()
|
||||
}()
|
||||
r.consumerHandler.UnmarshalFailureCallback(r.ctx, serverInfo, res, err)
|
||||
}
|
||||
|
||||
// handler 处理订阅到的数据
|
||||
func (r *redisConsumer) handler(serverInfo *define.EventConsumerServerInfo, data *define.EventData) {
|
||||
// 处理消息(同步或异步)
|
||||
if r.consumerHandler.Async(r.ctx, serverInfo, data) {
|
||||
// 异步处理
|
||||
go func() {
|
||||
r.dealMessageData(serverInfo, data)
|
||||
}()
|
||||
} else {
|
||||
r.dealMessageData(serverInfo, data)
|
||||
}
|
||||
}
|
||||
|
||||
// dealMessageData 处理消息数据
|
||||
func (r *redisConsumer) dealMessageData(serverInfo *define.EventConsumerServerInfo, data *define.EventData) {
|
||||
defer func() {
|
||||
// 同步处理
|
||||
if err := recover(); nil != err {
|
||||
// 出现panic, 回调 panic 处理
|
||||
r.consumerHandler.PanicCallback(r.ctx, serverInfo, data, err)
|
||||
}
|
||||
}()
|
||||
|
||||
// 加锁, 若果不需要加锁, 实现接口时函数体留空即可
|
||||
if err := r.consumerHandler.Lock(r.ctx, serverInfo, data); nil != err {
|
||||
r.consumerHandler.LockFailureCallback(r.ctx, serverInfo, data, err)
|
||||
return
|
||||
}
|
||||
defer func() {
|
||||
// 释放锁
|
||||
if err := r.consumerHandler.Unlock(r.ctx, serverInfo, data); nil != err {
|
||||
r.consumerHandler.UnlockFailureCallback(r.ctx, serverInfo, data, err)
|
||||
return
|
||||
}
|
||||
}()
|
||||
// 执行对消息的处理
|
||||
if err := r.consumerHandler.MessageLogic(r.ctx, serverInfo, data); nil != err {
|
||||
r.consumerHandler.MessageLogicFailureCallback(r.ctx, serverInfo, data, err)
|
||||
}
|
||||
}
|
||||
|
||||
// Stop 停止消费者
|
||||
func (r *redisConsumer) Stop() {
|
||||
r.stopChan <- true
|
||||
}
|
||||
Reference in New Issue
Block a user