combine good parts from other forks

Author: zouyi

README.md

@@ -1,8 +1,7 @@
 # delay-queue
-[![Go Report Card](https://goreportcard.com/badge/github.com/ouqiang/delay-queue)](https://goreportcard.com/report/github.com/ouqiang/delay-queue)
-[![Downloads](https://img.shields.io/github/downloads/ouqiang/delay-queue/total.svg)](https://github.com/ouqiang/delay-queue/releases)
-[![license](https://img.shields.io/github/license/mashape/apistatus.svg?maxAge=2592000)](https://github.com/ouqiang/delay-queue/blob/master/LICENSE)
-[![Release](https://img.shields.io/github/release/ouqiang/delay-queue.svg?label=Release)](https://github.com/ouqiang/delay-queue/releases)
+
+基于[ouqiang/delay-queue](https://github.com/ouqiang/delay-queue) [suhuaguo/delay-queue](https://github.com/suhuaguo/delay-queue) [moshuipan/delay-queue](https://github.com/moshuipan/delay-queue/) 代码并加入编译优化脚本
+
 
 基于Redis实现的延迟队列, 参考[有赞延迟队列设计](http://tech.youzan.com/queuing_delay)实现
 

build.sh

@@ -0,0 +1,15 @@
+mkdir -p dist
+if [[ `uname -s` == "Darwin" ]]; then
+    echo "system is macos"
+    CGO_ENABLED=0 GOOS=linux GOARCH=amd64 go build -ldflags '-w' -o dist/delay-queue-linux-64
+    CGO_ENABLED=0 GOOS=windows GOARCH=amd64 go build -ldflags '-w' -o dist/delay-queue-win-64
+    go build -ldflags '-w' -o dist/delay-queue-macos-64
+else
+    echo "system is macos"
+    CGO_ENABLED=0 GOOS=macos GOARCH=amd64 go build -ldflags '-w' -o dist/delay-queue-linux-64
+    CGO_ENABLED=0 GOOS=windows GOARCH=amd64 go build -ldflags '-w' -o dist/delay-queue-win-64
+    go build -ldflags '-w' -o dist/delay-queue-macos-64
+fi
+upx -9 dist/delay-queue-linux-64
+upx -9 dist/delay-queue-win-64
+upx -9 dist/delay-queue-macos-64

cmd/cmd.go

@@ -7,9 +7,9 @@ import (
 	"net/http"
 	"os"
 
-	"github.com/ouqiang/delay-queue/config"
-	"github.com/ouqiang/delay-queue/delayqueue"
-	"github.com/ouqiang/delay-queue/routers"
+	"github.com/php-cpm/delay-queue/config"
+	"github.com/php-cpm/delay-queue/delayqueue"
+	"github.com/php-cpm/delay-queue/routers"
 )
 
 // Cmd 应用入口Command
@@ -54,6 +54,7 @@ func (cmd *Cmd) parseCommandArgs() {
 
 // 运行Web Server
 func (cmd *Cmd) runWeb() {
+	// 有详细的说明：https://studygolang.com/resources/4657 讲的非常详细
 	http.HandleFunc("/push", routers.Push)
 	http.HandleFunc("/pop", routers.Pop)
 	http.HandleFunc("/finish", routers.Delete)

config/config.go

@@ -40,12 +40,14 @@ const (
 	DefaultRedisReadTimeout    = 180000
 	// DefaultRedisWriteTimeout Redis写入超时时间, 单位毫秒
 	DefaultRedisWriteTimeout   = 3000
+	// DefaultKeyName key名称
+	DefaultKeyName = "dq_key_"
 )
 
 // Config 应用配置
 type Config struct {
 	BindAddress       string      // http server 监听地址
-	BucketSize        int         // bucket数量
+	BucketSize        int         // bucket数量,timer 的数量
 	BucketName        string      // bucket在redis中的键名,
 	QueueName         string      // ready queue在redis中的键名
 	QueueBlockTimeout int         // 调用blpop阻塞超时时间, 单位秒, 修改此项, redis.read_timeout必须做相应调整
@@ -66,29 +68,36 @@ type RedisConfig struct {
 
 // Init 初始化配置
 func Init(path string) {
+	// &Config{} ：结构体初始化
+	// http://blog.csdn.net/xxx9001/article/details/52574501 结构体初始化
 	Setting = &Config{}
 	if path == "" {
+		// 加载默认的配置文件
 		Setting.initDefaultConfig()
 		return
 	}
 
+	// 转换配置文件
 	Setting.parse(path)
 }
 
 // 解析配置文件
 func (config *Config) parse(path string) {
+	// 加载配置文件
 	file, err := ini.Load(path)
 	if err != nil {
 		log.Fatalf("无法解析配置文件#%s", err.Error())
 	}
 
+	// bucket 参数
 	section := file.Section("")
 	config.BindAddress = section.Key("bind_address").MustString(DefaultBindAddress)
 	config.BucketSize = section.Key("bucket_size").MustInt(DefaultBucketSize)
 	config.BucketName = section.Key("bucket_name").MustString(DefaultBucketName)
 	config.QueueName = section.Key("queue_name").MustString(DefaultQueueName)
 	config.QueueBlockTimeout = section.Key("queue_block_timeout").MustInt(DefaultQueueBlockTimeout)
 
+	// redis 相关的参数
 	config.Redis.Host = section.Key("redis.host").MustString(DefaultRedisHost)
 	config.Redis.Db = section.Key("redis.db").MustInt(DefaultRedisDb)
 	config.Redis.Password = section.Key("redis.password").MustString(DefaultRedisPassword)
@@ -104,7 +113,7 @@ func (config *Config) initDefaultConfig() {
 	config.BindAddress = DefaultBindAddress
 	config.BucketSize = DefaultBucketSize
 	config.BucketName = DefaultBucketName
-	config.QueueName = DefaultQueueName
+	config.QueueName = DefaultQueueName    // ready queue
 	config.QueueBlockTimeout = DefaultQueueBlockTimeout
 
 	config.Redis.Host = DefaultRedisHost

delayqueue/bucket.go

@@ -6,8 +6,8 @@ import (
 
 // BucketItem bucket中的元素
 type BucketItem struct {
-	timestamp int64
-	jobId     string
+	timestamp int64  // 时间戳
+	jobId     string // jobId
 }
 
 // 添加JobId到bucket中
@@ -19,6 +19,7 @@ func pushToBucket(key string, timestamp int64, jobId string) error {
 
 // 从bucket中获取延迟时间最小的JobId
 func getFromBucket(key string) (*BucketItem, error) {
+	// 返回有序集 key 中，指定区间内的成员。其中成员的位置按 score 值递增(从小到大)来排序。
 	value, err := execRedisCommand("ZRANGE", key, 0, 0, "WITHSCORES")
 	if err != nil {
 		return nil, err

delayqueue/delay_queue.go

@@ -6,7 +6,7 @@ import (
 	"log"
 	"time"
 
-	"github.com/ouqiang/delay-queue/config"
+	"github.com/php-cpm/delay-queue/config"
 )
 
 var (
@@ -18,8 +18,14 @@ var (
 
 // Init 初始化延时队列
 func Init() {
+	// 初始化 redis 连接池
 	RedisPool = initRedisPool()
+
+	// 初始化一些列 Timer
 	initTimers()
+
+	// golang 函数：http://blog.csdn.net/mungo/article/details/52481285
+	// 进行初始化，产生一个 go routine
 	bucketNameChan = generateBucketName()
 }
 
@@ -34,6 +40,8 @@ func Push(job Job) error {
 		log.Printf("添加job到job pool失败#job-%+v#%s", job, err.Error())
 		return err
 	}
+
+	// 轮询的方式存放。ZADD 命令 。存放在有序集合中。将会从这里获取 要运行的 job
 	err = pushToBucket(<-bucketNameChan, job.Delay, job.Id)
 	if err != nil {
 		log.Printf("添加job到bucket失败#job-%+v#%s", job, err.Error())
@@ -45,7 +53,9 @@ func Push(job Job) error {
 
 // Pop 轮询获取Job
 func Pop(topics []string) (*Job, error) {
-	jobId, err := blockPopFromReadyQueue(topics, config.Setting.QueueBlockTimeout)
+	// ready queue 里面只有 topic  作为 key
+	// jobId, err := blockPopFromReadyQueue(topics, config.Setting.QueueBlockTimeout)
+	jobId, err := blPopFromReadyQueue(topics, config.Setting.QueueBlockTimeout)
 	if err != nil {
 		return nil, err
 	}
@@ -66,14 +76,20 @@ func Pop(topics []string) (*Job, error) {
 		return nil, nil
 	}
 
+	// 重新放到 Bucket 中，等待重新消费。实现至少一次的逻辑。如果客户端删除了 job ，那么。调度到此 jobId 的时候，发现 job 不存在，直接在 bucket 中删除
 	timestamp := time.Now().Unix() + job.TTR
-	err = pushToBucket(<-bucketNameChan, timestamp, job.Id)
+	// 表示从 <-bucketNameChan 。这个 channel 接收一个值
+	err = pushToBucket(fmt.Sprintf(config.Setting.BucketName, 1), timestamp, job.Id) //待确认的消息放入bucket1
 
 	return job, err
 }
 
 // Remove 删除Job
 func Remove(jobId string) error {
+	err := removeFromBucket(fmt.Sprintf(config.Setting.BucketName, 1), jobId)
+	if err != nil {
+		return err
+	}
 	return removeJob(jobId)
 }
 
@@ -93,10 +109,22 @@ func Get(jobId string) (*Job, error) {
 
 // 轮询获取bucket名称, 使job分布到不同bucket中, 提高扫描速度
 func generateBucketName() <-chan string {
+	// 阻塞 channel
 	c := make(chan string)
+	// 1、为什么这么写呢？为什么不直接写个 for 死循环呢？
+	// 如果直接写 for 循环，那在初始化的时候，会阻塞其他 init 函数。如果另起一个 go routine 的话，就不会阻塞其他的
+
+	// 2、每次都 产生一个 go routine 。是怎么销毁的呀？
+	// 因为把这个函数 赋给某个 变量了。在 init 中初始化了。只有一个 go routine 。
+
+	// 3、我感觉到 里面的 i 变量好像没有作用的呀，因为都没有和其他 go routine 交换。
+	// 因为在 初始化 init 一下。每次都从 bucketNameChan 这个 channel 读取信息。
 	go func() {
 		i := 1
+
+		// 死循环
 		for {
+			// chan <-  发送消息
 			c <- fmt.Sprintf(config.Setting.BucketName, i)
 			if i >= config.Setting.BucketSize {
 				i = 1
@@ -109,21 +137,27 @@ func generateBucketName() <-chan string {
 	return c
 }
 
-// 初始化定时器
+// 初始化定时器 https://yq.aliyun.com/articles/69303
 func initTimers() {
 	timers = make([]*time.Ticker, config.Setting.BucketSize)
 	var bucketName string
 	for i := 0; i < config.Setting.BucketSize; i++ {
+
+		 // 每 1s 执行一次
 		timers[i] = time.NewTicker(1 * time.Second)
+
+		// 如果这里部署多实例的话，就会产生竞争
 		bucketName = fmt.Sprintf(config.Setting.BucketName, i+1)
+
+		// 并发执行
 		go waitTicker(timers[i], bucketName)
 	}
 }
 
 func waitTicker(timer *time.Ticker, bucketName string) {
 	for {
 		select {
-		case t := <-timer.C:
+		case t := <-timer.C: //  我们启动一个新的goroutine，来以阻塞的方式从Timer的C这个channel中，等待接收一个值，这个值是到期的时间。
 			tickHandler(t, bucketName)
 		}
 	}
@@ -132,6 +166,7 @@ func waitTicker(timer *time.Ticker, bucketName string) {
 // 扫描bucket, 取出延迟时间小于当前时间的Job
 func tickHandler(t time.Time, bucketName string) {
 	for {
+		// 拿到第一个元素。bucket 存放 jobid 和时间戳
 		bucketItem, err := getFromBucket(bucketName)
 		if err != nil {
 			log.Printf("扫描bucket错误#bucket-%s#%s", bucketName, err.Error())
@@ -165,11 +200,12 @@ func tickHandler(t time.Time, bucketName string) {
 		if job.Delay > t.Unix() {
 			// 从bucket中删除旧的jobId
 			removeFromBucket(bucketName, bucketItem.jobId)
-			// 重新计算delay时间并放入bucket中
+			// 重新计算delay时间并放入其他的 bucket 中
 			pushToBucket(<-bucketNameChan, job.Delay, bucketItem.jobId)
 			continue
 		}
 
+		// 放到 Ready 队列中，普通的 redis list 即可。RPUSH 方式
 		err = pushToReadyQueue(job.Topic, bucketItem.jobId)
 		if err != nil {
 			log.Printf("JobId放入ready queue失败#bucket-%s#job-%+v#%s",

delayqueue/job.go

@@ -1,21 +1,22 @@
 package delayqueue
 
 import (
+	"github.com/php-cpm/delay-queue/config"
 	"github.com/vmihailenco/msgpack"
 )
 
 // Job 使用msgpack序列化后保存到Redis,减少内存占用
 type Job struct {
-	Topic string `json:"topic" msgpack:"1"`
-	Id    string `json:"id" msgpack:"2"`    // job唯一标识ID
+	Topic string `json:"topic" msgpack:"1"` // topic ，唯一
+	Id    string `json:"id" msgpack:"2"`    // job唯一标识ID 。客户端需要保证唯一性。有关联关系的
 	Delay int64  `json:"delay" msgpack:"3"` // 延迟时间, unix时间戳
-	TTR   int64  `json:"ttr" msgpack:"4"`
-	Body  string `json:"body" msgpack:"5"`
+	TTR   int64  `json:"ttr" msgpack:"4"` // 超时时间,TTR的设计目的是为了保证消息传输的可靠性。
+	Body  string `json:"body" msgpack:"5"`// body
 }
 
 // 获取Job
 func getJob(key string) (*Job, error) {
-	value, err := execRedisCommand("GET", key)
+	value, err := execRedisCommand("GET", config.DefaultKeyName+key)
 	if err != nil {
 		return nil, err
 	}
@@ -39,14 +40,14 @@ func putJob(key string, job Job) error {
 	if err != nil {
 		return err
 	}
-	_, err = execRedisCommand("SET", key, value)
+	_, err = execRedisCommand("SET", config.DefaultKeyName+key, value)
 
 	return err
 }
 
 // 删除Job
 func removeJob(key string) error {
-	_, err := execRedisCommand("DEL", key)
+	_, err := execRedisCommand("DEL", config.DefaultKeyName+key)
 
 	return err
 }