Skip to content

Commit 35d75ae

Browse files
ouqiangouqiang
committed
升级到版本v0.4
1 parent 42a683b commit 35d75ae

File tree

9 files changed

+54
-18
lines changed

9 files changed

+54
-18
lines changed

‎README.md‎

Lines changed: 6 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,4 +1,9 @@
11
# delay-queue
2+
[![Go Report Card](https://goreportcard.com/badge/github.com/ouqiang/delay-queue)](https://goreportcard.com/report/github.com/ouqiang/delay-queue)
3+
[![Downloads](https://img.shields.io/github/downloads/ouqiang/delay-queue/total.svg)](https://github.com/ouqiang/delay-queue/releases)
4+
[![license](https://img.shields.io/github/license/mashape/apistatus.svg?maxAge=2592000)](https://github.com/ouqiang/delay-queue/blob/master/LICENSE)
5+
[![Release](https://img.shields.io/github/release/ouqiang/delay-queue.svg?label=Release)](https://github.com/ouqiang/delay-queue/releases)
6+
27
基于Redis实现的延迟队列, 参考[有赞延迟队列设计](http://tech.youzan.com/queuing_delay)实现
38

49
## 应用场景
@@ -30,7 +35,7 @@
3035
* Redis
3136

3237
## 下载
33-
[v0.3](https://github.com/ouqiang/delay-queue/releases/tag/v0.3)
38+
[releases](https://github.com/ouqiang/delay-queue/releases)
3439

3540

3641
## 源码安装

‎cmd/cmd.go‎

Lines changed: 7 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -5,14 +5,14 @@ import (
55
"fmt"
66
"log"
77
"net/http"
8-
_ "net/http/pprof"
98
"os"
109

1110
"github.com/ouqiang/delay-queue/config"
1211
"github.com/ouqiang/delay-queue/delayqueue"
1312
"github.com/ouqiang/delay-queue/routers"
1413
)
1514

15+
// Cmd 应用入口Command
1616
type Cmd struct{}
1717

1818
var (
@@ -21,9 +21,11 @@ var (
2121
)
2222

2323
const (
24-
AppVersion = "0.3"
24+
// AppVersion 应用版本号
25+
AppVersion = "0.4"
2526
)
2627

28+
// Run 运行应用
2729
func (cmd *Cmd) Run() {
2830

2931
// 解析命令行参数
@@ -41,6 +43,7 @@ func (cmd *Cmd) Run() {
4143
cmd.runWeb()
4244
}
4345

46+
// 解析命令行参数
4447
func (cmd *Cmd) parseCommandArgs() {
4548
// 配置文件
4649
flag.StringVar(&configFile, "c", "", "./delay-queue -c /path/to/delay-queue.conf")
@@ -49,6 +52,7 @@ func (cmd *Cmd) parseCommandArgs() {
4952
flag.Parse()
5053
}
5154

55+
// 运行Web Server
5256
func (cmd *Cmd) runWeb() {
5357
http.HandleFunc("/push", routers.Push)
5458
http.HandleFunc("/pop", routers.Pop)
@@ -58,7 +62,5 @@ func (cmd *Cmd) runWeb() {
5862

5963
log.Printf("listen %s\n", config.Setting.BindAddress)
6064
err := http.ListenAndServe(config.Setting.BindAddress, nil)
61-
if err != nil {
62-
log.Fatalln(err)
63-
}
65+
log.Fatalln(err)
6466
}

‎config/config.go‎

Lines changed: 19 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -9,25 +9,40 @@ import (
99
// 解析配置文件
1010

1111
var (
12+
// Setting 配置实例
1213
Setting *Config
1314
)
1415

1516
const (
17+
// DefaultBindAddress 监听地址
1618
DefaultBindAddress = "0.0.0.0:9277"
19+
// DefaultBucketSize bucket数量
1720
DefaultBucketSize = 3
21+
// DefaultBucketName bucket名称
1822
DefaultBucketName = "dq_bucket_%d"
23+
// DefaultQueueName 队列名称
1924
DefaultQueueName = "dq_queue_%s"
25+
// DefaultQueueBlockTimeout 轮询队列超时时间
2026
DefaultQueueBlockTimeout = 178
27+
// DefaultRedisHost Redis连接地址
2128
DefaultRedisHost = "127.0.0.1:6379"
29+
// DefaultRedisDb Redis数据库编号
2230
DefaultRedisDb = 1
31+
// DefaultRedisPassword Redis密码
2332
DefaultRedisPassword = ""
33+
// DefaultRedisMaxIdle Redis连接池闲置连接数
2434
DefaultRedisMaxIdle = 10
35+
// DefaultRedisMaxActive Redis连接池最大激活连接数, 0为不限制
2536
DefaultRedisMaxActive = 0
37+
// DefaultRedisConnectTimeout Redis连接超时时间,单位毫秒
2638
DefaultRedisConnectTimeout = 5000
39+
// DefaultRedisReadTimeout Redis读取超时时间, 单位毫秒
2740
DefaultRedisReadTimeout = 180000
41+
// DefaultRedisWriteTimeout Redis写入超时时间, 单位毫秒
2842
DefaultRedisWriteTimeout = 3000
2943
)
3044

45+
// Config 应用配置
3146
type Config struct {
3247
BindAddress string // http server 监听地址
3348
BucketSize int // bucket数量
@@ -37,6 +52,7 @@ type Config struct {
3752
Redis RedisConfig // redis配置
3853
}
3954

55+
// RedisConfig Redis配置
4056
type RedisConfig struct {
4157
Host string
4258
Db int
@@ -48,6 +64,7 @@ type RedisConfig struct {
4864
WriteTimeout int // 写入超时, 单位毫秒
4965
}
5066

67+
// Init 初始化配置
5168
func Init(path string) {
5269
Setting = &Config{}
5370
if path == "" {
@@ -58,6 +75,7 @@ func Init(path string) {
5875
Setting.parse(path)
5976
}
6077

78+
// 解析配置文件
6179
func (config *Config) parse(path string) {
6280
file, err := ini.Load(path)
6381
if err != nil {
@@ -81,6 +99,7 @@ func (config *Config) parse(path string) {
8199
config.Redis.WriteTimeout = section.Key("redis.write_timeout").MustInt(DefaultRedisWriteTimeout)
82100
}
83101

102+
// 初始化默认配置
84103
func (config *Config) initDefaultConfig() {
85104
config.BindAddress = DefaultBindAddress
86105
config.BucketSize = DefaultBucketSize

‎delayqueue/bucket.go‎

Lines changed: 5 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -4,17 +4,20 @@ import (
44
"strconv"
55
)
66

7+
// BucketItem bucket中的元素
78
type BucketItem struct {
89
timestamp int64
910
jobId string
1011
}
1112

13+
// 添加JobId到bucket中
1214
func pushToBucket(key string, timestamp int64, jobId string) error {
1315
_, err := execRedisCommand("ZADD", key, timestamp, jobId)
1416

1517
return err
1618
}
1719

20+
// 从bucket中获取延迟时间最小的JobId
1821
func getFromBucket(key string) (*BucketItem, error) {
1922
value, err := execRedisCommand("ZRANGE", key, 0, 0, "WITHSCORES")
2023
if err != nil {
@@ -31,12 +34,13 @@ func getFromBucket(key string) (*BucketItem, error) {
3134
}
3235
timestampStr := string(valueBytes[1].([]byte))
3336
item := &BucketItem{}
34-
item.timestamp, err = strconv.ParseInt(timestampStr, 10, 64)
37+
item.timestamp, _ = strconv.ParseInt(timestampStr, 10, 64)
3538
item.jobId = string(valueBytes[0].([]byte))
3639

3740
return item, nil
3841
}
3942

43+
// 从bucket中删除JobId
4044
func removeFromBucket(bucket string, jobId string) error {
4145
_, err := execRedisCommand("ZREM", bucket, jobId)
4246

‎delayqueue/delay_queue.go‎

Lines changed: 5 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -16,13 +16,14 @@ var (
1616
bucketNameChan <-chan string
1717
)
1818

19+
// Init 初始化延时队列
1920
func Init() {
2021
RedisPool = initRedisPool()
2122
initTimers()
2223
bucketNameChan = generateBucketName()
2324
}
2425

25-
// 添加一个Job到队列中
26+
// Push 添加一个Job到队列中
2627
func Push(job Job) error {
2728
if job.Id == "" || job.Topic == "" || job.Delay < 0 || job.TTR <= 0 {
2829
return errors.New("invalid job")
@@ -42,7 +43,7 @@ func Push(job Job) error {
4243
return nil
4344
}
4445

45-
// 获取Job
46+
// Pop 轮询获取Job
4647
func Pop(topics []string) (*Job, error) {
4748
jobId, err := blockPopFromReadyQueue(topics, config.Setting.QueueBlockTimeout)
4849
if err != nil {
@@ -71,12 +72,12 @@ func Pop(topics []string) (*Job, error) {
7172
return job, err
7273
}
7374

74-
// 删除Job
75+
// Remove 删除Job
7576
func Remove(jobId string) error {
7677
return removeJob(jobId)
7778
}
7879

79-
// 查询Job
80+
// Get 查询Job
8081
func Get(jobId string) (*Job, error) {
8182
job, err := getJob(jobId)
8283
if err != nil {

‎delayqueue/job.go‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -4,6 +4,7 @@ import (
44
"github.com/vmihailenco/msgpack"
55
)
66

7+
// Job 使用msgpack序列化后保存到Redis,减少内存占用
78
type Job struct {
89
Topic string `json:"topic" msgpack:"1"`
910
Id string `json:"id" msgpack:"2"` // job唯一标识ID

‎delayqueue/ready_queue.go‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -6,15 +6,15 @@ import (
66
"github.com/ouqiang/delay-queue/config"
77
)
88

9-
type ReadyQueue struct{}
10-
9+
// 添加JobId到队列中
1110
func pushToReadyQueue(queueName string, jobId string) error {
1211
queueName = fmt.Sprintf(config.Setting.QueueName, queueName)
1312
_, err := execRedisCommand("RPUSH", queueName, jobId)
1413

1514
return err
1615
}
1716

17+
// 从队列中阻塞获取JobId
1818
func blockPopFromReadyQueue(queues []string, timeout int) (string, error) {
1919
var args []interface{}
2020
for _, queue := range queues {

‎delayqueue/redis.go‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -9,6 +9,7 @@ import (
99
)
1010

1111
var (
12+
// RedisPool RedisPool连接池实例
1213
RedisPool *redis.Pool
1314
)
1415

‎routers/routers.go‎

Lines changed: 8 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -11,15 +11,17 @@ import (
1111
"github.com/ouqiang/delay-queue/delayqueue"
1212
)
1313

14+
// TopicRequest Job类型请求json
1415
type TopicRequest struct {
1516
Topic string `json:"topic"`
1617
}
1718

19+
// IdRequest JobId请求json
1820
type IdRequest struct {
1921
Id string `json:"id"`
2022
}
2123

22-
// 添加job
24+
// Push 添加job
2325
func Push(resp http.ResponseWriter, req *http.Request) {
2426
var job delayqueue.Job
2527
err := readBody(resp, req, &job)
@@ -54,14 +56,14 @@ func Push(resp http.ResponseWriter, req *http.Request) {
5456
err = delayqueue.Push(job)
5557

5658
if err != nil {
57-
log.Printf("添加job失败", err.Error())
59+
log.Printf("添加job失败#%s", err.Error())
5860
resp.Write(generateFailureBody(err.Error()))
5961
} else {
6062
resp.Write(generateSuccessBody("添加成功", nil))
6163
}
6264
}
6365

64-
// 获取job
66+
// Pop 获取job
6567
func Pop(resp http.ResponseWriter, req *http.Request) {
6668
var topicRequest TopicRequest
6769
err := readBody(resp, req, &topicRequest)
@@ -102,7 +104,7 @@ func Pop(resp http.ResponseWriter, req *http.Request) {
102104
resp.Write(generateSuccessBody("操作成功", data))
103105
}
104106

105-
// 删除job
107+
// Delete 删除job
106108
func Delete(resp http.ResponseWriter, req *http.Request) {
107109
var idRequest IdRequest
108110
err := readBody(resp, req, &idRequest)
@@ -126,7 +128,7 @@ func Delete(resp http.ResponseWriter, req *http.Request) {
126128
resp.Write(generateSuccessBody("操作成功", nil))
127129
}
128130

129-
// 查询job
131+
// Get 查询job
130132
func Get(resp http.ResponseWriter, req *http.Request) {
131133
var idRequest IdRequest
132134
err := readBody(resp, req, &idRequest)
@@ -153,6 +155,7 @@ func Get(resp http.ResponseWriter, req *http.Request) {
153155
resp.Write(generateSuccessBody("操作成功", job))
154156
}
155157

158+
// ResponseBody 响应Body格式
156159
type ResponseBody struct {
157160
Code int `json:"code"`
158161
Message string `json:"message"`

0 commit comments

Comments
 (0)