一种消息分发方法和装置的制造方法
一种消息分发方法和装置的制造方法
【技术领域】
[0001] 本发明涉及计算机通信领域技术，特别涉及一种消息分发方法和装置。
【背景技术】
[0002] JMS，即JavaMessageService，Java消息服务，是Java标准化组织（JCP)开 发的一种标准（代号为JSR914)，它是一个Java平台中关于面向消息中间件（MOM)的 API(ApplicationProgrammingInterface,应用程序编程接口）。JMS用于在两个应用程 序之间，或分布式系统中发送消息，进行异步通信。
[0003] JMS提供了创建、发送、接收、读取消息的服务，JMS规范下有两种基本消息模型， 即PTP(点对点）和Pub/Sub(发布/订阅）。其中，Pub/Sub模型可以定义多个消费者消费 消息，它在企业级开发中广泛采用，方便了系统间消息的传递，省去了大量开发时间以及节 省了存储消息的资源空间。
[0004] JMS的Pub/Sub模型中，消息的订阅分发通常采用基于JMSTOPIC规范来实现。JMS TOPIC规范有两种实现方式，第一种是不提供存储而直接转发消息，第二种是为每个订阅者 提供缓存队列。
[0005] 现有的基于JMS TOPIC规范的消息分发步骤如下：
[0006] 1)消息发送方创建一条消息，并将该条消息发送给TopicManager(主题管理 器）；
[0007] 2)Topic Manager查询该条消息的所有订阅者；
[0008] 3)将该条消息转发给其所有订阅者；
[0009] 4)在预设时间周期后重复步骤1)~3)。
[0010] 从上述现有的消息分发步骤来看，消息的订阅者不能指定开始订阅点，JMSTOPIC 规范只支持订阅者被动接收实时消息，不支持订阅者对历史消息的订阅。同时，对于订阅者 来说，消息的获得并不支持断点续传，当订阅者连接超时或者中断，并重新连接后，不能进 行断点续传，从上次成功订阅消息的时间点继续订阅消息。另外，JMSTOPIC规范也不支持 消息的核对，对于消息一致性要求很高的业务场景（例如财务数据，订单数据等）需要周期 性核对消息，而现有的JMSTOPIC规范不支持历史消息的查阅，因此无法核对历史消息的一 致性，当出现消息不一致的情况时也难以核查。
[0011] 另外，虽然上述JMSTOPIC规范的两种实现方式中的后一种为每个订阅者提供了 缓存，但受限于缓存队列的深度，无法保证消息不丢失，同时因为针对每个订阅者分别提供 缓存，进而同时存储了多份消息拷贝，进而造成了对存储资源的浪费。
【发明内容】
[0012] 有鉴于此，本发明提供一种消息分发方法和装置，以实现订阅消息的持久化和消 息订阅的记录功能，进而为消息订阅者提供可订阅消息范围的查询、订阅、以及断点续传等 服务。
[0013] 本发明的技术方案是这样实现的：
[0014] -种消息分发方法，包括：
[0015] 接收消息生产者所创建的消息，并将所接收的消息写入消息数据库；
[0016] 获取消息订阅者的消息订阅范围，根据所述消息订阅范围确定所述消息数据库中 的起始消息订阅点；
[0017] 从所述起始消息订阅点开始，将所述消息订阅者的消息订阅范围内的各条消息分 发给所述消息订阅者。
[0018] 进一步，将所接收的消息写入消息数据库时，在所述消息数据库中同时记录写入 消息时的时间值和写入的消息的顺序值。
[0019] 进一步，根据所述消息订阅范围确定所述消息数据库中的起始消息订阅点包括：
[0020] 根据所述消息订阅范围，在所述消息数据库中确定所述消息订阅范围内的最早写 入的消息，并将其作为所述起始消息订阅点。
[0021] 进一步，从所述起始消息订阅点开始，将所述消息订阅者的消息订阅范围内的各 条消息分发给所述消息订阅者，包括：
[0022] 依据所述时间值和顺序值，将所述消息订阅者的消息订阅范围内的各条消息逐条 分发给所述消息订阅者。
[0023] 进一步，在将所述消息订阅者的消息订阅范围内的各条消息逐条分发给所述消息 订阅者的同时，将所分发的各条消息的时间值和顺序值同时发送给所述消息订阅者；
[0024] 当连接中断或者所述消息订阅者所接收的消息丢失时，将最近一次成功分发的消 息作为新的起始消息订阅点，并从所述新的起始消息订阅点开始，将所述消息订阅者的消 息订阅范围内尚未分发的各条消息分发给所述消息订阅者。
[0025] 进一步，若消息订阅者的消息订阅范围的最晚时间晚于当前时间，且当所述消息 数据库中的所有消息订阅范围内的消息全部分发给所述消息订阅者时，等待消息数据库中 写入新消息，并将写入的新消息分发给消息订阅者。
[0026] 进一步，当所述消息数据库的存储容量不足时，依据所述时间值和顺序值，逐条删 除最早写入所述消息数据库的消息。
[0027] 进一步，所述方法还包括：
[0028] 设置清理周期，在每个清理周期中，依据所述时间值和顺序值，逐条删除最早写入 所述消息数据库的消息。
[0029] 进一步，所述方法还包括：
[0030] 设置所述消息数据库的数据量阈值，所述数据量阈值小于所述消息数据库的存储 容量；
[0031] 在删除最早写入所述消息数据库的消息时，当所述消息数据库的数据量降到所述 数据量阈值时，停止继续对所述消息数据库中消息的删除。
[0032] 一种消息分发装置，包括：
[0033]消息接收模块，用于接收消息生产者所创建的消息，并将所接收的消息写入消息 数据库；
[0034] 消息订阅模块，用于获取消息订阅者的消息订阅范围，根据所述消息订阅范围确 定所述消息数据库中的起始消息订阅点；
[0035] 消息分发模块，用于从所述起始消息订阅点开始，将所述消息订阅者的消息订阅 范围内的各条消息分发给所述消息订阅者；以及，
[0036] 消息数据库，用于存储所述消息。
[0037]进一步，所述消息接收模块，还用于将所接收的消息写入消息数据库时，在所述消 息数据库中同时记录写入消息时的时间值和写入的消息的顺序值。
[0038]进一步，所述消息订阅模块根据所述消息订阅范围，在所述消息数据库中确定所 述消息订阅范围内的最早写入的消息，并将其作为所述起始消息订阅点。
[0039]进一步，所述消息分发模块依据所述时间值和顺序值，将所述消息订阅者的消息 订阅范围内的各条消息逐条分发给所述消息订阅者。
[0040]进一步，在将所述消息订阅者的消息订阅范围内的各条消息逐条分发给所述消息 订阅者的同时，所述消息分发模块将所分发的各条消息的时间值和顺序值同时发送给所述 消息订阅者；
[0041] 当连接中断或者所述消息订阅者所接收的消息丢失时：
[0042]所述消息订阅模块将最近一次成功分发的消息作为新的起始消息订阅点； [0043]所述消息分发模块从所述新的起始消息订阅点开始，将所述消息订阅者的消息订 阅范围内尚未分发的各条消息分发给所述消息订阅者。
[0044]进一步，所述消息分发模块在消息订阅者的消息订阅范围的最晚时间晚于当前时 间，且当所述消息数据库中的所有消息订阅范围内的消息全部分发给所述消息订阅者时， 等待消息数据库中写入新消息，并将写入的新消息分发给消息订阅者。
[0045] 进一步，所述装置还包括：
[0046]数据清理模块，用于当所述消息数据库的存储容量不足时，依据所述时间值和顺 序值，逐条删除最早写入所述消息数据库的消息。
[0047] 进一步，所述装置还包括：
[0048]数据清理模块，用于设置清理周期，并在每个清理周期中，依据所述时间值和顺序 值，逐条删除最早写入所述消息数据库的消息。
[0049]进一步，所述数据清理模块在删除最早写入所述消息数据库的消息时，当所述消 息数据库的数据量降到所述数据量阈值时，停止继续对所述消息数据库中消息的删除；
[0050] 其中，所述数据量阈值小于所述消息数据库的存储容量。
[0051] 从上述方案可以看出，本发明的消息分发方法和装置中，利用消息数据库对消息 生产者所创建的消息进行保存，并根据消息订阅者的消息订阅范围向所述消息订阅者提供 所保存的消息，另外，在消息生产者所创建的消息进行保存的同时，在所述消息数据库中还 记录写入消息时的时间值和写入的消息的顺序值，进而实现了依据时间值和顺序值确定消 息订阅者所获得的消息以及尚未获得的消息，当连接丢失或者消息订阅者所接收的消息丢 失时，依据时间值和顺序值可实现消息订阅者对订阅消息的续传和补传，同时，通过重新设 定消息订阅范围还可实现消息订阅者对已订阅消息的一致性检查，例如消息订阅者通过设 定消息订阅范围以取出任意一段时间范围内的消息进而与其已经接收到的消息进行一致 性检查。本发明的消息分发方法和装置保证了消息订阅时，当订阅消息已经分发到订阅者 后，仍然对所订阅消息的保存，同时针对多个消息订阅者仅使用一份消息数据库的拷贝，多 个消息订阅者仅通过其所接收的消息的时间值和顺序值确定各自已经获得的所订阅的消 息和尚未获得的消息，而不必为每个消息订阅者分别使用不同的消息数据的拷贝，进而节 约了消息的存储资源。
【附图说明】
[0052] 图1为本发明的消息分发方法的实施例流程图；
[0053] 图2为本发明的消息分发装置的实施例结构图。
【具体实施方式】
[0054] 为了使本发明的目的、技术方案及优点更加清楚明白，以下参照附图并举实施例， 对本发明作进一步详细说明。
[0055] 如图1所示，本发明的消息分发方法实施例，主要包括以下步骤：
[0056] 步骤1、接收消息生产者所创建的消息，并将所接收的消息写入消息数据库；
[0057] 步骤2、获取消息订阅者的消息订阅范围，根据所述消息订阅范围确定所述消息数 据库中的起始消息订阅点；
[0058] 步骤3、从所述起始消息订阅点开始，将所述消息订阅者的消息订阅范围内的各条 消息分发给所述消息订阅者。
[0059] 为了实现消息订阅者对所订阅消息的断点续传、消息核对等功能，在步骤1中还 进一步包括：
[0060] 将所接收的消息写入消息数据库时，在所述消息数据库中同时记录写入消息时的 时间值和写入的消息的顺序值。
[0061] 其中，时间值包括写入消息的日期及写入消息的时刻，例如写入的某条消息的时 间值为2015-5-25 13:00:36,表明该条消息的写入时间（即在消息数据库中该条消息记录 的创建时间）为2015年5月25日下午13点00分36秒。
[0062] 顺序值为在单位时间内写入的信息的数量从小到大进行排列的顺序值。例如单位 时间按秒计算，则若在2015年5月25日下午13点00分36秒写入数据库300条消息，则从 2015年5月25日下午13点00分36秒的1秒钟时间段内，按照先后顺序将这300条消息 从[000001]到[000300]分别赋予顺序值，之后若在2015年5月25日下午13点00分37 秒写入数据库500条消息，则从2015年5月25日下午13点00分37秒的1秒钟时间段内， 按照先后顺序将这500条消息从[000001]到[000500]分别赋予顺序值，以此类推。通过 消息数据库中的每条消息的时间值和顺序值，即可得知该条消息是消息数据库中在哪一时 间段的第几条消息，进而，为随后的消息订阅范围提供依据并可进一步实现消息断点续传、 消息核对等效果，参见后续说明。
[0063] 步骤2中，根据所述消息订阅范围确定所述消息数据库中的起始消息订阅点包 括：
[0064] 根据所述消息订阅范围，在所述消息数据库中确定所述消息订阅范围内的最早写 入的消息，并将其作为所述起始消息订阅点。
[0065] 例如，若消息订阅者的消息订阅范围设定为2015年5月25日下午13点00分36 秒以后的所有消息，则可通过时间值，在所述消息数据库中确定所述消息订阅范围内的最 早写入的消息。结合表1对此进行说明。
[0066]表1:消息数据库中的消息数据
[0068] 表1表示消息数据库中存储的各条消息数据记录，其中各条消息的顺序值在每 个时间周期（秒）内依据写入消息数据库的顺序而递加，其中例如时间值为2015-05-25 13:00:36顺序值为[000125]的消息，表明该消息是2015年05月25日13点00分36秒的 这一秒时间内写入的第125条消息，其余各条消息数据与之相似，不再赘述。若消息订阅者 的消息订阅范围设定为2015年5月25日13点00分36秒以后的所有消息，则通过时间值， 在所述消息数据库中确定所述消息订阅范围内的最早写入的消息的时间值为2015-05-25 13:00:36、顺序值为[000001]的消息。
[0069] 结合所述消息数据库中同时记录写入消息时的时间值和写入的消息的顺序值，本 发明步骤3的从所述起始消息订阅点开始，将所述消息订阅者的消息订阅范围内的各条消 息分发给所述消息订阅者，包括：
[0070] 依据所述时间值和顺序值，将所述消息订阅者的消息订阅范围内的各条消息逐条 分发给所述消息订阅者。
[0071] 继续结合表1进行说明。本发明步骤3的实施例中，依据所述时间值和顺序值， 将2015年5月25日13点00分36秒以后的所有消息的订阅范围内的各条消息，从时间 值为2015-05-25 13:00:36、顺序值为[000001]的消息开始逐条分发给所述消息订阅者， 即先分发时间值为2015-05-25 13:00:36、顺序值为[000001]的消息，然后分发时间值为 2015-05-25 13:00:36、顺序值为[000002]的消息，待时间值为 2015-05-25 13:00:36 所有 消息分发完毕后，再分发时间值为2015-05-25 13:00:37的各条消息，消息的分发顺序按 照顺序值从小到大进行，之后再分发 时间值为2015-05-25 13:00:38的各条消息，以此类 推。
[0072] 本发明步骤3的实施例中，在将所述消息订阅者的消息订阅范围内的各条消息逐 条分发给所述消息订阅者的同时，将所分发的各条消息的时间值和顺序值同时发送给所述 消息订阅者；进而，当连接中断或者所述消息订阅者所接收的消息丢失时，可确定最近一次 成功分发的消息的时间值和顺序值，进而将最近一次成功分发的消息作为新的起始消息订 阅点，并从所述新的起始消息订阅点开始，将所述消息订阅者的消息订阅范围内尚未分发 的各条消息分发给所述消息订阅者。例如，在将所述消息订阅者的消息订阅范围内的各条 消息逐条分发给所述消息订阅者时，当成功分发了时间值为2015-05-25 13:00:36、顺序值 为[000078]的消息后连接中断，则将时间值为2015-05-25 13:00:36、顺序值为[000078] 的消息作为新的起始消息订阅点，在连接回复后，从时间值为2015-05-25 13:00:36、顺序 值为[000078]的消息开始，将所述消息订阅者的消息订阅范围内尚未分发的各条消息分 发给所述消息订阅者。
[0073] 本发明的消息分发方法中，若消息订阅者的消息订阅范围的最晚时间晚于当前时 间，且当所述消息数据库中的所有消息订阅范围内的消息全部分发给所述消息订阅者时， 等待消息数据库中写入新消息，并将写入的新消息分发给消息订阅者。例如，若消息订阅者 的消息订阅范围设定为2015年5月25日13点00分36秒以后的所有消息，则消息订阅者 的消息订阅范围的最晚时间是晚于当前时间的，在将当前消息数据库中所述消息订阅者的 消息订阅范围内的所有消息全部分发给消息订阅者后，因为消息订阅者的消息订阅范围的 最晚时间晚于当前时间，则继续等待消息数据库中写入的新消息，当消息数据库中写入新 的消息后，将写入的新消息分发给消息订阅者，并继续等待消息数据库中写入的新消息，以 此循环。
[0074] 本发明的消息分发方法实施例中，当所述消息数据库的存储容量不足时，依据所 述时间值和顺序值，逐条删除最早写入所述消息数据库的消息。进而，保证新的消息能够及 时保存，避免丢失，同时过早的消息能够进行清理以避免对存储资源的不必要占用。另外， 本发明的消息分发方法实施例中还包括，设置清理周期，在每个清理周期中，依据所述时间 值和顺序值，逐条删除最早写入所述消息数据库的消息。进而同样能够保证新的消息能够 及时保存，避免丢失，以及过早的消息能够进行清理以避免对存储资源的不必要占用。
[0075] 在删除旧有数据的同时，需要保存较近一段时间所保存的信息，以保证消息订阅 者的消息获取，本发明的消息分发方法实施例中还包括，设置所述消息数据库的数据量阈 值，所述数据量阈值小于所述消息数据库的存储容量；在删除最早写入所述消息数据库的 消息时，当所述消息数据库的数据量降到所述数据量阈值时，停止继续对所述消息数据库 中消息的删除。其中，数据量阈值例如保存的信息条数或者保存的时间范围等。通过设定 恰当的数据量阈值，能够保证消息订阅者的对所需要的有用消息的获取，以及对过早的已 经无用的消息的清理，实现消息数据库中存储空间的循环使用。
[0076] 本发明实施例同时还提供了一种消息分发装置，如图2所示，其包括消息数据库 11、消息接收模块12、消息订阅模块13和消息分发模块14。其中，所述消息数据库11用于 存储消息。所述消息接收模块12,用于接收消息生产者2所创建的消息，并将所接收的消 息写入消息数据库11。进一步地，所述消息接收模块12,还用于将所接收的消息写入消息 数据库11时，在所述消息数据库11中同时记录写入消息时的时间值和写入的消息的顺序 值。消息订阅模块13,用于获取消息订阅者3的消息订阅范围，根据所述消息订阅范围确定 所述消息数据库11中的起始消息订阅点。具体地，所述消息订阅模块13根据所述消息订 阅范围，在所述消息数据库11中确定所述消息订阅范围内的最早写入的消息，并将其作为 所述起始消息订阅点。消息分发模块14,用于从所述起始消息订阅点开始，将所述消息订阅 者3的消息订阅范围内的各条消息分发给所述消息订阅者3。具体地，所述消息分发模块 14依据所述时间值和顺序值，将所述消息订阅者3的消息订阅范围内的各条消息逐条分发 给所述消息订阅者。
[0077]另外，本发明的消息分发装置实施例中，在将所述消息订阅者3的消息订阅范围 内的各条消息逐条分发给所述消息订阅者3的同时，所述消息分发模块14将所分发的各条 消息的时间值和顺序值同时发送给所述消息订阅者。进而，当连接中断或者所述消息订阅 者所接收的消息丢失时，所述消息订阅模块13将最近一次成功分发给消息订阅者3的消息 作为新的起始消息订阅点；所述消息分发模块14从所述新的起始消息订阅点开始，将所述 消息订阅者3的消息订阅范围内尚未分发的各条消息分发给所述消息订阅者3。进而实现 了断点续传功能。
[0078] 本发明的消息分发装置实施例中，所述消息分发模块14在消息订阅者3的消息订 阅范围的最晚时间晚于当前时间，且当所述消息数据库11中的所有消息订阅范围内的消 息全部分发给所述消息订阅者时，等待消息数据库11中写入新消息，并将写入的新消息分 发给消息订阅者3。进而实现了对当前数据的及时分发。
[0079] 本发明的消息分发装置还进一步包括数据清理模块15,用于当所述消息数据库 11的存储容量不足时，依据所述时间值和顺序值，逐条删除最早写入所述消息数据库11的 消息。例如，当所述消息数据库11的存储容量过满而无法继续存储新的消息时，所述数据 清理模块15依据所述时间值和顺序值，逐条删除最早写入所述消息数据库11的消息。所述 数据清理模块15,还用于设置清理周期，并在每个清理周期中，依据所述时间值和顺序值， 逐条删除最早写入所述消息数据库11的消息。所述数据清理模块15在删除最早写入所述 消息数据库11的消息时，当所述消息数据库11的数据量降到所述数据量阈值时，停止继续 对所述消息数据库11中消息的删除。其中，所述数据量阈值小于所述消息数据库11的存 储容量。
[0080] 本发明中，将所分发的各条消息的时间值和顺序值同时发送给所述消息订阅者， 进而间接地实现了消息订阅者通过消息的时间值和顺序值对自己所订阅消息的管理。通过 消息的时间值和订阅值，消息订阅者可获知自己所订阅的消息，进而不必由消息分发装置 单独记录多个消息订阅者各自所订阅的消息范围和已经分发给各个消息订阅者的消息。由 此可见，本发明的消息分发方法和装置仅采用一个消息数据库也同样实现了对多个消息订 阅者消息订阅的支持。
[0081 ] 本发明的消息分发方法和装置，利用消息数据库对消息生产者所创建的消息进行 保存，并根据消息订阅者的消息订阅范围向所述消息订阅者提供所保存的消息，另外，在消 息生产者所创建的消息进行保存的同时，在所述消息数据库中还记录写入消息时的时间值 和写入的消息的顺序值，进而实现了依据时间值和顺序值确定消息订阅者所获得的消息以 及尚未获得的消息，当连接丢失或者消息订阅者所接收的消息丢失时，依据时间值和顺序 值可实现消息订阅者对订阅消息的续传和补传 ，同时，通过重新设定消息订阅范围还可实 现消息订阅者对已订阅消息的一致性检查，例如消息订阅者通过设定消息订阅范围以取出 任意一段时间范围内的消息进而与其已经接收到的消息进行一致性检查。本发明的消息分 发方法和装置保证了消息订阅时，当订阅消息已经分发到订阅者后，仍然对所订阅消息的 保存，同时针对多个消息订阅者仅使用1份消息数据库的拷贝，多个消息订阅者仅通过其 所接收的消息的时间值和顺序值确定各自已经获得的所订阅的消息和尚未获得的消息，而 不必为每个消息订阅者分别使用不同的消息数据的拷贝，进而节约了消息的存储资源。 [0082] 以上所述仅为本发明的较佳实施例而已，并不用以限制本发明，凡在本发明的精 神和原则之内，所做的任何修改、等同替换、改进等，均应包含在本发明保护的范围之内。
【主权项】
1. 一种消息分发方法，包括： 接收消息生产者所创建的消息，并将所接收的消息写入消息数据库； 获取消息订阅者的消息订阅范围，根据所述消息订阅范围确定所述消息数据库中的起 始消息订阅点； 从所述起始消息订阅点开始，将所述消息订阅者的消息订阅范围内的各条消息分发给 所述消息订阅者。2. 根据权利要求1所述的消息分发方法，其特征在于：将所接收的消息写入消息数据 库时，在所述消息数据库中同时记录写入消息时的时间值和写入的消息的顺序值。3. 根据权利要求2所述的消息分发方法，其特征在于，根据所述消息订阅范围确定所 述消息数据库中的起始消息订阅点包括： 根据所述消息订阅范围，在所述消息数据库中确定所述消息订阅范围内的最早写入的 消息，并将其作为所述起始消息订阅点。4. 根据权利要求2所述的消息分发方法，其特征在于，从所述起始消息订阅点开始，将 所述消息订阅者的消息订阅范围内的各条消息分发给所述消息订阅者，包括： 依据所述时间值和顺序值，将所述消息订阅者的消息订阅范围内的各条消息逐条分发 给所述消息订阅者。5. 根据权利要求4所述的消息分发方法，其特征在于： 在将所述消息订阅者的消息订阅范围内的各条消息逐条分发给所述消息订阅者的同 时，将所分发的各条消息的时间值和顺序值同时发送给所述消息订阅者； 当连接中断或者所述消息订阅者所接收的消息丢失时，将最近一次成功分发的消息作 为新的起始消息订阅点，并从所述新的起始消息订阅点开始，将所述消息订阅者的消息订 阅范围内尚未分发的各条消息分发给所述消息订阅者。6. 根据权利要求2所述的消息分发方法，其特征在于： 若消息订阅者的消息订阅范围的最晚时间晚于当前时间，且当所述消息数据库中的所 有消息订阅范围内的消息全部分发给所述消息订阅者时，等待消息数据库中写入新消息， 并将写入的新消息分发给消息订阅者。7. 根据权利要求2所述的消息分发方法，其特征在于：当所述消息数据库的存储容量 不足时，依据所述时间值和顺序值，逐条删除最早写入所述消息数据库的消息。8. 根据权利要求2所述的消息分发方法，其特征在于，所述方法还包括： 设置清理周期，在每个清理周期中，依据所述时间值和顺序值，逐条删除最早写入所述 消息数据库的消息。9. 根据权利要求7或8所述的消息分发方法，其特征在于，所述方法还包括： 设置所述消息数据库的数据量阈值，所述数据量阈值小于所述消息数据库的存储容 量； 在删除最早写入所述消息数据库的消息时，当所述消息数据库的数据量降到所述数据 量阈值时，停止继续对所述消息数据库中消息的删除。10. -种消息分发装置，其特征在于，包括： 消息接收模块，用于接收消息生产者所创建的消息，并将所接收的消息写入消息数据 库； 消息订阅模块，用于获取消息订阅者的消息订阅范围，根据所述消息订阅范围确定所 述消息数据库中的起始消息订阅点； 消息分发模块，用于从所述起始消息订阅点开始，将所述消息订阅者的消息订阅范围 内的各条消息分发给所述消息订阅者；以及， 消息数据库，用于存储所述消息。11. 根据权利要求10所述的消息分发装置，其特征在于： 所述消息接收模块，还用于将所接收的消息写入消息数据库时，在所述消息数据库中 同时记录写入消息时的时间值和写入的消息的顺序值。12. 根据权利要求11所述的消息分发装置，其特征在于： 所述消息订阅模块根据所述消息订阅范围，在所述消息数据库中确定所述消息订阅范 围内的最早写入的消息，并将其作为所述起始消息订阅点。13. 根据权利要求11所述的消息分发装置，其特征在于： 所述消息分发模块依据所述时间值和顺序值，将所述消息订阅者的消息订阅范围内的 各条消息逐条分发给所述消息订阅者。14. 根据权利要求13所述的消息分发装置，其特征在于： 在将所述消息订阅者的消息订阅范围内的各条消息逐条分发给所述消息订阅者的同 时，所述消息分发模块将所分发的各条消息的时间值和顺序值同时发送给所述消息订阅 者； 当连接中断或者所述消息订阅者所接收的消息丢失时： 所述消息订阅模块将最近一次成功分发的消息作为新的起始消息订阅点； 所述消息分发模块从所述新的起始消息订阅点开始，将所述消息订阅者的消息订阅范 围内尚未分发的各条消息分发给所述消息订阅者。15. 根据权利要求11所述的消息分发装置，其特征在于： 所述消息分发模块在消息订阅者的消息订阅范围的最晚时间晚于当前时间，且当所述 消息数据库中的所有消息订阅范围内的消息全部分发给所述消息订阅者时，等待消息数据 库中写入新消息，并将写入的新消息分发给消息订阅者。16. 根据权利要求11所述的消息分发装置，其特征在于，所述装置还包括： 数据清理模块，用于当所述消息数据库的存储容量不足时，依据所述时间值和顺序值， 逐条删除最早写入所述消息数据库的消息。17. 根据权利要求11所述的消息分发装置，其特征在于，所述装置还包括： 数据清理模块，用于设置清理周期，并在每个清理周期中，依据所述时间值和顺序值， 逐条删除最早写入所述消息数据库的消息。18. 根据权利要求16或17所述的消息分发装置，其特征在于： 所述数据清理模块在删除最早写入所述消息数据库的消息时，当所述消息数据库的数 据量降到所述数据量阈值时，停止继续对所述消息数据库中消息的删除； 其中，所述数据量阈值小于所述消息数据库的存储容量。
【专利摘要】本发明公开了一种消息分发方法和装置，其包括：接收消息生产者所创建的消息，并将所接收的消息写入消息数据库；获取消息订阅者的消息订阅范围，根据所述消息订阅范围确定所述消息数据库中的起始消息订阅点；从所述起始消息订阅点开始，将所述消息订阅者的消息订阅范围内的各条消息分发给所述消息订阅者。本发明实现了依据时间值和顺序值确定消息订阅者所获得的消息以及尚未获得的消息，实现了消息订阅者对订阅消息的续传和补传，以及对订阅消息的一致性检查，本发明同时针对多个消息订阅者仅使用一份消息数据库的拷贝，而不必为每个消息订阅者分别使用不同的消息数据的拷贝，进而节约了消息的存储资源。
【IPC分类】G06F17/30
【公开号】CN104899277
【申请号】CN201510287977
【发明人】许文靖
【申请人】北京京东尚科信息技术有限公司, 北京京东世纪贸易有限公司
【公开日】2015年9月9日
【申请日】2015年5月29日
【技术领域】
[0001] 本发明涉及计算机通信领域技术，特别涉及一种消息分发方法和装置。
【背景技术】
[0002] JMS，即JavaMessageService，Java消息服务，是Java标准化组织（JCP)开 发的一种标准（代号为JSR914)，它是一个Java平台中关于面向消息中间件（MOM)的 API(ApplicationProgrammingInterface,应用程序编程接口）。JMS用于在两个应用程 序之间，或分布式系统中发送消息，进行异步通信。
[0003] JMS提供了创建、发送、接收、读取消息的服务，JMS规范下有两种基本消息模型， 即PTP(点对点）和Pub/Sub(发布/订阅）。其中，Pub/Sub模型可以定义多个消费者消费 消息，它在企业级开发中广泛采用，方便了系统间消息的传递，省去了大量开发时间以及节 省了存储消息的资源空间。
[0004] JMS的Pub/Sub模型中，消息的订阅分发通常采用基于JMSTOPIC规范来实现。JMS TOPIC规范有两种实现方式，第一种是不提供存储而直接转发消息，第二种是为每个订阅者 提供缓存队列。
[0005] 现有的基于JMS TOPIC规范的消息分发步骤如下：
[0006] 1)消息发送方创建一条消息，并将该条消息发送给TopicManager(主题管理 器）；
[0007] 2)Topic Manager查询该条消息的所有订阅者；
[0008] 3)将该条消息转发给其所有订阅者；
[0009] 4)在预设时间周期后重复步骤1)~3)。
[0010] 从上述现有的消息分发步骤来看，消息的订阅者不能指定开始订阅点，JMSTOPIC 规范只支持订阅者被动接收实时消息，不支持订阅者对历史消息的订阅。同时，对于订阅者 来说，消息的获得并不支持断点续传，当订阅者连接超时或者中断，并重新连接后，不能进 行断点续传，从上次成功订阅消息的时间点继续订阅消息。另外，JMSTOPIC规范也不支持 消息的核对，对于消息一致性要求很高的业务场景（例如财务数据，订单数据等）需要周期 性核对消息，而现有的JMSTOPIC规范不支持历史消息的查阅，因此无法核对历史消息的一 致性，当出现消息不一致的情况时也难以核查。
[0011] 另外，虽然上述JMSTOPIC规范的两种实现方式中的后一种为每个订阅者提供了 缓存，但受限于缓存队列的深度，无法保证消息不丢失，同时因为针对每个订阅者分别提供 缓存，进而同时存储了多份消息拷贝，进而造成了对存储资源的浪费。
【发明内容】
[0012] 有鉴于此，本发明提供一种消息分发方法和装置，以实现订阅消息的持久化和消 息订阅的记录功能，进而为消息订阅者提供可订阅消息范围的查询、订阅、以及断点续传等 服务。
[0013] 本发明的技术方案是这样实现的：
[0014] -种消息分发方法，包括：
[0015] 接收消息生产者所创建的消息，并将所接收的消息写入消息数据库；
[0016] 获取消息订阅者的消息订阅范围，根据所述消息订阅范围确定所述消息数据库中 的起始消息订阅点；
[0017] 从所述起始消息订阅点开始，将所述消息订阅者的消息订阅范围内的各条消息分 发给所述消息订阅者。
[0018] 进一步，将所接收的消息写入消息数据库时，在所述消息数据库中同时记录写入 消息时的时间值和写入的消息的顺序值。
[0019] 进一步，根据所述消息订阅范围确定所述消息数据库中的起始消息订阅点包括：
[0020] 根据所述消息订阅范围，在所述消息数据库中确定所述消息订阅范围内的最早写 入的消息，并将其作为所述起始消息订阅点。
[0021] 进一步，从所述起始消息订阅点开始，将所述消息订阅者的消息订阅范围内的各 条消息分发给所述消息订阅者，包括：
[0022] 依据所述时间值和顺序值，将所述消息订阅者的消息订阅范围内的各条消息逐条 分发给所述消息订阅者。
[0023] 进一步，在将所述消息订阅者的消息订阅范围内的各条消息逐条分发给所述消息 订阅者的同时，将所分发的各条消息的时间值和顺序值同时发送给所述消息订阅者；
[0024] 当连接中断或者所述消息订阅者所接收的消息丢失时，将最近一次成功分发的消 息作为新的起始消息订阅点，并从所述新的起始消息订阅点开始，将所述消息订阅者的消 息订阅范围内尚未分发的各条消息分发给所述消息订阅者。
[0025] 进一步，若消息订阅者的消息订阅范围的最晚时间晚于当前时间，且当所述消息 数据库中的所有消息订阅范围内的消息全部分发给所述消息订阅者时，等待消息数据库中 写入新消息，并将写入的新消息分发给消息订阅者。
[0026] 进一步，当所述消息数据库的存储容量不足时，依据所述时间值和顺序值，逐条删 除最早写入所述消息数据库的消息。
[0027] 进一步，所述方法还包括：
[0028] 设置清理周期，在每个清理周期中，依据所述时间值和顺序值，逐条删除最早写入 所述消息数据库的消息。
[0029] 进一步，所述方法还包括：
[0030] 设置所述消息数据库的数据量阈值，所述数据量阈值小于所述消息数据库的存储 容量；
[0031] 在删除最早写入所述消息数据库的消息时，当所述消息数据库的数据量降到所述 数据量阈值时，停止继续对所述消息数据库中消息的删除。
[0032] 一种消息分发装置，包括：
[0033]消息接收模块，用于接收消息生产者所创建的消息，并将所接收的消息写入消息 数据库；
[0034] 消息订阅模块，用于获取消息订阅者的消息订阅范围，根据所述消息订阅范围确 定所述消息数据库中的起始消息订阅点；
[0035] 消息分发模块，用于从所述起始消息订阅点开始，将所述消息订阅者的消息订阅 范围内的各条消息分发给所述消息订阅者；以及，
[0036] 消息数据库，用于存储所述消息。
[0037]进一步，所述消息接收模块，还用于将所接收的消息写入消息数据库时，在所述消 息数据库中同时记录写入消息时的时间值和写入的消息的顺序值。
[0038]进一步，所述消息订阅模块根据所述消息订阅范围，在所述消息数据库中确定所 述消息订阅范围内的最早写入的消息，并将其作为所述起始消息订阅点。
[0039]进一步，所述消息分发模块依据所述时间值和顺序值，将所述消息订阅者的消息 订阅范围内的各条消息逐条分发给所述消息订阅者。
[0040]进一步，在将所述消息订阅者的消息订阅范围内的各条消息逐条分发给所述消息 订阅者的同时，所述消息分发模块将所分发的各条消息的时间值和顺序值同时发送给所述 消息订阅者；
[0041] 当连接中断或者所述消息订阅者所接收的消息丢失时：
[0042]所述消息订阅模块将最近一次成功分发的消息作为新的起始消息订阅点； [0043]所述消息分发模块从所述新的起始消息订阅点开始，将所述消息订阅者的消息订 阅范围内尚未分发的各条消息分发给所述消息订阅者。
[0044]进一步，所述消息分发模块在消息订阅者的消息订阅范围的最晚时间晚于当前时 间，且当所述消息数据库中的所有消息订阅范围内的消息全部分发给所述消息订阅者时， 等待消息数据库中写入新消息，并将写入的新消息分发给消息订阅者。
[0045] 进一步，所述装置还包括：
[0046]数据清理模块，用于当所述消息数据库的存储容量不足时，依据所述时间值和顺 序值，逐条删除最早写入所述消息数据库的消息。
[0047] 进一步，所述装置还包括：
[0048]数据清理模块，用于设置清理周期，并在每个清理周期中，依据所述时间值和顺序 值，逐条删除最早写入所述消息数据库的消息。
[0049]进一步，所述数据清理模块在删除最早写入所述消息数据库的消息时，当所述消 息数据库的数据量降到所述数据量阈值时，停止继续对所述消息数据库中消息的删除；
[0050] 其中，所述数据量阈值小于所述消息数据库的存储容量。
[0051] 从上述方案可以看出，本发明的消息分发方法和装置中，利用消息数据库对消息 生产者所创建的消息进行保存，并根据消息订阅者的消息订阅范围向所述消息订阅者提供 所保存的消息，另外，在消息生产者所创建的消息进行保存的同时，在所述消息数据库中还 记录写入消息时的时间值和写入的消息的顺序值，进而实现了依据时间值和顺序值确定消 息订阅者所获得的消息以及尚未获得的消息，当连接丢失或者消息订阅者所接收的消息丢 失时，依据时间值和顺序值可实现消息订阅者对订阅消息的续传和补传，同时，通过重新设 定消息订阅范围还可实现消息订阅者对已订阅消息的一致性检查，例如消息订阅者通过设 定消息订阅范围以取出任意一段时间范围内的消息进而与其已经接收到的消息进行一致 性检查。本发明的消息分发方法和装置保证了消息订阅时，当订阅消息已经分发到订阅者 后，仍然对所订阅消息的保存，同时针对多个消息订阅者仅使用一份消息数据库的拷贝，多 个消息订阅者仅通过其所接收的消息的时间值和顺序值确定各自已经获得的所订阅的消 息和尚未获得的消息，而不必为每个消息订阅者分别使用不同的消息数据的拷贝，进而节 约了消息的存储资源。
【附图说明】
[0052] 图1为本发明的消息分发方法的实施例流程图；
[0053] 图2为本发明的消息分发装置的实施例结构图。
【具体实施方式】
[0054] 为了使本发明的目的、技术方案及优点更加清楚明白，以下参照附图并举实施例， 对本发明作进一步详细说明。
[0055] 如图1所示，本发明的消息分发方法实施例，主要包括以下步骤：
[0056] 步骤1、接收消息生产者所创建的消息，并将所接收的消息写入消息数据库；
[0057] 步骤2、获取消息订阅者的消息订阅范围，根据所述消息订阅范围确定所述消息数 据库中的起始消息订阅点；
[0058] 步骤3、从所述起始消息订阅点开始，将所述消息订阅者的消息订阅范围内的各条 消息分发给所述消息订阅者。
[0059] 为了实现消息订阅者对所订阅消息的断点续传、消息核对等功能，在步骤1中还 进一步包括：
[0060] 将所接收的消息写入消息数据库时，在所述消息数据库中同时记录写入消息时的 时间值和写入的消息的顺序值。
[0061] 其中，时间值包括写入消息的日期及写入消息的时刻，例如写入的某条消息的时 间值为2015-5-25 13:00:36,表明该条消息的写入时间（即在消息数据库中该条消息记录 的创建时间）为2015年5月25日下午13点00分36秒。
[0062] 顺序值为在单位时间内写入的信息的数量从小到大进行排列的顺序值。例如单位 时间按秒计算，则若在2015年5月25日下午13点00分36秒写入数据库300条消息，则从 2015年5月25日下午13点00分36秒的1秒钟时间段内，按照先后顺序将这300条消息 从[000001]到[000300]分别赋予顺序值，之后若在2015年5月25日下午13点00分37 秒写入数据库500条消息，则从2015年5月25日下午13点00分37秒的1秒钟时间段内， 按照先后顺序将这500条消息从[000001]到[000500]分别赋予顺序值，以此类推。通过 消息数据库中的每条消息的时间值和顺序值，即可得知该条消息是消息数据库中在哪一时 间段的第几条消息，进而，为随后的消息订阅范围提供依据并可进一步实现消息断点续传、 消息核对等效果，参见后续说明。
[0063] 步骤2中，根据所述消息订阅范围确定所述消息数据库中的起始消息订阅点包 括：
[0064] 根据所述消息订阅范围，在所述消息数据库中确定所述消息订阅范围内的最早写 入的消息，并将其作为所述起始消息订阅点。
[0065] 例如，若消息订阅者的消息订阅范围设定为2015年5月25日下午13点00分36 秒以后的所有消息，则可通过时间值，在所述消息数据库中确定所述消息订阅范围内的最 早写入的消息。结合表1对此进行说明。
[0066]表1:消息数据库中的消息数据
[0068] 表1表示消息数据库中存储的各条消息数据记录，其中各条消息的顺序值在每 个时间周期（秒）内依据写入消息数据库的顺序而递加，其中例如时间值为2015-05-25 13:00:36顺序值为[000125]的消息，表明该消息是2015年05月25日13点00分36秒的 这一秒时间内写入的第125条消息，其余各条消息数据与之相似，不再赘述。若消息订阅者 的消息订阅范围设定为2015年5月25日13点00分36秒以后的所有消息，则通过时间值， 在所述消息数据库中确定所述消息订阅范围内的最早写入的消息的时间值为2015-05-25 13:00:36、顺序值为[000001]的消息。
[0069] 结合所述消息数据库中同时记录写入消息时的时间值和写入的消息的顺序值，本 发明步骤3的从所述起始消息订阅点开始，将所述消息订阅者的消息订阅范围内的各条消 息分发给所述消息订阅者，包括：
[0070] 依据所述时间值和顺序值，将所述消息订阅者的消息订阅范围内的各条消息逐条 分发给所述消息订阅者。
[0071] 继续结合表1进行说明。本发明步骤3的实施例中，依据所述时间值和顺序值， 将2015年5月25日13点00分36秒以后的所有消息的订阅范围内的各条消息，从时间 值为2015-05-25 13:00:36、顺序值为[000001]的消息开始逐条分发给所述消息订阅者， 即先分发时间值为2015-05-25 13:00:36、顺序值为[000001]的消息，然后分发时间值为 2015-05-25 13:00:36、顺序值为[000002]的消息，待时间值为 2015-05-25 13:00:36 所有 消息分发完毕后，再分发时间值为2015-05-25 13:00:37的各条消息，消息的分发顺序按 照顺序值从小到大进行，之后再分发 时间值为2015-05-25 13:00:38的各条消息，以此类 推。
[0072] 本发明步骤3的实施例中，在将所述消息订阅者的消息订阅范围内的各条消息逐 条分发给所述消息订阅者的同时，将所分发的各条消息的时间值和顺序值同时发送给所述 消息订阅者；进而，当连接中断或者所述消息订阅者所接收的消息丢失时，可确定最近一次 成功分发的消息的时间值和顺序值，进而将最近一次成功分发的消息作为新的起始消息订 阅点，并从所述新的起始消息订阅点开始，将所述消息订阅者的消息订阅范围内尚未分发 的各条消息分发给所述消息订阅者。例如，在将所述消息订阅者的消息订阅范围内的各条 消息逐条分发给所述消息订阅者时，当成功分发了时间值为2015-05-25 13:00:36、顺序值 为[000078]的消息后连接中断，则将时间值为2015-05-25 13:00:36、顺序值为[000078] 的消息作为新的起始消息订阅点，在连接回复后，从时间值为2015-05-25 13:00:36、顺序 值为[000078]的消息开始，将所述消息订阅者的消息订阅范围内尚未分发的各条消息分 发给所述消息订阅者。
[0073] 本发明的消息分发方法中，若消息订阅者的消息订阅范围的最晚时间晚于当前时 间，且当所述消息数据库中的所有消息订阅范围内的消息全部分发给所述消息订阅者时， 等待消息数据库中写入新消息，并将写入的新消息分发给消息订阅者。例如，若消息订阅者 的消息订阅范围设定为2015年5月25日13点00分36秒以后的所有消息，则消息订阅者 的消息订阅范围的最晚时间是晚于当前时间的，在将当前消息数据库中所述消息订阅者的 消息订阅范围内的所有消息全部分发给消息订阅者后，因为消息订阅者的消息订阅范围的 最晚时间晚于当前时间，则继续等待消息数据库中写入的新消息，当消息数据库中写入新 的消息后，将写入的新消息分发给消息订阅者，并继续等待消息数据库中写入的新消息，以 此循环。
[0074] 本发明的消息分发方法实施例中，当所述消息数据库的存储容量不足时，依据所 述时间值和顺序值，逐条删除最早写入所述消息数据库的消息。进而，保证新的消息能够及 时保存，避免丢失，同时过早的消息能够进行清理以避免对存储资源的不必要占用。另外， 本发明的消息分发方法实施例中还包括，设置清理周期，在每个清理周期中，依据所述时间 值和顺序值，逐条删除最早写入所述消息数据库的消息。进而同样能够保证新的消息能够 及时保存，避免丢失，以及过早的消息能够进行清理以避免对存储资源的不必要占用。
[0075] 在删除旧有数据的同时，需要保存较近一段时间所保存的信息，以保证消息订阅 者的消息获取，本发明的消息分发方法实施例中还包括，设置所述消息数据库的数据量阈 值，所述数据量阈值小于所述消息数据库的存储容量；在删除最早写入所述消息数据库的 消息时，当所述消息数据库的数据量降到所述数据量阈值时，停止继续对所述消息数据库 中消息的删除。其中，数据量阈值例如保存的信息条数或者保存的时间范围等。通过设定 恰当的数据量阈值，能够保证消息订阅者的对所需要的有用消息的获取，以及对过早的已 经无用的消息的清理，实现消息数据库中存储空间的循环使用。
[0076] 本发明实施例同时还提供了一种消息分发装置，如图2所示，其包括消息数据库 11、消息接收模块12、消息订阅模块13和消息分发模块14。其中，所述消息数据库11用于 存储消息。所述消息接收模块12,用于接收消息生产者2所创建的消息，并将所接收的消 息写入消息数据库11。进一步地，所述消息接收模块12,还用于将所接收的消息写入消息 数据库11时，在所述消息数据库11中同时记录写入消息时的时间值和写入的消息的顺序 值。消息订阅模块13,用于获取消息订阅者3的消息订阅范围，根据所述消息订阅范围确定 所述消息数据库11中的起始消息订阅点。具体地，所述消息订阅模块13根据所述消息订 阅范围，在所述消息数据库11中确定所述消息订阅范围内的最早写入的消息，并将其作为 所述起始消息订阅点。消息分发模块14,用于从所述起始消息订阅点开始，将所述消息订阅 者3的消息订阅范围内的各条消息分发给所述消息订阅者3。具体地，所述消息分发模块 14依据所述时间值和顺序值，将所述消息订阅者3的消息订阅范围内的各条消息逐条分发 给所述消息订阅者。
[0077]另外，本发明的消息分发装置实施例中，在将所述消息订阅者3的消息订阅范围 内的各条消息逐条分发给所述消息订阅者3的同时，所述消息分发模块14将所分发的各条 消息的时间值和顺序值同时发送给所述消息订阅者。进而，当连接中断或者所述消息订阅 者所接收的消息丢失时，所述消息订阅模块13将最近一次成功分发给消息订阅者3的消息 作为新的起始消息订阅点；所述消息分发模块14从所述新的起始消息订阅点开始，将所述 消息订阅者3的消息订阅范围内尚未分发的各条消息分发给所述消息订阅者3。进而实现 了断点续传功能。
[0078] 本发明的消息分发装置实施例中，所述消息分发模块14在消息订阅者3的消息订 阅范围的最晚时间晚于当前时间，且当所述消息数据库11中的所有消息订阅范围内的消 息全部分发给所述消息订阅者时，等待消息数据库11中写入新消息，并将写入的新消息分 发给消息订阅者3。进而实现了对当前数据的及时分发。
[0079] 本发明的消息分发装置还进一步包括数据清理模块15,用于当所述消息数据库 11的存储容量不足时，依据所述时间值和顺序值，逐条删除最早写入所述消息数据库11的 消息。例如，当所述消息数据库11的存储容量过满而无法继续存储新的消息时，所述数据 清理模块15依据所述时间值和顺序值，逐条删除最早写入所述消息数据库11的消息。所述 数据清理模块15,还用于设置清理周期，并在每个清理周期中，依据所述时间值和顺序值， 逐条删除最早写入所述消息数据库11的消息。所述数据清理模块15在删除最早写入所述 消息数据库11的消息时，当所述消息数据库11的数据量降到所述数据量阈值时，停止继续 对所述消息数据库11中消息的删除。其中，所述数据量阈值小于所述消息数据库11的存 储容量。
[0080] 本发明中，将所分发的各条消息的时间值和顺序值同时发送给所述消息订阅者， 进而间接地实现了消息订阅者通过消息的时间值和顺序值对自己所订阅消息的管理。通过 消息的时间值和订阅值，消息订阅者可获知自己所订阅的消息，进而不必由消息分发装置 单独记录多个消息订阅者各自所订阅的消息范围和已经分发给各个消息订阅者的消息。由 此可见，本发明的消息分发方法和装置仅采用一个消息数据库也同样实现了对多个消息订 阅者消息订阅的支持。
[0081 ] 本发明的消息分发方法和装置，利用消息数据库对消息生产者所创建的消息进行 保存，并根据消息订阅者的消息订阅范围向所述消息订阅者提供所保存的消息，另外，在消 息生产者所创建的消息进行保存的同时，在所述消息数据库中还记录写入消息时的时间值 和写入的消息的顺序值，进而实现了依据时间值和顺序值确定消息订阅者所获得的消息以 及尚未获得的消息，当连接丢失或者消息订阅者所接收的消息丢失时，依据时间值和顺序 值可实现消息订阅者对订阅消息的续传和补传 ，同时，通过重新设定消息订阅范围还可实 现消息订阅者对已订阅消息的一致性检查，例如消息订阅者通过设定消息订阅范围以取出 任意一段时间范围内的消息进而与其已经接收到的消息进行一致性检查。本发明的消息分 发方法和装置保证了消息订阅时，当订阅消息已经分发到订阅者后，仍然对所订阅消息的 保存，同时针对多个消息订阅者仅使用1份消息数据库的拷贝，多个消息订阅者仅通过其 所接收的消息的时间值和顺序值确定各自已经获得的所订阅的消息和尚未获得的消息，而 不必为每个消息订阅者分别使用不同的消息数据的拷贝，进而节约了消息的存储资源。 [0082] 以上所述仅为本发明的较佳实施例而已，并不用以限制本发明，凡在本发明的精 神和原则之内，所做的任何修改、等同替换、改进等，均应包含在本发明保护的范围之内。
【主权项】
1. 一种消息分发方法，包括： 接收消息生产者所创建的消息，并将所接收的消息写入消息数据库； 获取消息订阅者的消息订阅范围，根据所述消息订阅范围确定所述消息数据库中的起 始消息订阅点； 从所述起始消息订阅点开始，将所述消息订阅者的消息订阅范围内的各条消息分发给 所述消息订阅者。2. 根据权利要求1所述的消息分发方法，其特征在于：将所接收的消息写入消息数据 库时，在所述消息数据库中同时记录写入消息时的时间值和写入的消息的顺序值。3. 根据权利要求2所述的消息分发方法，其特征在于，根据所述消息订阅范围确定所 述消息数据库中的起始消息订阅点包括： 根据所述消息订阅范围，在所述消息数据库中确定所述消息订阅范围内的最早写入的 消息，并将其作为所述起始消息订阅点。4. 根据权利要求2所述的消息分发方法，其特征在于，从所述起始消息订阅点开始，将 所述消息订阅者的消息订阅范围内的各条消息分发给所述消息订阅者，包括： 依据所述时间值和顺序值，将所述消息订阅者的消息订阅范围内的各条消息逐条分发 给所述消息订阅者。5. 根据权利要求4所述的消息分发方法，其特征在于： 在将所述消息订阅者的消息订阅范围内的各条消息逐条分发给所述消息订阅者的同 时，将所分发的各条消息的时间值和顺序值同时发送给所述消息订阅者； 当连接中断或者所述消息订阅者所接收的消息丢失时，将最近一次成功分发的消息作 为新的起始消息订阅点，并从所述新的起始消息订阅点开始，将所述消息订阅者的消息订 阅范围内尚未分发的各条消息分发给所述消息订阅者。6. 根据权利要求2所述的消息分发方法，其特征在于： 若消息订阅者的消息订阅范围的最晚时间晚于当前时间，且当所述消息数据库中的所 有消息订阅范围内的消息全部分发给所述消息订阅者时，等待消息数据库中写入新消息， 并将写入的新消息分发给消息订阅者。7. 根据权利要求2所述的消息分发方法，其特征在于：当所述消息数据库的存储容量 不足时，依据所述时间值和顺序值，逐条删除最早写入所述消息数据库的消息。8. 根据权利要求2所述的消息分发方法，其特征在于，所述方法还包括： 设置清理周期，在每个清理周期中，依据所述时间值和顺序值，逐条删除最早写入所述 消息数据库的消息。9. 根据权利要求7或8所述的消息分发方法，其特征在于，所述方法还包括： 设置所述消息数据库的数据量阈值，所述数据量阈值小于所述消息数据库的存储容 量； 在删除最早写入所述消息数据库的消息时，当所述消息数据库的数据量降到所述数据 量阈值时，停止继续对所述消息数据库中消息的删除。10. -种消息分发装置，其特征在于，包括： 消息接收模块，用于接收消息生产者所创建的消息，并将所接收的消息写入消息数据 库； 消息订阅模块，用于获取消息订阅者的消息订阅范围，根据所述消息订阅范围确定所 述消息数据库中的起始消息订阅点； 消息分发模块，用于从所述起始消息订阅点开始，将所述消息订阅者的消息订阅范围 内的各条消息分发给所述消息订阅者；以及， 消息数据库，用于存储所述消息。11. 根据权利要求10所述的消息分发装置，其特征在于： 所述消息接收模块，还用于将所接收的消息写入消息数据库时，在所述消息数据库中 同时记录写入消息时的时间值和写入的消息的顺序值。12. 根据权利要求11所述的消息分发装置，其特征在于： 所述消息订阅模块根据所述消息订阅范围，在所述消息数据库中确定所述消息订阅范 围内的最早写入的消息，并将其作为所述起始消息订阅点。13. 根据权利要求11所述的消息分发装置，其特征在于： 所述消息分发模块依据所述时间值和顺序值，将所述消息订阅者的消息订阅范围内的 各条消息逐条分发给所述消息订阅者。14. 根据权利要求13所述的消息分发装置，其特征在于： 在将所述消息订阅者的消息订阅范围内的各条消息逐条分发给所述消息订阅者的同 时，所述消息分发模块将所分发的各条消息的时间值和顺序值同时发送给所述消息订阅 者； 当连接中断或者所述消息订阅者所接收的消息丢失时： 所述消息订阅模块将最近一次成功分发的消息作为新的起始消息订阅点； 所述消息分发模块从所述新的起始消息订阅点开始，将所述消息订阅者的消息订阅范 围内尚未分发的各条消息分发给所述消息订阅者。15. 根据权利要求11所述的消息分发装置，其特征在于： 所述消息分发模块在消息订阅者的消息订阅范围的最晚时间晚于当前时间，且当所述 消息数据库中的所有消息订阅范围内的消息全部分发给所述消息订阅者时，等待消息数据 库中写入新消息，并将写入的新消息分发给消息订阅者。16. 根据权利要求11所述的消息分发装置，其特征在于，所述装置还包括： 数据清理模块，用于当所述消息数据库的存储容量不足时，依据所述时间值和顺序值， 逐条删除最早写入所述消息数据库的消息。17. 根据权利要求11所述的消息分发装置，其特征在于，所述装置还包括： 数据清理模块，用于设置清理周期，并在每个清理周期中，依据所述时间值和顺序值， 逐条删除最早写入所述消息数据库的消息。18. 根据权利要求16或17所述的消息分发装置，其特征在于： 所述数据清理模块在删除最早写入所述消息数据库的消息时，当所述消息数据库的数 据量降到所述数据量阈值时，停止继续对所述消息数据库中消息的删除； 其中，所述数据量阈值小于所述消息数据库的存储容量。
【专利摘要】本发明公开了一种消息分发方法和装置，其包括：接收消息生产者所创建的消息，并将所接收的消息写入消息数据库；获取消息订阅者的消息订阅范围，根据所述消息订阅范围确定所述消息数据库中的起始消息订阅点；从所述起始消息订阅点开始，将所述消息订阅者的消息订阅范围内的各条消息分发给所述消息订阅者。本发明实现了依据时间值和顺序值确定消息订阅者所获得的消息以及尚未获得的消息，实现了消息订阅者对订阅消息的续传和补传，以及对订阅消息的一致性检查，本发明同时针对多个消息订阅者仅使用一份消息数据库的拷贝，而不必为每个消息订阅者分别使用不同的消息数据的拷贝，进而节约了消息的存储资源。
【IPC分类】G06F17/30
【公开号】CN104899277
【申请号】CN201510287977
【发明人】许文靖
【申请人】北京京东尚科信息技术有限公司, 北京京东世纪贸易有限公司
【公开日】2015年9月9日
【申请日】2015年5月29日
最新回复(0)