基于Redis实现分布式消息队列(一)

1、为什么需要消息队列? 当系统中出现“生产“和“消费“的速度或稳定性等因素不一致的时候,就需要消息队列,作为抽象层,弥合双方的差异。

举个例子:业务系统触发短信发送申请,但短信发送模块速度跟不上,需要将来不及处理的消息暂存一下,缓冲压力。 再举个例子:调远程系统下订单成本较高,且因为网络等因素,不稳定,攒一批一起发送。 再举个栗子,交互模块5:00到24:00和电商系统联通,和内部ERP断开。1:00到4:00和ERP联通,和电商系统断开。 再举个例子,服务员点菜快,厨师做菜慢。 再举个例子,到银行办事的人多,提供服务的窗口少。 乖乖排队吧。

2、使用消息队列有什么好处? 2.1、提高系统响应速度 使用了消息队列,生产者一方,把消息往队列里一扔,就可以立马返回,响应用户了。无需等待处理结果。

处理结果可以让用户稍后自己来取,如医院取化验单。也可以让生产者订阅(如:留下手机号码或让生产者实现listener接口、加入监听队列),有结果了通知。获得约定将结果放在某处,无需通知。

2.2、提高系统稳定性 考虑电商系统下订单,发送数据给生产系统的情况。 电商系统和生产系统之间的网络有可能掉线,生产系统可能会因维护等原因暂停服务。

如果不使用消息队列,电商系统数据发布出去,顾客无法下单,影响业务开展。 两个系统间不应该如此紧密耦合。应该通过消息队列解耦。同时让系统更健壮、稳定。

3、为什么需要分布式? 3.1、多系统协作需要分布式 消息队列中的数据需要在多个系统间共享数据才能发挥价值。 所以必须提供分布式通信机制、协同机制。

3.2、单系统内部署环境需要分布式 单系统内部,为了更好的性能、为了避免单点故障,多为集群环境。 集群环境中,应用运行在多台服务器的多个JVM中;数据也保存在各种类型的数据库或非数据库的多个节点上。 为了满足多节点协作需要,需要提供分布式的解决方案。

4、分布式环境下需要解决哪些问题 4.1、并发问题 需进行良好的并发控制。确保“线程安全“。

不要出现一个订单被出货两次。不要出现顾客A下的单,发货发给了顾客B等情况。

4.2、简单的、统一的操作机制 需定义简单的,语义明确的,业务无关的,恰当稳妥的统一的访问方式。

4.3、容错 控制好单点故障,确保数据安全。

4.4、可横向扩展 可便捷扩容。

5、如何实现? 成熟的消息队列中间件产品太多了,族繁不及备载。

成熟产品经过验证,接口规范,可扩展性强。

结合事业环境因素、组织过程遗产、实施运维考虑、技术路线考虑、开发人员情况等原因综合考虑,基于Redis自己做一个是最可行的选择

1、消息队列需提供哪些功能? 在功能设计上,我崇尚奥卡姆剃刀法则。 对于消息队列,只需要两个方法: 生产 和 消费。 具体的业务场景是任务队列,代码设计如下:

public abstract class TaskQueue{ private final String name ; public String getName(){return this.name;}

public abstract void addTask(Serializable taskId); public abstract Serializable popTask(); }

同时支持多个队列,每个队列都应该有个名字。final确保TaskQueue是线程安全的。TaskQueue的实现类也应该确保线程安全。

addTask向队列中添加一个任务。队列中仅保存任务的id,不存储任务的业务数据。

popTask从队列中取出一个任务来执行。 这种设计不是特别友好,因为她需要调用者自行保证任务执行成功,如果执行失败,自行确保重新把任务放回队列。 无论如何,这种机制是可以工作的。想想奥卡姆剃刀法则,我们先按照这个设计实现出来看看。 如果调用者把业务数据存在数据库中,业务数据中包含“状态“列,标识任务是否被执行,调用者需要自行管理这个状态,并控制事务。

popTask采用阻塞方式,还是非阻塞方式呢? 如果采用阻塞方式,队列中没任务的时候,客户端不会断开连接,只是等。 一般情况下,客户端会有多个worker抢着干活儿,几条狼一起等一个肉包子,画面太美。连接是重要资源,如果一直没活儿干,先放回池里,也不错。 先采用非阻塞的方式吧,如果队列是空的,popTask返回null,立即返回。

2、后续可能提供的功能 2.1、引入Task生命周期概念 应用场景不同,需求也不同。 在严格的应用场景中,需要确保每个Task执行“成功“了。 对于上面提到的popTask后不管的“模式“,这是另外一种“运行模式“,两种模式可以并行存在。

在这种新模式下,Task状态有3种:新创建(new,刚调用addTask加到队列中)、正在执行(in-process,调用popTask后,调用finish前)、完成(done,执行OK了,调用finishTask后)。 调整后的代码如下:

public abstract class TaskQueue{

private final String name ; public String getName(){return this.name;}

public abstract int getMode();

public abstract void addTask(Serializable taskId); public abstract Serializable popTask(); public abstract void finishTask(Serializable taskId); }

2.2、增加批量取出任务的功能 popTask()一次取出一个任务,太磨叽了。 好比我们要买5瓶水,开车去超市买,每去一次买1瓶,有点儿啥。 我们需要一个一次取多个任务的方法。

public abstract class TaskQueue{ ... ... public abstract Serializable[] popTasks(long cnt); }1 2.3、增加阻塞等待机制 想象一种场景: 小明同学,取出一个任务,发现干不了,放回队列,再去取,取出来发现还是干不了,又放回去。反反复复。 小明童鞋肿么了?可能是他干活需要网络,网络断了。可能是他做任务需要写磁盘,磁盘满了。

如果小明像邻居家的孩子一样优秀,当他发现哪里不对的时候,他应该冷静下来,歇会儿。

但他万一不是呢?只有我们能帮他了。

假如队列中有10000个待办任务。 这时候小明来了。他失败100次后,我们应该拦他吗?不应该,除非他主动要求(在系统参数中配置)。5000次后呢?也不应该,除非他主动要求。我们的原则是:我们做的所有事情,对于调用者,都是可以预期的。

我们可以在系统参数中要求调用者设置一个阀值N,如果不设置,默认为100。连续失败N次后,让调用者睡一会儿,睡多长时间,让调用者配置。

假如我们的底层实现中包含待办子队列、重做子队列和完成子队列(这种设计好复杂!pop的时候先pop重做,还是先pop待办,复杂死了!但愿不需要这样)。 待办子队列中有10000个任务。

在小明失败10000次后,所有的任务都在重做子队列了。这时候我们应该拦他吗? 重做子队列要不要设置大小,超过之后,让下一个访问者等。 等的话就会涉及超时,超时后,任务也不能丢弃。 太复杂 了!设置一个连续失败次数的限制就够了!

2.4、考虑增加Task类 不保存任务的相关数据是基本原则,绝对不动摇。 增加Task类可以管理下生命周期,更有用的是,可以把Task本身设计成Listener,代码大概时这样的:

public abstract class Task{

public Serializable getId(); public int getState();

pubic void doTask();

public void whenAdded(final TaskQueue tq); public void whenPoped(final TaskQueue tq); // public void whenFaild(final TaskQueue tq); public void whenFinished(final TaskQueue tq); }

通过Task接口,我们可以对调用过程进行更强势的管理(如进行事务控制),对调用者施加更强的控制,用户也可以获得更多的交互机会,同TaskQueue有更好的交互(如在whenFinished中做持久化工作)。

但这些真的有必要吗?是不是太侵入了?注解的方式会好些吗? 再考虑吧。

2.5、增加系统参数 貌似需要个Config类了,不爽! 本来想做一个很小很精致的小东西的,如果必须再加吧。 如果做的话,需要支持properties、注解设置、api方式设置、Spring注入式设置,烦。

1、Redis是什么鬼? Redis是一个简单的,高效的,分布式的,基于内存的缓存工具。 假设好服务器后,通过网络连接(类似数据库),提供Key-Value式缓存服务。

简单,是Redis突出的特色。 简单可以保证核心功能的稳定和优异。

2、性能 性能方面:Redis是足够高效的。 和Memecached对比,在数据量较小大情况下,Redis性能更优秀。 数据量大到一定程度的时候,Memecached性能稍好。

简单结论:但总体上讲Redis性能已经足够好。

// Ref: Redis性能测试 http://www.cnblogs.com/lulu/archive/2013/06/10/3130878.html 原则:Value大小不要超过1390Byte。

经实验得知: List操作和字符串操作性能相当,略差,几乎可以忽略。 使用Jedis自带pool,“每次从pool中取用完放回“ 和 “重用单个连接“ 相比,平均用时是3倍。这部分需要继续研究底层机制,采用更合理的实验方法进一步获得数据。 使用Jedis自带pool,性能上是满足当前访问量需要的,等有时间了再进一步深入。

3、数据类型 Redis支持5种数据类型:字符串、Map、List、Set、Sorted Set。 List特别适合用于实现队列。提供的操作包括: 从左侧(或右侧)放入一个元素,从右侧(或左侧)取出一个元素,读取某个范围的元素,删除某个范围的元素。

Sorted Set中元素是唯一的,可以通过名字找。 Map可以高效地通过key找。 假如我们需要实现finishTash(taskId),需要通过名字在队列中找元素,上面两个可能会用到。

4、原子操作 实现分布式队列首要问题是:不能出现并发问题。

Redis是底层是单线程的,命令执行是原子操作,支持事务,契合了我们的需求。

Redis直接提供的命令都是原子操作,包括lpush、rpop、blpush、brpop等。

Redis支持事务。通过类似 begin…[cancel]…commit的语法,提供begin…commit之间的命令为原子操作的功能,之间命令对数据的改变对其他操作是不可见的。类似关系型数据库中的存储过程,同时提供了最高级别的事务隔离级别。

Redis支持脚本,每个脚本的执行是原子性的。

做了一下并发测试: 写了个小程序,随机对List做push或pop操作,push的比pop的稍多。 记录每次处理的详细信息到数据库。 最后把List中数据都pop出来,详细记录每次pop详细信息。 统计push和pop是否相等,统计针对每条数据是否都有push和pop。 500并发,没有出现并发问题。

5、集群 实现分布式队列另一个重要问题是:不能出现单点故障。

Redis支持Master-Slave数据复制,从服务器设置 slave-of master-ip:port 即可。 集群功能可以由客户端提供。 客户端使用哨兵,可自动切换主服务器。

由于队列操作都是写操作,从服务器主要目的是备份数据,保证数据安全。

如果想基于 sharding 做多master集群,可以结合 zookeeper 自己做。

Redis 3.0支持集群了,还没细看,应该是个好消息,等大家都用起来,没什么问题的话,可以考虑试试看。

如果 master 宕掉,怎么办? “哨兵”会选出一个新的master来。产生过程中,消息队列暂停服务。 最极端的情况,所有Redis都停了,当消息队列发现Redis停止响应时,对业务系统的请求应抛出异常,停止队列服务。 这样会影响业务,业务系统下订单、审批等操作会失败。如果可以接受,这是一种方案。 Redis整个集群宕掉,这种情况很少发生,如果真发生了,业务系统停止服务也是可以理解的。

如果想要在Redis整个集群宕掉的情况下,消息队列仍继续提供服务。 方法是这样的: 启用备用存储机制,可以是zookeeper、可以是关系型数据库、可以是另外可用的Memecached等。 本地内存存储是不可取的,首先,同步多个客户端虚拟机内存数据太复杂,相当于自己实现了一个Redis,其次,保证内存数据存储安全太复杂。 备用存储机制相当于实现了另外一个版本的消息队列,逻辑一致,底层存储不同。这个实现可以性能低一些,保证最基本的原则即可。 想要保证不出现并发问题,由于消息队列程序同时运行在多个虚拟机中,对象锁、方法锁无效。需要有一个独立于虚拟机的锁机制,zookeeper是个好选择。 将关系型数据库设置为最高级别的事务隔离级别,太傻了。除了zk有其他好办法吗?

Redis集群整个宕掉的同时Zookeeper也全军覆没怎么办? 这个问题是没有尽头的,提供了第二备用存储、第三备用存储、第四备用存储、…,理论上也会同时宕掉,那时候怎么办? 有钱任性的土豪可以继续,预算有限的情况,能做到哪步就做到哪步。

6、持久化 分布式队列的应用场景和缓存的应用场景是不一样的。

如果有没来得及持久化的数据怎么办? 从业务系统的角度,已经成功发送给消息队列了。 消息队列也以为Redis妥妥地收好了。 可Redis还没写到日记里,更没有及时通知小伙伴,挂了。可能是断电了,可能是进程被kill了。

后果会怎样? 已经执行过的任务会再次执行一遍。 已经放到队列中的任务,消失了。 标记为已经完成的任务,状态变为“进行中”了,然后又被执行了一遍。 后果不可接受。

分布式队列不允许丢数据。 从业务角度,哪怕丢1条数据也是无法接受的。 从运维角度,Redis丢数据后,如果可以及时发现并补救,也是可以接受的。

从架构角度,队列保存在Redis中,业务数据(包括任务状态)保存在关系型数据库中。 任务状态是从业务角度确定的,消息队列不应该干涉。如果业务状态没有统一的规范和定义,从业务数据比对任务队列是否全面正确,就只能交给业务开发方来做。 从分工上来看,任务队列的目的是管理任务执行的状态,业务系统把这个职责交给了任务队列,业务系统自身的任务状态维护未必准确。 结论:任务队列不能推卸责任,不能丢数据是核心功能,不能打折扣。

采用 Master-Slave 数据复制模式,配置bgsave,追加存储到aof。

在从服务器上配置bgsave,不影响master性能。

队列操作都是写操作,master任务繁重,能让slave分担的持久化工作,就不要master做。

rdb和aof两种方法都用上,多重保险。 appendfsync设为always。// 单节点测性能,连续100000次算平均时间,和per second比对,性能损失不大。 性能会有些许损失,但任务执行为异步操作,无需用户同步等待,为了保证数据安全,这样是值得的。

当运维需要重启Master服务器的时候,采取这样的顺序: 1. 通过 cli shutdown 停止master服务器, master交代完后事后,关掉自己。这时候“哨兵”会找一个新的master出来。 万万不可以直接kill或者直接打开防火墙中断master和slave之间的连接。 master 对外防火墙,停止对外服务,Master 自动切换到其他服务器上, 原 Master 继续持久化 aof,发送到原来各从服务器。 2. 在原 master 上进行运维操作。 3. 启动原 master,这时候它已经是从服务器了。耐心等待它从新 master 获取最新数据。观察 redis 日志输出,确认数据安全。 4. 对新的 master 重复1-3的操作。 5. 将以上操作写成脚本,自动化执行,避免人为错误。

原文发布于微信公众号 - nginx(nginx-study)

原文发表时间:2016-04-15

本文参与腾讯云自媒体分享计划,欢迎正在阅读的你也加入,一起分享。

发表于

我来说两句

0 条评论
登录 后参与评论

相关文章

来自专栏向治洪

android mtp简介

经作者同意,转发我们公司MTP专家同事huirong的一篇文章。大家也可在程序员第5期看到。 MTP in Android MTP的全称是Media Tra...

48760
来自专栏Albert陈凯

终极 Shell-MacTalk-池建强的随想录Customize to your needs…

终极 Shell Posted on 2013 年 7 月 23 日 http://macshuo.com/?p=676 ? zsh 在开始今天的 Mac...

35280
来自专栏玄魂工作室

CTF实战13 逻辑漏洞

比如我们可以提交异常数据进行测试,包括了提交负数/超大转账金额,提交负数/超大积分,或者积分转换输入输出账户互换和转账输入输出账户互换

16910
来自专栏杨建荣的学习笔记

关于数据库无法登录的问题反思(r6笔记第49天)

昨天接到了同事的一个电话,说有一个数据库无法访问了,希望能够让我来看看,赶紧连过去,发现错误还是一个看似很简单的ora错误。 $ sqlplus / as sy...

28450
来自专栏小怪聊职场

HTTP|GET 和 POST 区别?网上多数答案都是错的!

361100
来自专栏美团技术团队

Android动态日志系统Holmes

背景 美团点评公司是全球领先的一站式生活服务平台,为6亿多消费者和超过450万优质商户提供连接线上线下的电子商务网络。美团点评的业务覆盖了超过200个丰富品类和...

702100
来自专栏程序员互动联盟

【专业领域】jvm虚拟机android移植-接口篇

上一篇我们讲了如何下载phoneme虚拟机开源代码,如何配置编译脚本,编译测试验证基本功能。但是要让她在android上显示运行java小游戏我们还得费点劲。好...

36570
来自专栏杨建荣的学习笔记

通过shell脚本添加备库日志 (r9笔记第94天)

今天下午的时候,准备顺手写一个简单的脚本,但是发现很多事情较真起来真是寸步难行。在写脚本的过程中碰到了太多的问题,很多时候感觉像要实现的功能更通用,就得做更多的...

39060
来自专栏Golang语言社区

[Go语言]一种用于网游服务器的支持多路复用的网络协议处理框架

简介: 本文描述了使用Go语言实现的、适应于Go语言并发模型的一种支持多路复用的网络协议处理框架,并提供了框架的代码实现。作者将这种框架用于网络游戏服务器中的协...

386100
来自专栏逆向与安全

漏洞分析入门一

0x00: 什么是漏洞及漏洞分类 1. 漏洞是指信息系统在生命周期的各个阶段(设计、实现、运维等过程)中产生的某类问题,这些问题会对系统的安全(机密性、完整性、...

24320

扫码关注云+社区

领取腾讯云代金券