ouqiang / delay-queue Goto Github PK

View Code? Open in Web Editor NEW

1.2K 47.0 241.0 115 KB

延迟队列

License: MIT License

Go 100.00%

delayqueue delayed-job redis golang delay-queue

delay-queue's Introduction

delay-queue

基于Redis实现的延迟队列, 参考有赞延迟队列设计实现

应用场景

订单超过30分钟未支付，自动关闭
订单完成后, 如果用户一直未评价, 5天后自动好评
会员到期前15天, 到期前3天分别发送短信提醒

支付宝异步通知实现

支付宝异步通知时间间隔是如何实现的(通知的间隔频率一般是：2m,10m,10m,1h,2h,6h,15h)

订单支付成功后, 生成通知任务, 放入消息队列中.
任务内容包含Array{0,0,2m,10m,10m,1h,2h,6h,15h}和通知到第几次N(这里N=1, 即第1次).
消费者从队列中取出任务, 根据N取得对应的时间间隔为0, 立即发送通知.

第1次通知失败, N += 1 => 2
从Array中取得间隔时间为2m, 添加一个延迟时间为2m的任务到延迟队列, 任务内容仍包含Array和N

第2次通知失败, N += 1 => 3, 取出对应的间隔时间10m, 添加一个任务到延迟队列, 同上
......
第7次通知失败, N += 1 => 8, 取出对应的间隔时间15h, 添加一个任务到延迟队列, 同上
第8次通知失败, N += 1 => 9, 取不到间隔时间, 结束通知

实现原理

利用Redis的有序集合，member为JobId, score为任务执行的时间戳,
每秒扫描一次集合，取出执行时间小于等于当前时间的任务.

依赖

Redis

下载

releases

源码安装

go语言版本1.7+
go get -d github.com/ouqiang/delay-queue
go build

运行

./delay-queue -c delay-queue.conf

HTTP Server监听0.0.0.0:9277, Redis连接地址127.0.0.1:6379, 数据库编号1

客户端

PHP

HTTP接口

请求方法 POST
请求Body及返回值均为json

返回值

{
  "code": 0,
  "message": "添加成功",
  "data": null
}

参数名	类型	含义	备注
code	int	状态码	0: 成功非0: 失败
message	string	状态描述信息
data	object, null	附加信息

添加任务

URL地址 /push

{
  "topic": "order",
  "id": "15702398321",
  "delay": 3600,
  "ttr": 120,
  "body": "{\"uid\": 10829378,\"created\": 1498657365 }"
}

参数名	类型	含义	备注
topic	string	Job类型
id	string	Job唯一标识	需确保JobID唯一
delay	int	Job需要延迟的时间, 单位：秒
ttr	int	Job执行超时时间, 单位：秒
body	string	Job的内容，供消费者做具体的业务处理，如果是json格式需转义

轮询队列获取任务

服务端会Hold住连接, 直到队列中有任务或180秒后超时返回,
任务执行完成后需调用finish接口删除任务, 否则任务会重复投递, 消费端需能处理同一任务的多次投递

URL地址 /pop

{
  "topic": "order"
}

参数名	类型	含义	备注
topic	string	Job类型

队列中有任务返回值

{
  "code": 0,
  "message": "操作成功",
  "data": {
    "id": "15702398321",
    "body": "{\"uid\": 10829378,\"created\": 1498657365 }"
  }
}

队列为空返回值

{
  "code": 0,
  "message": "操作成功",
  "data": null
}

删除任务

URL地址 /delete

{
  "id": "15702398321"
}

参数名	类型	含义	备注
id	string	Job唯一标识

完成任务

URL地址 /finish

{
  "id": "15702398321"
}

参数名	类型	含义	备注
id	string	Job唯一标识

查询任务

URL地址 /get

{
  "id": "15702398321"
}

参数名	类型	含义	备注
id	string	Job唯一标识

返回值

{
    "code": 0,
    "message": "操作成功",
    "data": {
        "topic": "order",
        "id": "15702398321",
        "delay": 1506787453,
        "ttr": 60,
        "body": "{\"uid\": 10829378,\"created\": 1498657365 }"
    
    }
}

参数名	类型	含义
topic	string	Job类型
id	string	Job唯一标识
delay	int	Job延迟执行的时间戳
ttr	int	Job执行超时时间, 单位：秒
body	string	Job内容，供消费者做具体的业务处理

Job不存在返回值

{
  "code": 0,
  "message": "操作成功",
  "data": null
}

delay-queue's People

Contributors

Stargazers

Watchers

Forkers

shiftwinting guoqing1988 andy521 zhanglei poppysix hulkwan clearluo daniel-007 cluo devopsmi superdubbo yangchen92 francissoung suhuaguo larrychina gmqllf blackhilloldmonster dewey363 ahao123 chakra-coder phonehold moshuipan jululy yuanleilei wechatpu yecao007 znlnzi dikang123 dragonriver1990 flylee85 my-golang halloheihei huangdehui2013 javacodemood smalltoken jangocheng jovichin kingpro zouhuigang qfz9527 chiwenheng jaciyu greatxj wormcc onehands howepeng xiaozhao1795 php-cpm youthlab summerhearts happyiwin smll8402 xywhnh renyuanwei ddwapple fanmeilexiu 15393702382 viyond paypei kangxi94 zhwei820 xusoybean jpfss 1407010218 jash16 coldinfire aland-zhang skyformat99 microchick yixiaoyi hisunwei chenglei12345 renyinew elambda sunshinecolor ycqqq nangualiang getfloret cailin186 ecosakurasou gitbufenshuo auto124 charliexp percyzhang alberthchang wyyrepo fishonline thomasauto sysusr kenvinwei mrsjiuer kissthink rpdict uglyghost07 ld140319 scofier twomiao apple006 danrui l5071134

delay-queue's Issues

建议

delay_queue.go 添加一个job.Id查询Job的和添加一个查询队列数量的两个接口方便管理和查询

代理的逻辑有问题

	// 再次确认元信息中delay是否小于等于当前时间
	if job.Delay > t.Unix() {  // 这里的逻辑有问题。
		// 重新计算delay时间并放入bucket中
		pushToBucket(<-bucketNameChan, job.Delay, bucketItem.jobId)

		// 从bucket中删除之前的bucket
		removeFromBucket(bucketName, bucketItem.jobId)

		continue
	}

在 delay_queue.go 中，有如上的代码片段。if job.Delay > t.Unix() 这个写的不对。
应该是：if job.Delay < t.Unix()

因为，如果延迟时间大于当前时间了，应该立马投递到 ready queue 里面，供消费端消费了？
不知道理解的是否正确？

同一topic下，若多个job delay相同，loop时是否有冲突
业务方jobid必须要保证唯一，若有多个业务方，服务端是否要保证全局jobid的唯一性？（结合redis）
依靠客户端不断轮训获取延迟job，是否采用pub/sub方式更好？
期待您的回复！

你这业务逻辑在哪写？

如果出现原子性问题怎么解决呢

比如putJob成功了，但是pushToBucket失败了，是需要client端手动去删除job吗