Blender 三维绘图及渲染软件
Blender是一款免费开源的 3D 创作套件。使用 Blender,你可以创建3D可视化效果,例如静止图像,3D动画,V……...
1. 概述
gmq
是基于redis
提供的特性,使用go
语言开发的一个简单易用的队列;关于 redis 使用特性可以参考之前本人写过一篇很简陋的文章Redis 实现队列;gmq
的灵感和设计是基于有赞延迟队列设计,文章内容清晰而且很好理解,但是没有提供源码,在文章的最后也提到了一些未来架构方向;gmq
不是简单按照有赞延迟队列的设计实现功能,在它的基础上,做了一些修改和优化,主要如下:
功能上
多种任务模式,不单单只是延迟队列;例如:延迟队列,普通队列,优先级队列
架构上:
添加 job 由dispatcher
调度分配各个bucket
,而不是由timer
每个bucket
维护一个timer
,而不是所有bucket一个timer
timer
每次扫描bucket
到期job
时,会一次性返回多个到期job
,而不是每次只返回一个job
timer
的扫描时钟由bucket
中下个job
到期时间决定,而不是每秒扫描一次
2. 应用场景
延迟任务
延迟任务,例如用户下订单一直处于未支付状态,半个小时候自动关闭订单
异步任务
异步任务,一般用于耗时操作,例如群发邮件等批量操作
超时任务
规定时间内(TTR)
没有执行完毕或程序被意外中断,则消息重新回到队列再次被消费,一般用于数据比较敏感,不容丢失的
优先级任务
当多个任务同时产生时,按照任务设定等级优先被消费,例如a,b两种类型的job,优秀消费a,然后再消费b
3. 安装3.1 源码运行
配置文件位于gmq/conf.ini
,可以根据自己项目需求修改配置
cd $GOPATH/src # 进入gopath/src目录git clone https://github.com/wuzhc/gmq.gitcd gmqgo get -u -v github.com/kardianos/govendor # 如果有就不需要安装了govendor sync -v # 如果很慢,可能需要翻墙go run main.go start
3.2 执行文件运行
cd $GOPATH/src/gmq# 编译成可执行文件go build # 启动./gmq start# 停止./gmq stop# 守护进程模式启动,不输出日志到consolenohup ./gmq start >/dev/null 2>&1 &# 守护进程模式下查看日志输出(配置文件conf.ini需要设置target_type=file,filename=gmq.log)tail -f gmq.log
4. 客户端
目前只实现python,go,php语言的客户端的demo,参考:https://github.com/wuzhc/demo/tree/master/mq
运行
# php# 生产者php producer.php# 消费者php consumer.php# python# 生产者python producer.py# 消费者python consumer.py
一条消息结构
{ "id": "xxxx",# 任务id,这个必须是一个唯一值,将作为redis的缓存键 "topic": "xxx", # topic是一组job的分类名,消费者将订阅topic来消费该分类下的job "body": "xxx", # 消息内容 "delay": "111", # 延迟时间,单位秒 "TTR": "11111", # 执行超时时间,单位秒 "status": 1, # job执行状态,该字段由gmq生成 "consumeNum":1, # 被消费的次数,主要记录TTR>0时,被重复消费的次数,该字段由gmq生成}
延迟任务
$data = [ 'id' => 'xxxx_id' . microtime(true) . rand(1,999999999), 'topic' => ["topic_xxx"], 'body' => 'this is a rpc test', 'delay' => '1800', // 单位秒,半个小时后执行 'TTR' => '0' ];
超时任务
$data = [ 'id' => 'xxxx_id' . microtime(true) . rand(1,999999999), 'topic' => ["topic_xxx"], 'body' => 'this is a rpc test', 'delay' => '0', 'TTR' => '100' // 100秒后还未得到消费者ack确认,则再次添加到队列,将再次被被消费 ];
异步任务
$data = [ 'id' => 'xxxx_id' . microtime(true) . rand(1,999999999), 'topic' => ["topic_xxx"], 'body' => 'this is a rpc test', 'delay' => '0', 'TTR' => '0' ];
优先级任务
$data = [ 'id' => 'xxxx_id' . microtime(true) . rand(1,999999999), 'topic' => ["topic_A","topic_B","topic_C"], //优先消费topic_A,消费完后消费topic_B,最后再消费topic_C 'body' => 'this is a rpc test', 'delay' => '0', 'TTR' => '0' ];
5. gmq 流程图如下:
5.1 延迟时间 delay
当 job.delay>0时,job 会被分配到 bucket 中,bucket 会有周期性扫描到期 job,如果到期,job 会被 bucket 移到ready queue
,等待被消费
当 job.delay=0 时,job 会直接加到ready queue
,等待被消费
5.2 执行超时时间 TTR
参考第一个图的流程,当 job 被消费者读取后,如果job.TTR>0,
即 job 设置了执行超时时间,那么 job 会在读取后会被添加到 TTRBucket(专门存放设置了超时时间的 job),并且设置job.delay = job.TTR
,如果在 TTR 时间内没有得到消费者 ack 确认然后删除 job,job 将在 TTR 时间之后添加到ready queue
,然后再次被消费(如果消费者在 TTR 时间之后才请求 ack,会得到失败的响应)
5.3 确认机制
主要和TTR的设置有关系,确认机制可以分为两种:
当 job.TTR=0 时,消费者pop
出 job 时,即会自动删除job pool
中的 job 元数据
当 job.TTR>0 时,即 job 执行超时时间,这个时间是指用户pop
出 job 时开始到用户ack
确认删除结束这段时间,如果在这段时间没有ACK,
job 会被再次加入到ready queue
,然后再次被消费,只有用户调用了ACK,
才会去删除job pool
中 job 元数据
6. web 监控
gmq
提供了一个简单 web 监控平台(后期会提供根据 job.Id 追踪消息的功能),方便查看当前堆积任务数,默认监听端口为8000,
例如:http://127.0.0.1:8000,界面如下:后台模板来源于https://github.com/george518/PPGo_Job
7. 遇到问题
以下是开发遇到的问题,以及一些粗糙的解决方案
7.1 安全退出
如果强行中止gmq
的运行,可能会导致一些数据丢失,例如下面一个例子:如果发生上面的情况,就会出现 job 不在bucket
中,也不在ready queue,
这就出现了 job 丢失的情况,而且将没有任何机会去删除job pool
中已丢失的 job,长久之后job pool
可能会堆积很多的已丢失 job 的元数据;所以安全退出需要在接收到退出信号时,应该等待所有goroutine
处理完手中的事情,然后再退出
7.1.1gmq
退出流程
首先gmq
通过 context 传递关闭信号给dispatcher
,dispatcher
接收到信号会关闭dispatcher.closed
,每个bucket
会收到close
信号,然后先退出timer
检索,再退出bucket
,dispatcher
等待所有 bucket 退出后,然后退出
dispatcher
退出顺序流程:timer
->bucket
->dispatcher
7.1.2 注意
不要使用kill -9 pid
来强制杀死进程,因为系统无法捕获 SIGKILL 信号,导致 gmq 可能执行到一半就被强制中止,应该使用kill -15 pid
,kill -1 pid
或kill -2 pid
,各个数字对应信号如下:
9 对应 SIGKILL
15 对应 SIGTERM
1 对应 SIGHUP
2 对应 SIGINT
信号参考https://www.jianshu.com/p/5729fc095b2a
7.2 智能定时器
每一个bucket
都会维护一个timer
,不同于有赞设计,timer
不是每秒轮询一次,而是根据bucket
下一个 job 到期时间来设置timer
的定时时间 ,这样的目的在于如果bucket
没有 job 或 job 到期时间要很久才会发生,就可以减少不必要的轮询;
timer
只有处理完一次业务后才会重置定时器,这样的目的在于可能出现上一个时间周期还没执行完毕,下一个定时事件又发生了
如果到期的时间很相近,timer
就会频繁重置定时器时间,就目前使用来说,还没出现什么性能上的问题
7.3 原子性问题
我们知道 redis 的命令是排队执行,在一个复杂的业务中可能会多次执行 redis 命令,如果在大并发的场景下,这个业务有可能中间插入了其他业务的命令,导致出现各种各样的问题;redis 保证整个事务原子性和一致性问题一般用multi/exec
或lua脚本,
gmq
在操作涉及复杂业务时使用的是lua脚本,
因为lua 脚本
除了有multi/exec
的功能外,还有Pipepining
功能(主要打包命令,减少和redis server
通信次数),下面是一个gmq
定时器扫描 bucket 集合到期 job 的 lua 脚本:
-- 获取到期的50个joblocal jobIds = redis.call('zrangebyscore',KEYS[1], 0, ARGV[4], 'withscores', 'limit', 0, 50)local res = {}for k,jobId in ipairs(jobIds) do if k%2~=0 thenlocal jobKey = string.format('%s:%s', ARGV[3], jobId)local status = redis.call('hget', jobKey, 'status')-- 检验job状态if tonumber(status) == tonumber(ARGV[1]) or tonumber(status) == tonumber(ARGV[2]) then-- 先移除集合中到期的job,然后到期的job返回给timerlocal isDel = redis.call('zrem', KEYS[1], jobId)if isDel == 1 thentable.insert(res, jobId)endendendendlocal nextTime-- 计算下一个job执行时间,用于设置timer下一个时钟周期local nextJob = redis.call('zrange', KEYS[1], 0, 0, 'withscores')if next(nextJob) == nil thennextTime = -1elsenextTime = tonumber(nextJob[2]) - tonumber(ARGV[4])if nextTime < 0 thennextTime = 1endendtable.insert(res,1,nextTime)return res
7.4 redis 连接池
可能一般 phper 写业务很少会接触到连接池,其实这是由 php 本身所决定他应用不大,当然在 php 的扩展swoole
还是很有用处的gmq
的 redis 连接池是使用gomodule/redigo/redis
自带连接池,它带来的好处是限制 redis 连接数,通过复用 redis 连接来减少开销,另外可以防止 tcp 被消耗完,这在生产者大量生成数据时会很有用
// gmq/mq/redis.goRedis = &RedisDB{ Pool: &redis.Pool{ MaxIdle: 30, // 最大空闲链接 MaxActive: 10000, // 最大链接 IdleTimeout: 240 * time.Second, // 空闲链接超时 Wait: true, // 当连接池耗尽时,是否阻塞等待 Dial: func() (redis.Conn, error) { c, err := redis.Dial("tcp", "127.0.0.1:6379", redis.DialPassword("")) if err != nil { return nil, err } return c, nil}, TestOnBorrow: func(c redis.Conn, t time.Time) error { if time.Since(t) < time.Minute { return nil } _, err := c.Do("PING") return err }, },}
8. 注意问题
job.id 在job pool
是唯一的,它将作为 redis 的缓存键;gmq
不自动为 job 生成唯一 id 值是为了用户可以根据自己生成的 job.id 来追踪 job 情况,如果 job.id 是重复的,push 时会报重复 id 的错误
bucket 数量不是越多越好,一般来说,添加到ready queue
的速度取决与 redis 性能,而不是 bucket 数量
9. 使用中可能出现的问题9.1 客户端出现大量的 TIME_WAIT 状态,并且新的连接被拒绝
netstat -anp | grep 9503 | wc -ltcp 0 0 10.8.8.188:41482 10.8.8.185:9503 TIME_WAIT -
这个是正常现象,由 tcp 四次挥手可以知道,当接收到 LAST_ACK 发出的 FIN 后会处于TIME_WAIT
状态,主动关闭方(客户端)为了确保被动关闭方(服务端)收到 ACK,会等待 2MSL 时间,这个时间是为了再次发送 ACK,例如被动关闭方可能因为接收不到 ACK 而重传 FIN;另外也是为了旧数据过期,不影响到下一个链接;如果要避免大量TIME_WAIT
的连接导致 tcp 被耗尽,一般方法如下:
使用长连接
配置文件,网上很多教程,就是让系统尽快的回收TIME_WAIT
状态的连接
使用连接池,当连接池耗尽时,阻塞等待,直到回收再利用
10. 相关链接
有赞延迟队列设计
Redis 实现队列
Blender是一款免费开源的 3D 创作套件。使用 Blender,你可以创建3D可视化效果,例如静止图像,3D动画,V……...
Epoxy 是一个基于 RecyclerView 实现复杂屏幕效果的 Android 库。 Epoxy 通过将 ViewHolder、项目类型、……...
magnum 一个 OpenStack 项目,它提供容器编排引擎,以作为 OpenStack 中的一流资源来部署和管理容器。它……...
前言 进行移动端网页开发时,想要查看手机浏览器信息从来都不是一件容易的事。特别是当目标环境为APP内置……...
DophinPHP(海豚PHP)是一个基于ThinkPHP5.0.3开发的开源PHP快速开发框架,秉承极简、极速、极致的开发理……...
AdGuard 是一种快速、轻量级的广告拦截浏览器扩展程序,可有效拦截所有类型的广告和跟踪器。专注于高级隐……...
React(有时叫React.js或ReactJS)是 Facebook 推出的一个为数据提供渲染为 HTML 视图,用来构建用户界面的开源 JavaScript 库。 React 视图通常采用包含以自定义 HTML 标记规定的其他组件的组件渲染。React 为程……
Paramiko是用于建立 SSH2 连接(客户端或服务器)的库。重点是使用 SSH2 作为 SSL 的替代方法,以在 python 脚本之间建立安全连接。支持所有主要密码和哈希方法。也支持 SFTP 客户端和服务器模式。 示例代码: imp……
Reactive 是简单灵活的模板和绑定视图的引擎,支持自定义绑定,支持在模块修改的时候实时更新。 快速开始: varview=reactive('<p>Hello{name}!</p>',{name:'Adam'});//youcanaddtheview"element"toth……