// Msg 消息
type Msg struct {
   Topic     string        // 消息的主題
   Key       string        // 消息的Key
   Body      []byte        // 消息的Body
   Delay     time.Duration // 延遲時(shí)間（秒）
   ReadyTime time.Time     // 消息準(zhǔn)備好執(zhí)行的時(shí)間（now + delay）
}

const delayQueuePushRedisScript = `
-- KEYS[1]: topicZSet
-- KEYS[2]: topicHash
-- ARGV[1]: 消息的Key
-- ARGV[2]: 消息的Body
-- ARGV[3]: 消息準(zhǔn)備好執(zhí)行的時(shí)間

local topicZSet = KEYS[1]
local topicHash = KEYS[2]
local key = ARGV[1]
local body = ARGV[2]
local readyTime = tonumber(ARGV[3])

-- 添加readyTime到zset
local count = redis.call("zadd", topicZSet, readyTime, key)
-- 消息已經(jīng)存在
if count == 0 then 
   return 0
end
-- 添加body到hash
redis.call("hsetnx", topicHash, key, body)
return 1
`

func (q *SimpleRedisDelayQueue) Push(ctx context.Context, msg *Msg) error {
   // 如果設(shè)置了ReadyTime，就使用RedisTime
   var readyTime int64
   if !msg.ReadyTime.IsZero() {
      readyTime = msg.ReadyTime.Unix()
   } else {
      // 否則使用Delay
      readyTime = time.Now().Add(msg.Delay).Unix()
   }
   success, err := q.pushScript.Run(ctx, q.client, []string{q.topicZSet(msg.Topic), q.topicHash(msg.Topic)},
      msg.Key, msg.Body, readyTime).Bool()
   if err != nil {
      return err
   }
   if !success {
      return ErrDuplicateMessage
   }
   return nil
}

const delayQueueDelRedisScript = `
-- KEYS[1]: topicZSet
-- KEYS[2]: topicHash
-- ARGV[1]: 消息的Key

local topicZSet = KEYS[1]
local topicHash = KEYS[2]
local key = ARGV[1]

-- 刪除zset和hash關(guān)于這條消息的內(nèi)容
redis.call("zrem", topicZSet, key)
redis.call("hdel", topicHash, key)
return 1
`

func (q *SimpleRedisDelayQueue) Consume(topic string, batchSize int, fn func(msg *Msg) error) {
   for {
      // 批量獲取已經(jīng)準(zhǔn)備好執(zhí)行的消息
      now := time.Now().Unix()
      zs, err := q.client.ZRangeByScoreWithScores(context.Background(), q.topicZSet(topic), &redis.ZRangeBy{
         Min:   "-inf",
         Max:   strconv.Itoa(int(now)),
         Count: int64(batchSize),
      }).Result()
      // 如果獲取出錯(cuò)或者獲取不到消息，則休眠一秒
      if err != nil || len(zs) == 0 {
         time.Sleep(time.Second)
         continue
      }
      // 遍歷每個(gè)消息
      for _, z := range zs {
         key := z.Member.(string)
         // 獲取消息的body
         body, err := q.client.HGet(context.Background(), q.topicHash(topic), key).Bytes()
         if err != nil {
            continue
         }

         // 處理消息
         err = fn(&Msg{
            Topic:     topic,
            Key:       key,
            Body:      body,
            ReadyTime: time.Unix(int64(z.Score), 0),
         })
         if err != nil {
            continue
         }

         // 如果消息處理成功，刪除消息
         q.delScript.Run(context.Background(), q.client, []string{q.topicZSet(topic), q.topicHash(topic)}, key)
      }
   }
}

// Msg 消息
type Msg struct {
   Topic     string        // 消息的主題
   Key       string        // 消息的Key
   Body      []byte        // 消息的Body
   Partition int           // 分區(qū)號(hào)
   Delay     time.Duration // 延遲時(shí)間（秒）
   ReadyTime time.Time     // 消息準(zhǔn)備好執(zhí)行的時(shí)間
}

func (q *PartitionRedisDelayQueue) topicZSet(topic string, partition int) string {
   return fmt.Sprintf("%s:%d:z", topic, partition)
}

func (q *PartitionRedisDelayQueue) topicHash(topic string, partition int) string {
   return fmt.Sprintf("%s:%d:h", topic, partition)
}

func (q *PartitionRedisDelayQueue) Consume(topic string, batchSize, partition int, fn func(msg *Msg) error) {
    // ...
}