网易传媒技术团队：消息中间件实现延迟队列的应用与实践

早期需要延迟处理的业务场景，更多的是通过定时任务扫表，然后执行满足条件的记录，具有频率高、命中低、资源消耗大的缺点。随着消息中间件的普及，延迟消息可以很好的处理这种场景，本文主要介绍延迟消息的使用场景以及基于常见的消息中间件如何实现延迟队列，最后给出了一个在网易公开课使用延迟队列的实践。

1、有效期：限时活动、拼团。。。

2、超时处理：取消超时未支付订单、超时自动确认收货。。。

4、重试：网络异常重试、打车派单、依赖条件未满足重试。。。

5、定时任务：智能设备定时启动。。。

1、RabbitMQ

1）简介：基于AMQP协议，使用Erlang编写，实现了一个Broker框架

a、Broker：接收和分发消息的代理服务器

b、Virtual Host：虚拟主机之间相互隔离，可理解为一个虚拟主机对应一个消息服务

c、Exchange：交换机，消息发送到指定虚拟机的交换机上

d、Binding：交换机与队列绑定，并通过路由策略和routingKey将消息投递到一个或多个队列中

e、Queue：存放消息的队列，FIFO，可持久化

f、Channel：信道，消费者通过信道消费消息，一个TCP连接上可同时创建成百上千个信道，作为消息隔离

2）延迟队列实现：RabbitMQ的延迟队列基于消息的存活时间TTL（Time To Live）和死信交换机DLE（Dead Letter Exchanges）实现

a、TTL：RabbitMQ支持对队列和消息各自设置存活时间，取二者中较小的值，即队列无消费者连接或消息在队列中一直未被消费的过期时间

b、DLE：过期的消息通过绑定的死信交换机，路由到指定的死信队列，消费者实际上消费的是死信队列上的消息

3）缺点：

a、配置麻烦，额外增加一个死信交换机和一个死信队列的配置

b、脆弱性，配置错误或者生产者消费者连接的队列错误都有可能造成延迟失效

2、RocketMQ

1）简介：来源于阿里，目前为Apache顶级开源项目，使用Java编写，基于长轮询的拉取方式，支持事务消息，并解决了顺序消息和海量堆积的问题

a、Broker：存放Topic并根据读取Producer的提交日志，将逻辑上的一个Topic分多个Queue存储，每个Queue上存储消息在提交日志上的位置

b、Name Server：无状态的节点，维护Topic与Broker的对应关系以及Broker的主从关系

2）延迟队列实现：RocketMQ发送延时消息时先把消息按照延迟时间段发送到指定的队列中(rocketmq把每种延迟时间段的消息都存放到同一个队列中)，然后通过一个定时器进行轮训这些队列，查看消息是否到期，如果到期就把这个消息发送到指定topic的队列中

3）缺点：延迟时间粒度受限制（1s/5s/10s/30s/1m/2m/3m/4m/5m/6m/7m/8m/9m/10m/20m/30m/1h/2h）

3、Kafka

1）简介：来源于Linkedin，目前为Apache顶级开源项目，使用Scala和Java编写，基于zookeeper协调的分布式、流处理的日志系统，升级版为Jafka

2）延迟队列实现：Kafka支持延时生产、延时拉取、延时删除等，其基于时间轮和JDK的DelayQueue实现

a、时间轮(TimingWheel)：是一个存储定时任务的环形队列，底层采用数组实现，数组中的每个元素可以存放一个定时任务列表

b、定时任务列表(TimerTaskList)：是一个环形的双向链表，链表中的每一项表示的都是定时任务项

c、定时任务项(TimerTaskEntry)：封装了真正的定时任务TimerTask

d、层级时间轮：当任务的到期时间超过了当前时间轮所表示的时间范围时，就会尝试添加到上层时间轮中，类似于钟表就是一个三级时间轮

e、JDK DelayQueue：存储TimerTaskList，并根据其expiration来推进时间轮的时间，每推进一次除执行相应任务列表外，层级时间轮也会进行相应调整

3）缺点：

a、延迟精度取决于时间格设置

b、延迟任务除由超时触发还可能被外部事件触发而执行

4、ActiveMQ

1）简介：基于JMS协议，Java编写的Apache顶级开源项目，支持点对点和发布订阅两种模式。

a、点对点（point-to-point）：消息发送到指定的队列，每条消息只有一个消费者能够消费，基于拉模型

b、发布订阅（publish/subscribe）：消息发送到主题Topic上，每条消息会被订阅该Topic的所有消费者各自消费，基于推模型

2）延迟队列实现：需要延迟的消息会先存储在JobStore中，通过异步线程任务JobScheduler将到达投递时间的消息投递到相应队列上

a、Broker Filter：Broker中定义了一系列BrokerFilter的子类构成拦截器链，按顺序对消息进行相应处理

b、ScheduleBroker：当消息中指定了延迟相关属性，并且jobId为空时，会生成调度任务存储到JobStore中，此时消息不会进入到队列

c、JobStore：基于BTree存储，key为任务执行的时间戳，value为该时间戳下需要执行的任务列表

d、JobScheduler：取JobStore中最小的key执行（调度时间最早的），执行时间<=当前时间，将该任务列表依次投递到所属的队列，对于需要重复投递和投递失败的会再次存入JobStore中。

注： 此处JobScheduler的执行时间间隔可动态变化，默认05s，有新任务时会立即执行（Object->notifyAll()）并设置时间间隔为01s，没有新任务后，下次执行时间为最近任务的调度执行时间。

3）缺点：投递到队列失败，将消息重新存入JobStore，消息调度执行时间=系统当前时间+延迟时间，会导致消息被真实投递的时间可能为设置的延迟时间的整数倍

5、Redis

1）简介：基于Key-Value的NoSQL数据库，由于其极高的性能常被当作缓存来使用，其数据结构支持：字符串、哈希、列表、集合、有序集合

2）延迟队列实现：Redis的延迟队列基于有序集合，score为执行时间戳，value为任务实体或任务实体引用

3）缺点：

a、实现复杂，本身不支持

b、完全基于内存，延迟时间长浪费内存资源

6、消息队列对比

1、公开课延迟队列技术选型

1）业务场景：关闭超时未支付订单、限时优惠活动、拼团

2）性能要求：订单、活动、拼团 数据量可控，上述MQ均能满足要求

3）可靠性：使用ActiveMQ、RabbitMQ、RocketMQ作为延迟队列更普遍

4）可用性：ActiveMQ、RocketMQ自身支持延迟队列功能，且目前公开课业务中使用的中间件为ActiveMQ和Kafka

5）延迟时间灵活：活动的开始和结束时间比较灵活，而RocketMQ时间粒度较粗，Kafka会依赖时间格有精度缺失

结论： 最终选择ActiveMQ来作为延迟队列

2、业务场景：关闭未支付订单

1）关闭微信未支付订单

2）关闭IOS未支付订单

3、ActiveMQ使用方式

1）activemqxml中支持调度任务

2）发送消息时，设置message的延迟属性

其中：

a、延迟处理

AMQ_SCHEDULED_DELAY：设置多长时间后，投递给消费者（毫秒）

b、重复投递

AMQ_SCHEDULED_PERIOD：重复投递时间间隔（毫秒）

AMQ_SCHEDULED_REPEAT：重复投递次数

c、指定调度计划

AMQ_SCHEDULED_CRON：corn正则表达式

4、公开课使用中进行的优化

1）可靠性：针对实际投递时间可能翻倍的问题，结合ActiveMQ的重复投递，在消费者逻辑中做幂等处理来保证延迟时间的准确性

2）可追溯性：延迟消息及消费情况做数据库冗余存储

3）易用性：业务上定义好延迟枚举类型，直接使用JmsDelayTemplate发送，无需关心数据备份和参数等细节

1、无论是基于死信队列还是基于数据先存储后投递，本质上都是将延迟待发送的消息数据与正常订阅的队列分开存储，从而降低耦合度

2、无论是检查队头消息TTL还是调度存储的延迟数据，本质上都是通过定时任务来完成的，但是定时任务的触发策略以及延迟数据的存储方式决定了不同中间件之间的性能优劣

张浩，2018年加入网易传媒，高级Java开发工程师，目前在网易公开课主要做支付财务体系、版本迭代相关的工作。

左右
信息化管理系统主流语言主要有：Java、C++、C#、PHP、Python、JavaScript、Ruby、Go、Scala、Clojure、Objective-C、Perl等。
Java是一种面向对象的编程语言，它是一种可以运行在多个 *** 作系统上的跨平台语言，可以开发出可移植的、可重用的应用程序。C++是一种中级编程语言，它是一种面向对象的编程语言，可以用来开发高性能、高可靠性的应用程序。C#是微软公司推出的一种面向对象的编程语言，它可以用来开发Windows应用程序和Web应用程序。PHP是一种开源的脚本语言，它可以用来开发动态网站和Web应用程序。Python是一种解释型的编程语言，它可以用来开发各种类型的应用程序，包括网络应用程序、数据库应用程序、图形应用程序等。JavaScript是一种脚本语言，它可以用来开发动态网页，支持AJAX技术。Ruby是一种动态编程语言，它可以用来开发Web应用程序和服务器端应用程序。Go是一种新兴的编程语言，它可以用来开发高性能、可扩展的应用程序。Scala是一种面向对象的编程语言，它可以用来开发大型分布式系统。Clojure是一种函数式编程语言，它可以用来开发可扩展的应用程序。Objective-C是一种面向对象的编程语言，它可以用来开发iOS应用程序和OS X应用程序。Perl是一种脚本语言，它可以用来开发Web应用程序、

minecraft，今天朋友有空我们就联机玩minecraft 可是 我们没办法一起玩 服务器只能进去一个人 进去第二个人 就会崩溃 不知道什么情况 求大神帮助啊 这是服务端崩溃日志 求大神 急求！！！我的版本是164

Apache Spark 本身
1MLlib
AMPLab
Spark最初诞生于伯克利 AMPLab实验室，如今依然还是AMPLab所致力的项目，尽管这些不处于Apache Spark Foundation中，但是依然在你日常的github项目中享有相当的地位。
ML Base
Spark本身的MLLib位于三层ML Base中的最底层，MLI位于中间层，ML Optimizer则处于最为抽象的顶层。
2MLI
3ML Optimizer (又称 Ghostface)
Ghostware这个项目在2014年就开始进行了，不过从未对外公布。在这39个机器学习库中，这是唯一一个雾件，之所以能囊括在这列表中，全凭着AMPLab与ML Base的地位支撑。
ML Base之外
4Splash
这是近期2015年6月的一个项目，在运行随机梯度下降（SGD）时这套随机学习算法声称在性能上比Spark MLib中快了25%-75%。这是AMPLab实验室的sp标记项目，因此值得我们去阅读。
5Keystone ML
KML将端到端的机器学习管道引进到了Spark中，但在近期Spark版本中管道已经趋于成熟。同样也承诺具有一些计算机视觉能力，我曾经在博客中也提到过这也存在一些局限。
6Velox
作为一个服务器专门负责管理大量机器学习模型的收集。
7CoCoA
通过优化通信模式与shuffles来实现更快的机器学习，详情可见这篇论文的描述《高效通信分布式双坐标上升》。
框架
GPU-based
8DeepLearning4j
我曾经的一则博客有进行说明 《DeepLearning4J 增加了Spark gpu的支持》。
9Elephas
全新的概念，这也是我写这篇博客的初衷。它提供了一个接口给Keras。
Non-GPU-based
10DistML
模式并行下而并非数据并行的参数服务器（正如 Spark MLib）。
11Aerosolve
来自Airbnb，用于他们自动化定价。
12 Zen
逻辑斯谛回归、隐含狄利克雷分布（LDA）、因子分解机、神经网络、受限玻尔兹曼机。
13Distributed Data Frame
与Spark DataFrame类似，但是引擎是不可知的（例如在未来它将运行在引擎上而不是Spark）。其中包括了交叉验证和外部机器学习库的接口。
其他机器学习系统的接口
14 spark-corenlp
封装了斯坦福CoreNLP。
15 Sparkit-learn
给Python Scikit-learn的接口。
16 Sparkling Water
给 的接口。
17 hivemall-spark
封装了Hivemall,，在Hive中的机器学习。
18 spark-pmml-exporter-validator
可导出预测模型标记语言（PMML），一种用于传递机器学习模型的行业标准的XML格式。
附加组件：增强MLlib中现有的算法。
19 MLlib-dropout
为Spark MLLib 增加dropout能力。基于以下这篇论文进行的实现，《Dropout：一个简单的方法来防止神经网络中的过拟合》。
20generalized-kmeans-clustering
为K-Means算法增加任意距离函数。
21 spark-ml-streaming
可视化的流式机器学习算法内置于Spark MLlib。
算法
监督学习
22 spark-libFM
因子分解机。
23 ScalaNetwork
递归神经网络（RNNs）。
24 dissolve-struct
基于上文中提到的高性能Spark通信框架CoCoA下的支持向量机（SVM）。
25 Sparkling Ferns
基于以下这篇论文进行的实现，《通过使用随机森林与随机蕨算法的图像分类技术》。
26 streaming-matrix-factorization
矩阵分解推荐系统。

大数据主要学习的东西有6个方面：
第一阶段
JavaSE基础核心
第二阶段
数据库关键技术
第三阶段
大数据基础核心
第四阶段
Spark生态体系框架&大数据高薪精选项目
第五阶段
Spark生态体系框架&企业无缝对接项目
第六阶段
Flink流式数据处理框架

---