hdt3213 / delayqueue Goto Github PK

func (q *DelayQueue) SendScheduleMsgs(payloads []string, t []time.Time, opts ...interface{}) error {
	retryCount := q.defaultRetryCount
	for _, opt := range opts {
		switch o := opt.(type) {
		case retryCountOpt:
			retryCount = uint(o)
		case msgTTLOpt:
			q.msgTTL = time.Duration(o)
		}
	}
	pipe := q.redisCli.TxPipeline() // 这里是在warpper里添加的
	now := time.Now()
	ctx := context.Background()
	for i := 0; i < len(t); i++ {
		idStr := uuid.Must(uuid.NewRandom()).String()
		msgTTL := t[i].Sub(now) + q.msgTTL
		pipe.Set(ctx, q.genMsgKey(idStr), payloads[i], msgTTL)
		pipe.HSet(ctx, q.retryCountKey, idStr, strconv.Itoa(int(retryCount)))
		values := map[string]float64{idStr: float64(t[i].Unix())}
		var zs []redis.Z
		for member, score := range values {
			zs = append(zs, redis.Z{
				Score:  score,
				Member: member,
			})
		}
		pipe.ZAdd(ctx, q.pendingKey, zs...)
	}
	_, err := pipe.Exec(ctx)
	if err != nil {
		return fmt.Errorf("push to pending failed: %v", err)
	}
	return nil
}