基于ActiveMQ通信的多租戶任務(wù)調(diào)度框架設(shè)計(jì)與實(shí)現(xiàn)

摘要：任務(wù)調(diào)度被廣泛應(yīng)用于軟件應(yīng)用系統(tǒng)中，為了降低耦合度，解決多租戶任務(wù)調(diào)度系統(tǒng)中任務(wù)調(diào)度和執(zhí)行結(jié)果的及時(shí)通知問(wèn)題，提出了在任務(wù)調(diào)度框架中應(yīng)用ActiveMQ中間件構(gòu)建消息總線進(jìn)行消息傳輸和通信的技術(shù)思路。內(nèi)容包括Ac-tiveMQ介紹、主流消息中間件比較、基于ActiveMQ通信的多租戶調(diào)度框架構(gòu)建以及框架實(shí)現(xiàn)。

關(guān)鍵詞：多租戶;ActiveMQ;任務(wù)調(diào)度;消息中間件;低耦合

中圖分類號(hào)：TP311 文獻(xiàn)標(biāo)識(shí)碼：A

文章編號(hào)：1009-3044（2020）18-0007-03

開(kāi)放科學(xué)（資源服務(wù)）標(biāo)識(shí)碼（OSID）：

1 背景

在數(shù)據(jù)傳輸及解譯、消息通知、可執(zhí)行程序調(diào)度等業(yè)務(wù)應(yīng)用場(chǎng)景中，往往需要系統(tǒng)在指定時(shí)間進(jìn)行特定操作，基于此類需求，任務(wù)調(diào)度框架被廣泛應(yīng)用于軟件應(yīng)用系統(tǒng)中。文獻(xiàn)[1-3]介紹了部分任務(wù)調(diào)度算法及任務(wù)調(diào)度框架構(gòu)建方法。但任務(wù)調(diào)度框架僅關(guān)注任務(wù)的調(diào)度及觸發(fā)，對(duì)任務(wù)執(zhí)行過(guò)程、執(zhí)行結(jié)果信息的及時(shí)通知能力較為欠缺，尤其是在多并發(fā)、異步任務(wù)執(zhí)行時(shí)，出現(xiàn)任務(wù)執(zhí)行超時(shí)、中斷、異常的情況下，無(wú)法及時(shí)將任務(wù)執(zhí)行信息通知主系統(tǒng)進(jìn)行自動(dòng)處理，為調(diào)度系統(tǒng)的可靠性和可伸縮性帶來(lái)較大弊端，嚴(yán)重影響業(yè)務(wù)系統(tǒng)連續(xù)運(yùn)行，增加了人力資源成本和維護(hù)成本。

通過(guò)對(duì)ActiveMQ的研究與應(yīng)用，解決多任務(wù)調(diào)度框架中出現(xiàn)的上述問(wèn)題，通過(guò)ActiveMQ構(gòu)建消息總線，完成系統(tǒng)內(nèi)部通信及系統(tǒng)內(nèi)外通信交互，降低系統(tǒng)耦合度，提高系統(tǒng)任務(wù)調(diào)度執(zhí)行的可靠性。

2 ActiveMQ概述

ActiveMQ是由Apache基金會(huì)提供的基于Java語(yǔ)言開(kāi)發(fā)的多協(xié)議開(kāi)源消息中間件，支持OpenWire、Stomp、REST、WSNoti-fication、XMPP、AMQP等多種應(yīng)用協(xié)議，可在Java、C、C++、C#、Rubv、Perl、Pvthon、PHP、.net等多種開(kāi)發(fā)語(yǔ)言環(huán)境用使用，提供交叉語(yǔ)言功能，并可跨平臺(tái)應(yīng)用。目前ActiveMQ已經(jīng)支持MQTT（消息隊(duì)列遙測(cè)傳輸）協(xié)議，可應(yīng)用于物聯(lián)網(wǎng)設(shè)備。其基于JMS規(guī)范提供了多種消息類型的傳輸，包括無(wú)有效負(fù)載消息Message、序列化對(duì)象消息ObjectMessage、文本消息TextMes-sage、鍵值集消息MapMessage、字節(jié)消息BytesMessage、數(shù)據(jù)流消息StreamMessage等，基本滿足系統(tǒng)常用消息通信類型要求，同時(shí)對(duì)JDBC和Journal的支持也滿足了消息持久化的要求[4]。

ActiveMQ消息通訊有兩種模式：主題（Topic）模式、隊(duì)列（ Queue）模式。主題模式，亦可稱為發(fā)布/訂閱模式，這種模式下一個(gè)消息可以被多個(gè)訂閱者接收，訂閱者訂閱一個(gè)主題后，只能接收到自其訂閱之后發(fā)布的信息。訂閱者如果在發(fā)布者發(fā)送消息之后啟動(dòng)，是無(wú)法接收到已經(jīng)發(fā)送的消息的，除非發(fā)布者已經(jīng)對(duì)消息進(jìn)行了持久化存儲(chǔ)。隊(duì)列模式中，ActiveMQ構(gòu)建一個(gè)消息存儲(chǔ)隊(duì)列，多個(gè)發(fā)送端可同時(shí)向隊(duì)列中發(fā)送消息，隊(duì)列被多個(gè)接收端同時(shí)監(jiān)聽(tīng)，但一個(gè)消息只能被一個(gè)接收端獲取，一旦被獲取就會(huì)消失，如果沒(méi)有被獲取，則會(huì)一直等待。

ActiveMQ是站在開(kāi)發(fā)者的角度被設(shè)計(jì)的，可以通過(guò)JCon-sole和ActiveMQ自帶的WebConsole工具等形式來(lái)監(jiān)控不同層面的數(shù)據(jù)，可通過(guò)配置的方式快速集成到Spring框架中，對(duì)二次開(kāi)發(fā)和調(diào)試十分友好。

3 ActiveMQ框架及與主流消息中間件的對(duì)比

3.1 ActiveMQ框架介紹

ActiveMQ主框架部分從結(jié)構(gòu)上可大致分為協(xié)議連接域、消息域、信息存儲(chǔ)域和網(wǎng)絡(luò)服務(wù)域幾部分，總體架構(gòu)如圖1所示。

其中協(xié)議連接域（ Connectors）封裝各類通信協(xié)議，用于創(chuàng)建和管理與代理通信客戶端的連接;

消息域（圖1 TopicRegion和Quene Region部分）封裝了Ac-tiveMQ的核心內(nèi)容，除主題模式和隊(duì)列模式兩種消息通信模式相關(guān)實(shí)現(xiàn)外，其最主要部分為Transports，包括Transport，Trans-portServer和TransportFactory等;

消息存儲(chǔ)域（ Message Store）定義了信息的緩存或持久化操作相關(guān)內(nèi)容，持久化操作方面支持AMQ Message Store、Ka-haDB、JDBC、LeveIDB及Journal，目前ActiveMQ的默認(rèn)持久化存儲(chǔ)為AMQ Message Store;

網(wǎng)絡(luò)服務(wù)域（ Network Services）定義了用于支持代理的網(wǎng)絡(luò)服務(wù)組件，包括遠(yuǎn)程服務(wù)檢測(cè)與發(fā)現(xiàn)、消息存儲(chǔ)和轉(zhuǎn)發(fā)、DR集群等。

3.2 主流消息中間件對(duì)比

目前消息中間件較多，且各具優(yōu)勢(shì)，比價(jià)突出的有Ac-tiveMQ、RabbitMQ、Kafka、RocketMQ，ZeroMQ等，在實(shí)際應(yīng)用過(guò)程中應(yīng)根據(jù)實(shí)際需求的優(yōu)先級(jí)進(jìn)行選型，下面給出部分消息中間件的對(duì)比。

4 多租戶調(diào)度框架中ActiveMQ應(yīng)用

4.1 多租戶調(diào)度框架設(shè)計(jì)

基于ActiveMQ通信的多租戶調(diào)度框架可分為租戶應(yīng)用、數(shù)據(jù)存儲(chǔ)、核心服務(wù)及業(yè)務(wù)操作四個(gè)主要部分，整體框架設(shè)計(jì)如圖2所示。

其中租戶應(yīng)用為用戶可視化操作界面，用于定義和操作調(diào)度業(yè)務(wù)，查看與監(jiān)控作業(yè)執(zhí)行狀態(tài)等。數(shù)據(jù)存儲(chǔ)部分將數(shù)據(jù)進(jìn)行分區(qū)存儲(chǔ)，分為消息通信數(shù)據(jù)、任務(wù)調(diào)度數(shù)據(jù)、業(yè)務(wù)支撐數(shù)據(jù)。所有租戶操作均以消息形式通知ActiveMQ消息總線[6]，為防止任務(wù)被重復(fù)執(zhí)行，系統(tǒng)通過(guò)隊(duì)列（Queue）模式向任務(wù)驅(qū)動(dòng)模塊傳輸信息，任務(wù)調(diào)度端消息代理獲取到租戶信息后，進(jìn)行指定任務(wù)定義并根據(jù)任務(wù)觸發(fā)器定義信息進(jìn)行立即執(zhí)行或在指定時(shí)間執(zhí)行，各業(yè)務(wù)操作模塊均可使用獨(dú)立線程執(zhí)行，提高調(diào)度效率，降低系統(tǒng)耦合。任務(wù)執(zhí)行完成后，將執(zhí)行信息和結(jié)果寫(xiě)回消息隊(duì)列，客戶端獲取后刷新?tīng)顟B(tài)列表。

對(duì)于系統(tǒng)租戶使用者，可通過(guò)操作界面直接調(diào)用已經(jīng)定義好的任務(wù)，通過(guò)任務(wù)開(kāi)放式接口進(jìn)行參數(shù)設(shè)置。對(duì)于二次開(kāi)發(fā)者，可遵循頂層任務(wù)接口規(guī)范，自定義任務(wù)插件邏輯和處理流程，測(cè)試后上傳至任務(wù)插件庫(kù)自動(dòng)加載供租戶使用。

4.2 框架實(shí)現(xiàn)

框架實(shí)現(xiàn)的核心是調(diào)度和通信。通過(guò)租戶ID與任務(wù)的關(guān)聯(lián)，對(duì)任務(wù)加以區(qū)分和控制，系統(tǒng)調(diào)度部分采用Quartz中間件[7]，租戶可通過(guò)可視化界面進(jìn)行Quartz Job的定義與驅(qū)動(dòng)[8]，可根據(jù)業(yè)務(wù)需要擴(kuò)展定義新的任務(wù)，在擴(kuò)展定義時(shí)需遵循Quartz框架要求繼承和實(shí)現(xiàn)Job類，為了實(shí)現(xiàn)系統(tǒng)任務(wù)的動(dòng)態(tài)加載和調(diào)度，在核心服務(wù)層定義了通用Job創(chuàng)建工廠類，通過(guò)反射模式創(chuàng)建任務(wù)。核心代碼如下：

Class<！--？ extends Job--> jobClass=new CascadingClassLoad-HelperO.loadClass（tempjob.geUobClassName0， Job.class）;//動(dòng)態(tài)加載任務(wù)

JobDetail j obDetail= JobBuilder.newjob（j obClass）.withldenti-ty（tempjob. getjobKey0， tempjob. getGroupKey0）. withDescription（tempjob.getjobName O）.build0;//創(chuàng)建任務(wù)

Scheduler scheduler= SchedulerManager.getInstanceO.getDe-faultTemplatesO.getScheduler0;

scheduler.schedulejob（j obDetail.trigger）;//執(zhí)行任務(wù)

上述代碼中，tempjob為反序列化的任務(wù)屬性信息，由租戶定義并經(jīng)消息總線傳輸至服務(wù)層進(jìn)行處理，傳輸采用序列化文本信息。租戶應(yīng)用端與任務(wù)端均具備消息的接收與發(fā)送能力，其中租戶應(yīng)用端發(fā)送任務(wù)定義數(shù)據(jù)，接收任務(wù)執(zhí)行狀態(tài)和結(jié)果信息，而任務(wù)端與之相反。ActiveMQ傳輸部分核心代碼如下：

Templatejob tempjob=new Templatejob0. setName（”job”）.setGroup（"group”）.setTrigger（”25”）;//定義任務(wù)模板對(duì)象

TextMessage message=session.createTextMessage（tempjob.toString0）;//創(chuàng)建序列化任務(wù)消息對(duì)象

producer.send（message）;//發(fā)送消息

ActiveMQ支持多種消息類型，在實(shí)際業(yè)務(wù)場(chǎng)景中可考慮使用多種消息類型混合的方式，加強(qiáng)通信能力。

5 總結(jié)

基于ActiveMQ通信的多租戶任務(wù)調(diào)度框架即可供用戶直接使用，又可作為基礎(chǔ)框架快速完成二次開(kāi)發(fā)，該框架可用性高、耦合度低，可適用于多種業(yè)務(wù)場(chǎng)景。在對(duì)于資源集約化高度要求的情境下，可考慮將通過(guò)該框架對(duì)一個(gè)業(yè)務(wù)體系中的所有可執(zhí)行任務(wù)進(jìn)行統(tǒng)一管理調(diào)度，在框架通信穩(wěn)定性與服務(wù)高效性方面，可考慮ActiveMQ集群化部署配置，如何進(jìn)行分布式執(zhí)行與調(diào)度、優(yōu)化負(fù)載均衡將是框架進(jìn)一步的探究和擴(kuò)展方向。

參考文獻(xiàn)：

[1]汪瑩，陳新鵬，基于集群計(jì)算的任務(wù)調(diào)度算法研究[Jl.現(xiàn)代計(jì)算機(jī)，2020（9）：8-10，16.

[2]王秀，孫忠林，姜莉.任務(wù)定時(shí)調(diào)度在企業(yè)級(jí)開(kāi)發(fā)中的研究[J].電子科技，2015，28（10）：100-102，107.

[3]劉一田，劉士進(jìn).多租戶高可用并行任務(wù)調(diào)度框架[Jl.計(jì)算機(jī)系統(tǒng)應(yīng)用，2016，25（12）：280-284.

[4] Apache ActiveMQ. ActiveMQ HomePage[EB/OL].[2020-03-02].http：//activemq.apache.org.

[5]Apache ActiveMQ. ActiveMQ Architecture[EB/OLl. [2020-03-02].http：//activemq.apache.org/code-overview.

[6]戴俊，朱曉民.基于ActiveMQ的異步消息總線的設(shè)計(jì)與實(shí)現(xiàn)[J].計(jì)算機(jī)系統(tǒng)應(yīng)用，2010，19（8）： 215，254-257.

[7]張康.基于Quartz的分布式定時(shí)任務(wù)調(diào)度模塊的設(shè)計(jì)與實(shí)現(xiàn)[D].南京：南京大學(xué)，2019.

[8]葉剛.基于Quartz的可視化定時(shí)任務(wù)管理方案[J].電子技術(shù)與軟件工程，2018（17）：139-140.

【通聯(lián)編輯：謝媛媛】

作者簡(jiǎn)介：蘇志宏（1987-），男，廣西南寧人，高級(jí)工程師，本科，主要研究方向?yàn)橛?jì)算機(jī)應(yīng)用技術(shù)、氣象與自然資源應(yīng)用系統(tǒng)開(kāi)發(fā)研究。