timeSliceQueryPush

This commit is contained in:
ziming 2025-06-12 14:21:55 +08:00
parent 4e479ffaa8
commit 32c3b6c7a8
1 changed files with 52 additions and 52 deletions

View File

@ -13,6 +13,53 @@ import (
"voucher/internal/pkg/lock" "voucher/internal/pkg/lock"
) )
func (v *VoucherBiz) isCanNotice(ctx context.Context) error {
if v.bc.Cmb.NoticeStartDays == 0 {
return errors.New("订单定时通知,noticeStartDays eq 0")
}
if v.bc.Cmb.NoticeEndDays == 0 {
return errors.New("订单定时通知,noticeEndDays eq 0")
}
cache := vo.CmbBatchNoticeCacheKey.BuildCache([]string{""})
_, err := v.rdb.Rdb.Get(ctx, cache.Key).Result()
if err == nil {
return fmt.Errorf("订单定时通知,notice 获取redis缓存存在已被执行,本台服务不做执行")
}
if err != redis.Nil {
return fmt.Errorf(fmt.Sprintf("订单定时通知,notice 获取redis缓存%s异常:%v", cache.Key, err))
}
c := vo.CmbBatchNoticeLockKey.BuildCache([]string{""})
return lock.NewMutex(v.rdb.Rdb, c.TTL).Lock(ctx, c.Key, func(ctx context.Context) error {
// 二次获取,判定处理,以免获取锁后又执行了一次
cacheValue, err2 := v.rdb.Rdb.Get(ctx, cache.Key).Result()
if err2 != nil && err2 != redis.Nil {
return fmt.Errorf(fmt.Sprintf("订单定时通知,notice 二次获取redis缓存%s异常:%v", cache.Key, err2))
}
if len(cacheValue) > 0 {
return fmt.Errorf("订单定时通知,notice 二次获取redis缓存存在已被执行,本台服务不做执行")
}
if err = v.rdb.Rdb.Set(ctx, cache.Key, fmt.Sprintf("%d_%d", v.bc.Cmb.NoticeStartDays, v.bc.Cmb.NoticeEndDays), c.TTL).Err(); err != nil {
return fmt.Errorf(fmt.Sprintf("notice 设置redis缓存%s异常:%v", cache.Key, err))
}
log.Warnf("订单定时通知,notice 获取redis缓存,不存在,开始处理")
return nil
})
}
func (v *VoucherBiz) Notice(ctx context.Context) error { func (v *VoucherBiz) Notice(ctx context.Context) error {
if err := v.isCanNotice(ctx); err != nil { if err := v.isCanNotice(ctx); err != nil {
@ -36,7 +83,7 @@ func (v *VoucherBiz) Notice(ctx context.Context) error {
func (v *VoucherBiz) timeSliceQuery(ctx context.Context, startTime, endTime time.Time) error { func (v *VoucherBiz) timeSliceQuery(ctx context.Context, startTime, endTime time.Time) error {
duration := 1 * time.Hour duration := 2 * time.Hour
eg := new(errgroup.Group) eg := new(errgroup.Group)
eg.SetLimit(5) eg.SetLimit(5)
@ -64,7 +111,7 @@ func (v *VoucherBiz) timeSliceQuery(ctx context.Context, startTime, endTime time
defer func() { defer func() {
if err := recover(); err != nil { if err := recover(); err != nil {
log.Errorf("查询券订单状态发生错误:req:%+v,err:%v", err) log.Errorf("订单定时通知,发生错误:req:%+v,err:%v", req, err)
} }
}() }()
@ -84,7 +131,7 @@ func (v *VoucherBiz) ExecuteNotice(ctx context.Context, req *bo.FindInBatchesUse
for _, order := range rows { for _, order := range rows {
if err := v.notice(ctx, order); err != nil { if err := v.notice(ctx, order); err != nil {
log.Errorf("查询券订单状态发生错误,orderNo:%s,err:%v", order.OrderNo, err) log.Errorf("订单定时通知,券订单状态查询处理发生错误,orderNo:%s,err:%v", order.OrderNo, err)
} }
} }
@ -94,53 +141,6 @@ func (v *VoucherBiz) ExecuteNotice(ctx context.Context, req *bo.FindInBatchesUse
} }
func (v *VoucherBiz) isCanNotice(ctx context.Context) error {
if v.bc.Cmb.NoticeStartDays == 0 {
return errors.New("noticeStartDays eq 0")
}
if v.bc.Cmb.NoticeEndDays == 0 {
return errors.New("noticeEndDays eq 0")
}
cache := vo.CmbBatchNoticeCacheKey.BuildCache([]string{""})
_, err := v.rdb.Rdb.Get(ctx, cache.Key).Result()
if err == nil {
return fmt.Errorf("notice 获取redis缓存存在已被执行,本台服务不做执行")
}
if err != redis.Nil {
return fmt.Errorf(fmt.Sprintf("notice 获取redis缓存%s异常:%v", cache.Key, err))
}
c := vo.CmbBatchNoticeLockKey.BuildCache([]string{""})
return lock.NewMutex(v.rdb.Rdb, c.TTL).Lock(ctx, c.Key, func(ctx context.Context) error {
// 二次获取,判定处理,以免获取锁后又执行了一次
cacheValue, err2 := v.rdb.Rdb.Get(ctx, cache.Key).Result()
if err2 != nil && err2 != redis.Nil {
return fmt.Errorf(fmt.Sprintf("notice 二次获取redis缓存%s异常:%v", cache.Key, err2))
}
if len(cacheValue) > 0 {
return fmt.Errorf("notice 二次获取redis缓存存在已被执行,本台服务不做执行")
}
if err = v.rdb.Rdb.Set(ctx, cache.Key, fmt.Sprintf("%d_%d", v.bc.Cmb.NoticeStartDays, v.bc.Cmb.NoticeEndDays), c.TTL).Err(); err != nil {
return fmt.Errorf(fmt.Sprintf("notice 设置redis缓存%s异常:%v", cache.Key, err))
}
log.Warnf("notice 获取redis缓存,不存在,开始处理")
return nil
})
}
func (v *VoucherBiz) notice(ctx context.Context, order *bo.OrderBo) error { func (v *VoucherBiz) notice(ctx context.Context, order *bo.OrderBo) error {
// 批量通知不做数据存储,量会很大 // 批量通知不做数据存储,量会很大
@ -188,11 +188,11 @@ func (v *VoucherBiz) cmbNotice(ctx context.Context, order *bo.OrderBo, orderNoti
reply, err := v.CmbMixRepo.Request(ctx, request, order.NotifyUrl) reply, err := v.CmbMixRepo.Request(ctx, request, order.NotifyUrl)
if err != nil { if err != nil {
return fmt.Errorf("orderNo:%s,outBizNo:%s,%s", order.OrderNo, order.OutBizNo, err.Error()) return fmt.Errorf("订单定时通知,orderNo:%s,outBizNo:%s,%s", order.OrderNo, order.OutBizNo, err.Error())
} }
if reply.RespCode != vo.CmbResponseStatusSuccess.GetValue() { if reply.RespCode != vo.CmbResponseStatusSuccess.GetValue() {
return errors.New(reply.RespMsg) return errors.New("订单定时通知,招行返回:" + reply.RespMsg)
} }
return nil return nil