go-delay-queue

github项目位置

基于消息组件实现延迟队列

代码设计思路

github项目位置

1. 从上图看出，有四个主体：生产者、消费者、服务提供方、消息组件。先看消息组件，该方案基于组件实现，需要能够接入各个组件，那么可以就定义接口由各个组件库实现，有两个方法，也是消息队列组件最基本的功能，订阅、生产消息。

type Iqueue interface {
    // 订阅消息
    // topic 为队列名，dealFn为消息处理逻辑方法，[]byte为队列消息，当err发生则放回消息，默认为程序退出
	SubscribeMsg(topic string, dealFn func([]byte) (err error)) (err error)
	// 生产消息
	PublishMsg(topic string, msg []byte) (err error)
}

以redis实现为例 （其他实现:kafka ）

func (dr *Dredis) SubscribeMsg(topic string, dealFn func([]byte)(err error)) (err error){
	go func() {
		var tk = time.NewTicker(5 * time.Second)
		for ;; {
			// 获取队列消息
			if msg,err := dr.conn.RPop(topic).Bytes(); err == nil {
				err = dealFn(msg)
				if err != nil {
					// 程序退出，消息重新放回
					err = dr.conn.RPush(topic, msg).Err()
					if err != nil {
						// log
						log.Print("消息重新放回失败：", err)
					}
				}
			} else {
				if err != redis.Nil {
					// log
					log.Print("消息出队失败：", err)
				}
				<-tk.C
			}
		}
	}()
	return 
}

func  (dr *Dredis) PublishMsg(topic string, msg []byte) (err error){
	err = dr.conn.LPush(topic, msg).Err()
	if err != nil {
		// log
		log.Print("消息入队失败：", err)
	}
	return 
}

2. 生产者和消费者实质都为服务调用方，即客户端，那客户端和服务端之间的纽带就是消息，所以我们要构建消息队列，不同的消息队列代表了不同的延迟等级(1min,5min,1h等等)、处理逻辑等, 下面demo定义了三个等级（低中高）示例前往：

// 创建延迟消息
low := &go_delay_queue.DelayLevel{
	TopicName: "low", // 队列名
	Level:     0, // 延迟等级
	RetryNums: 1, // 重试次数，处理失败可重新放回该队列，超过次数就入到下一等级
	Ttl:       1 * time.Minute, // 消息延迟至少1min后可被使用
	DealFn: dealMsg, // 客户端消息处理逻辑方法
}
medium := &go_delay_queue.DelayLevel{
	TopicName: "medium",
	Level:     1,
	RetryNums: 1,
	Ttl:       5 * time.Minute,
	DealFn: dealMsg,
}
high := &go_delay_queue.DelayLevel{
	TopicName: "high",
	Level:     2,
	RetryNums: 1,
	Ttl:       1 * time.Hour,
	DealFn: dealMsg,
}

func dealMsg(dtm go_delay_queue.DelayTopicMsg) (err error) {
	fmt.Printf("开始处理消息：%#vn", dtm)
	return
}

3. 有了消息就可以把服务方和消息组件关联起来,延迟服务本身做的几个关键事

• 把延迟消息队列名交给对应消息接口实现，等待延迟消息

go dr.queuer.SubscribeMsg(item.TopicName, dr.dealMsg)

• 计算延迟消息过期时长，生效则将延迟消息传递给客户端处理

// 判断时间是否达到指定时间
var ttl = stru.ExpiredAt - dr.now
if dr.Debug {
	fmt.Println(fmt.Sprintf("消息未到达指定时间，等待 %dsn", ttl))
}
select {
	case <-time.After(time.Duration(ttl) * time.Second):
	case <-dr.clientCtx.Done():
		// 消息重回
		err = errors.New("reload")
		return err
	}
}

• 根据客户端处理结果，成功则丢弃消息，失败则重入消息

// 可以开始消费
go func(delayMsg *DelayTopicMsg) {
    err := dr.levelTopicMap[delayMsg.Level].DealFn(*delayMsg)
    if dr.Debug {
        fmt.Println("交付客户端处理:", err)
    }
    if err != nil {
        // 入下一等级消息
        dr.inQueue(true, delayMsg)
    }
}(stru)