94 lines
2.5 KiB
Go
94 lines
2.5 KiB
Go
package timeslicequery
|
||
|
||
import (
|
||
"context"
|
||
"fmt"
|
||
"github.com/go-kratos/kratos/v2/log"
|
||
"time"
|
||
"voucher/internal/biz/bo"
|
||
"voucher/internal/biz/do"
|
||
"voucher/internal/pkg/timeslice"
|
||
)
|
||
|
||
func (v *Query) execute(ctx context.Context, req *timeslice.Manager) error {
|
||
|
||
managerStartStr := req.StartTime.Format(time.DateTime)
|
||
managerEndStr := req.EndTime.Format(time.DateTime)
|
||
|
||
taskCount, err := timeslice.NewManager(v.callbackFunc).Run(ctx, req)
|
||
if err != nil {
|
||
log.Errorf("%s到%s,发生错误:%v", managerStartStr, managerEndStr, err)
|
||
}
|
||
|
||
fmt.Printf("%s到%s,总任务数:%d\n", managerStartStr, managerEndStr, taskCount)
|
||
log.Warnf("%s到%s,总任务数:%d", managerStartStr, managerEndStr, taskCount)
|
||
|
||
return nil
|
||
}
|
||
|
||
func (v *Query) callbackFunc(ctx context.Context, req *timeslice.Task) error {
|
||
|
||
startTimeStr := req.Process.Manager.StartTime.Format(time.DateTime)
|
||
endTimeStr := req.Process.Manager.EndTime.Format(time.DateTime)
|
||
|
||
currentStartTimeStr := req.CurrentStartTime.Format(time.DateTime)
|
||
currentEndTimeStr := req.CurrentEndTime.Format(time.DateTime)
|
||
|
||
start := time.Now()
|
||
startStr := start.Format(time.DateTime)
|
||
|
||
x := &do.WechatQuery{
|
||
StartTime: currentStartTimeStr,
|
||
EndTime: currentEndTimeStr,
|
||
ProductNo: req.Process.Manager.ProductNo,
|
||
}
|
||
|
||
n := 0
|
||
num := 0
|
||
notifyNum := 0
|
||
errNum := 0
|
||
|
||
err := v.orderRepo.FinSucByStockIdInBatches(ctx, x, func(ctx context.Context, rows []*bo.OrderBo) error {
|
||
|
||
n += 1
|
||
for _, order := range rows {
|
||
|
||
num += 1
|
||
if err := v.wechatQuery(ctx, order, ¬ifyNum); err != nil {
|
||
|
||
errNum += 1
|
||
|
||
logFields := map[string]string{
|
||
"order_no": order.OrderNo,
|
||
"coupon_id": order.VoucherNo,
|
||
"open_id": order.Account,
|
||
"err": err.Error(),
|
||
}
|
||
log.Errorf("%s到%s,第%d个任务，第%d组,发生错误:%+v", startTimeStr, endTimeStr, req.TaskID, n, logFields)
|
||
|
||
if errNum > 20 {
|
||
return fmt.Errorf("%s到%s,第%d个任务，已经连续发生20次错误%+v", startTimeStr, endTimeStr, req.TaskID, logFields)
|
||
}
|
||
}
|
||
|
||
}
|
||
|
||
return nil
|
||
})
|
||
|
||
end := time.Now()
|
||
|
||
logFields := map[string]interface{}{
|
||
"任务处理时间": currentStartTimeStr + "到" + currentEndTimeStr,
|
||
"总处理组数": n,
|
||
"总处理条数": num,
|
||
"总通知条数": notifyNum,
|
||
"执行任务开始时间": startStr,
|
||
"执行任务结束时间": end.Format(time.DateTime),
|
||
"总处理耗时": end.Sub(start).String(),
|
||
}
|
||
log.Warnf("%s到%s,第%d个任务,处理完毕:%+v", startTimeStr, endTimeStr, req.TaskID, logFields)
|
||
|
||
return err
|
||
}
|