基于golang的简单分布式延时队列服务的实现

curl -d 
'{
  "topic": "xxx", 								// 业务topic
  "timing_moment": ,							    // 单位秒,要定时时刻
  "content": "{}"								// 消息体,json串
}'
'http://127.0.0.1:xxxx/delay/task/add'

{
  "dm_error": 0,
  "error_msg": "操作成功",
  "task_id":112345465765
}

curl -d 
'{
  "topic": "xxxx", 								// 业务topic
  "task_id":1324568798765							// taskid,选填，有则返回特定消息
}'
'http://127.0.0.1:xxxx/delay/task/pull'

{
  "dm_error": 0,
  "error_msg": "操作成功"
  "content":"{"\xxx"\}"
}

curl -d 
'{
  "topic": "xxx", 						// 业务topic
  "timing_moment": ,						// 单位秒,要定时时刻
  "content": "{                        // 消息内容(json string)
	"sn":"message.call",                  // 服务发现名字（或为配置服务名）
	"url":"/ev/tp/xxxx",                  // 回调url
	"xxx":"xxx"                       // 其他字段
  }"
}'
'http://127.0.0.1:xxxx/delay/task/add'

curl -d '{
  "topic":"xxxx_push",
  "content":"{
    "uid":"111111",
    "sn":"other.server",
    "url":"/xxxx/callback",
    "msg_type":"gift",
  }",
  "timing_moment":1565700615
}' 
http://127.0.0.1:xxxx/delay/task/add

{
  "dm_error": 0,
  "error_msg": "操作成功",
  "task_id":112345465765
}

topic: delay_base_push

固定返回格式
{
  "topic": "xxxx",								// 业务topic
  "content": "{}"								// 单条生产消息content
}

func (c *CacheManager) OperGenTaskid() (uint64, error) {
	now := time.Now().Unix()
	key := c.getDelayTaskIdKey()
	reply, err := c.DelayRds.Do("INCR", key)
	if err != nil {
		log.Errorf("genTaskid INCR key:%s, error:%s", key, err)
		return 0, err
	}
	version := reply.(int64)
	if version == 1 {
    //默认认为1秒能创建100个任务
		c.DelayRds.Expire(key, time.Duration(100)*time.Second)
	}
	incrNum := version % 10000
	taskId := (uint64(now)*10000 + uint64(incrNum))
	log.Debugf("genTaskid INCR key:%s, taskId:%d", key, taskId)
	return taskId, nil
}

func (c *CacheManager) SetDelayTopicLock(ctx context.Context, topic string) (bool, error) {
	key := c.getDelayTopicReloadLockKey(topic)
	reply, err := c.DelayRds.Do("SET", key, "lock", "NX", "EX", 2)
	if err != nil {
		log.Errorf("SetDelayTopicLock SETNX key:%s, cal:%v, error:%s", key, "lock", err)
		return false, err
	}
	if reply == nil {
		return false, nil
	}
	log.Debugf("SetDelayTopicLock SETNXEX topic:%s lock:%d", topic, false)
	return true, nil
}

for {
  time.Sleep(time.Second)
  fmt.Println("test")
}

t1:=time.Tick(3*time.Second)
for {
  select {
  case <-t1:
    fmt.Println("test")
  }
}

t:=time.NewTicker(time.Second)
for {
  select {
  case <-t.C:
    fmt.Println("test")
    t.Stop()
  }
}