一種實現分布式消息處理的系統及方法

2023-05-07 17:15:51 1

一種實現分布式消息處理的系統及方法
【專利摘要】本發明公開了一種實現分布式消息處理的系統及方法，用於提高消息處理系統的通用性及靈活性，該系統包括：表達式模塊中包括有規則項、函數、運算符、表達式、宏變量及觸發器；設置模塊，用於接收用戶利用表達式模塊建立的消息處理邏輯表達式，並將消息處理邏輯表達式發送給同步模塊；消息處理邏輯表達式由規則項、函數、運算符、表達式、宏變量中的一種或多種組成；同步模塊，用於將消息處理邏輯表達式實時發送給消息處理模塊；消息處理模塊，用於根據消息處理邏輯表達式，更新自身保存的消息處理邏輯表達式；接收所需處理的消息，讀取最新的消息處理邏輯表達式，按照最新的消息處理邏輯表達式對應的消息處理邏輯對所需處理的消息進行處理。
【專利說明】一種實現分布式消息處理的系統及方法
【技術領域】
[0001]本發明涉及消息處理【技術領域】，具體涉及一種實現分布式消息處理的系統及方法。
【背景技術】
[0002]隨著網際網路的業務發展，造就了很多的大型系統，這些大型系統一般都採用分布式的方式增加系統的並發能力。分布式消息處理系統一般在大型系統中處理各模塊數據交互和邏輯、系統產生數據的加工處理等等，其特點是消息的異步和並發處理吞吐量大，橫向擴展容易。
[0003]在現有技術中，分布式消息處理系統消息處理方式和邏輯都需要在系統啟動前定義好，其消息處理流程和處理邏輯在系統運行期間不能改變、或者只能做微小變更。如果需要變更處理流程，需要管理人員變更配置重啟系統，如果需要變更處理邏輯，需要開發人員修改和重新實現新的處理邏輯並重啟系統。這樣，如果消息處理邏輯發生改變，只能重新修改系統處理邏輯代碼並重啟系統，系統通用性很差。

【發明內容】

[0004]有鑑於此，本發明提供一種實現分布式消息處理的系統及方法，以解決現有技術中分布式消息處理系統存在通用性差、無法靈活變更的技術問題。
[0005]為解決上述問題，本發明提供的技術方案如下:
[0006]—種實現分布式消息處理的系統,所述系統包括:
[0007]表達式模塊、設置模塊、同步模塊以及至少一個消息處理模塊；
[0008]所述表達式模塊中包括有規則項、函數、運算符、表達式、宏變量以及觸發器；所述觸發器用於當所需處理的消息滿足消息處理邏輯表達式時觸發對應的處理操作；
[0009]所述設置模塊，用於接收用戶利用所述表達式模塊建立的消息處理邏輯表達式，並將所述消息處理邏輯表達式發送給同步模塊；所述消息處理邏輯表達式由所述規則項、所述函數、所述運算符、所述表達式、所述宏變量中的一種或多種組成；
[0010]所述同步模塊，用於將接收到的所述消息處理邏輯表達式實時發送給所述消息處理模塊；
[0011]消息處理模塊，用於根據所述同步模塊實時發送的消息處理邏輯表達式，更新自身保存的消息處理邏輯表達式；接收所需處理的消息，讀取最新的消息處理邏輯表達式，按照所述最新的消息處理邏輯表達式對應的消息處理邏輯對所述所需處理的消息進行處理。
[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]所述表達式模塊判斷用戶建立的消息處理邏輯表達式是否滿足表達式合法性規貝1J，如果是，觸發所述設置模塊將所述消息處理邏輯表達式發送給同步模塊，如果否，通過所述設置模塊提示用戶重新建立消息處理邏輯表達式。
[0043]由此可見，本發明實施例具有如下有益效果:
[0044]本發明實施例中提供一套表達式系統，抽象各種消息處理邏輯和實現方式，可以將處理邏輯中的規則項、函數、宏變量以及相互之間的關係組成消息處理邏輯表達式，也即消息的處理邏輯可以使用表達式系統中的內容組合實現，具有靈活性；通過實時接收用戶利用表達式系統建立的消息處理邏輯表達式，並同步到消息處理模塊，在系統不需要重啟的情況下，用戶在線修改消息處理邏輯表達式就能夠實現消息處理邏輯和流程的變更。在本發明實施例中消息的處理邏輯都可以使用邏輯表達式靈活組合實現，且消息處理邏輯可以隨時變更，從而實現了一個通用的、高響應、靈活擴展的分布式消息處理系統。
【專利附圖】

【附圖說明】
[0045]圖1為本發明實施例中提供的實現分布式消息處理的系統實施例的示意圖；
[0046]圖2為本發明實施例中提供的實現分布式消息處理的系統實施例的數據流圖；
[0047]圖3為本發明實施例中提供的實現分布式消息處理的方法實施例一的流程圖；
[0048]圖4為本發明實施例中提供的實現分布式消息處理的方法實施例二的流程圖。
【具體實施方式】
[0049]為使本發明的上述目的、特徵和優點能夠更加明顯易懂，下面結合附圖和【具體實施方式】對本發明實施例作進一步詳細的說明。
[0050]本發明實施例提供的實現分布式消息處理的系統及方法，是針對現有技術中消息處理方式和邏輯都需要在系統啟動前定義好，無法中途變更；制定處理邏輯的同時，系統要有相對應的處理程序，靈活性很低；當有新種類數據時候，只能添加或修改系統處理邏輯代碼實現新數據的處理，系統通用性很差的技術問題，提出從消息處理抽象角度考慮，設計實現一套邏輯表達式系統，以靈活實現實時消息處理變更的需求，其通過將各種處理方式、處理邏輯抽象為各種原子函數，這些原子函數能夠組合成各種複雜的處理邏輯。通過表達式系統，用戶只需要修改和建立各種消息處理邏輯表達式即可實現各種消息處理邏輯的整合。利用靈活通用的表達式系統，實現不用修改系統處理程序和重啟系統情況下，變更消息處理邏輯實現新消息處理。
[0051]基於上述思想，參見圖1所示，是本發明實施例提供的實現分布式消息處理的系統實施例，該系統可以包括:
[0052]表達式模塊101、設置模塊102、同步模塊103以及至少一個消息處理模塊104。
[0053]其中，表達式模塊101中包括有規則項、函數、運算符、表達式、宏變量以及觸發器；觸發器用於當所需處理的消息滿足消息處理邏輯表達式時觸發對應的處理操作。
[0054]表達式模塊是分布式消息處理系統能夠實現動態組合變更消息處理邏輯的核心模塊。該模塊抽象出一套通用的表達式配置功能，用戶能夠使用定義的內容實現各種簡單以及複雜的消息處理邏輯。表達式模塊可以包括如下部分:觸發器(trigger)、規則項(item)、函數(function)、表達式(rules)、宏變量(macro)、運算符(operator)。
[0055]規則項代表表達式模塊中的規則子項，反映消息的種類、來源或者消息表達的含義，各種消息都可以抽象成不同的規則項。
[0056]函數為消息處理邏輯表達式中用於處理消息的邏輯操作，比如最大值、最小值、中間值、平局值、不同、差值、計數、總和、方差等等。函數和規則項配合使用可以組成消息的處理邏輯。
[0057]運算符代表表達式內部或表達式之間的關係及運算，例如包括大於(>)、小於(O、等於( = )、不等於(#)、關係與(&)、關係非(I)等運算操作。
[0058]表達式可以是利用上述的規則項、函數、運算符等按照用戶的處理邏輯組成的表明某種類型的消息按照怎樣的處理邏輯處理以及輸出怎樣的結果的公式，也即表達式可以表示一種消息的處理邏輯。
[0059]宏變量表示抽象的統一替代參數，宏變量可以是一種結果或變量的抽象，當具體到特定實體時，可以替換成具體的實體值。在定義消息處理輸出結果時，為了定義通用輸出結果模板，可以使用宏變量實現，在具體的實例中，宏變量會被替換成實體的值。
[0060]觸發器是消息處理邏輯表達式和具體實體綁定後的特殊單位，是為了通知和標記某實體設置的消息處理邏輯表達式有數據滿足條件，觸發該實體處理滿足條件的消息。如果不需要將結果綁定到單個實體，可以不使用觸發器。
[0061]設置模塊102，用於接收用戶利用表達式模塊建立的消息處理邏輯表達式，並將消息處理邏輯表達式發送給同步模塊；消息處理邏輯表達式由規則項、函數、運算符、表達式、宏變量中的一種或多種組成。
[0062]設置模塊是用戶設置管理消息處理邏輯表達式的模塊，例如用戶可以增加、刪除、修改、查詢消息處理邏輯表達式，也即設置模塊可以接收用戶利用表達式模塊增加、刪除、修改的消息處理邏輯表達式，並為用戶提供已有消息處理邏輯表達式的查詢。消息處理邏輯表達式是消息處理流程和處理邏輯的體現，用戶通過設置模塊調用表達式模塊中的內容(規則項、函數、運算符、表達式、宏變量)可以實現定義各種消息處理邏輯表達式，每一個消息處理邏輯表達式實現一種消息處理流程和邏輯。消息處理邏輯表達式的輸出結果還可以與觸發器綁定，當所需處理的消息滿足消息處理邏輯表達式時觸發對應的處理操作。
[0063]為了更清晰的理解消息處理邏輯表達式，通過一個具體示例對消息處理邏輯表達式進行說明。表達式{PD[DI_pdState].last(#l)}#10為一個邏輯比較簡單的消息處理邏輯表達式，在這個表達式中H)為規則項，表示物理磁碟；DI_pdState為規則子項，表示物理磁碟的狀態值；last為函數，表示最近一段時間的消息值，last括號裡的內容表示消息的時間範圍等，#1表示最近一次；最後面的#為運算符不等於。整個消息處理邏輯表達式意思為物理磁碟的狀態值在最新的一次採集數據中的值不等於10。如果某個消息符合該表達式，就符合處理條件，可以觸發對應的操作處理。該表達式只是簡單的示例，在實際應用中利用表達式模塊可以組裝出更為複雜的消息處理邏輯表達式，例如包括多重表達式等。
[0064]在本發明的一些實施例中，表達式模塊還用於:
[0065]判斷用戶建立的消息處理邏輯表達式是否滿足表達式合法性規則，如果是，觸發設置模塊將消息處理邏輯表達式發送給同步模塊，如果否，通過設置模塊提示用戶重新建立消息處理邏輯表達式。
[0066]對消息處理邏輯表達式進行合法性規則驗證，以滿足用戶輸入的消息處理邏輯表達式可以被實現，在本實施例中僅將合法的消息處理邏輯表達式發送給同步模塊，可以保證後續消息處理流程的正確性。
[0067]同步模塊103，用於將接收到的消息處理邏輯表達式實時發送給消息處理模塊。
[0068]用戶對消息處理邏輯表達式的修改，通過同步模塊實時同步更新到各分布式消息處理模塊上。同步模塊可以使用socket通信方式，實時將用戶消息處理模塊更新到各個消息處理模塊上。對於同步模塊與消息處理模塊之間的通信方式在此並不進行限制。
[0069]消息處理模塊104，用於根據同步模塊實時發送的消息處理邏輯表達式，更新自身保存的消息處理邏輯表達式；接收所需處理的消息，讀取最新的消息處理邏輯表達式，按照最新的消息處理邏輯表達式對應的消息處理邏輯對所需處理的消息進行處理。
[0070]在本發明的一些實施例中，消息處理模塊可以包括:
[0071]更新單元，用於根據同步模塊實時發送的消息處理邏輯表達式，更新自身保存的消息處理邏輯表達式，這樣實現了用戶通過在線修改消息處理邏輯表達式能夠使消息處理流程和邏輯實現進行變更。
[0072]消息初選單元，用於接收所需處理的消息，讀取最新的消息處理邏輯表達式，對所需處理的消息進行是否符合最新的消息處理邏輯表達式的篩選，將篩選通過的消息放入消息隊列。
[0073]綜合處理單元，用於從消息隊列讀取篩選通過的消息，讀取最新的消息處理邏輯表達式；按照最新的消息處理邏輯表達式對應的消息處理邏輯對篩選通過的消息進行處理。
[0074]這樣，消息處理可以分為兩層，一層為消息初選:對接收到的消息，使用消息處理邏輯表達式的處理邏輯，初步篩選符合消息處理邏輯表達式的消息，放入消息隊列，等待後續處理；另外一層為消息綜合處理:對篩選出的消息利用對應的消息處理邏輯表達式，按流程和處理邏輯處理數據。使用分層結構，是考慮到數據初選層會收集到大量的消息，這些消息中有很多不符合表達式匹配的消息，過濾掉這些消息，能夠讓消息綜合處理層性能更高，邏輯更簡單。
[0075]在本發明的一些實施例中，綜合處理單元可以包括:
[0076]第一讀取子單元，用於從消息隊列讀取篩選通過的消息；
[0077]第二讀取子單元，用於讀取最新的消息處理邏輯表達式；
[0078]拆分子單元，用於將最新的消息處理邏輯表達式拆分為表達式最小單元；
[0079]處理子單元，用於按照表達式最小單元對應的消息處理邏輯對篩選通過的消息進行處理。
[0080]也即綜合處理單元從消息隊列中取出篩選後的消息，對消息對應的消息處理邏輯表達式進行拆分出來，將複雜表達式分解為各個表達式最小單元，分解後的表達式子項都有對應的處理邏輯。按照實現各表達式子項的處理邏輯和處理函數逐項處理，直到整個表達式處理結束，一條消息處理完畢。這樣，通過對表達式進行拆分為最小單元，而最小單元都有對應的消息處理邏輯，可以實現用戶修改了消息處理邏輯表達式，但是消息處理邏輯依然可以正確運行，保證了系統的通用性。
[0081]消息初選單元與綜合處理單元使用的消息處理邏輯表達式雖然相同，但處理方式與目的並不相同。繼續使用消息處理邏輯表達式:{PD[DI_pdState].last (#1)} #10為例，在消息初選單元，讀取一些消息後會映射ro[DI_pdState]段，確定消息類型是:物理磁碟狀態值，同時確定該值是否#10 (不等於10)，如果該條消息中物理磁碟狀態值不等於10就符合條件，放入消息隊列；在綜合處理單元，使用拆分表達式的方式將該消息處理邏輯表達式拆分成4部分:PD、DI_pdState、last (#1)、#10,其中PD、DI_pdState兩個欄位表示消息類型，last(#l)是帶參數的函數，表示該類型數據的最近一次的值，最後一個#10，表示該值不等於10，當滿足這個條件後，會根據PD、DI_pdState，按預定義模板生成結果，送入結果隊列。
[0082]在本發明的一些實施例中，消息處理單元還可以包括:
[0083]確認更新單元，用於在更新自身保存的消息處理邏輯表達式之後，判斷更新是否成功，如果是，向同步模塊發送同步成功信號，以使同步模塊將同步成功信號發送給設置模塊，如果否，向同步模塊發送失敗信號，以使同步模塊將同步失敗信號發送給設置模塊，使設置模塊重新將接收用戶利用表達式模塊建立的消息處理邏輯表達式發送給同步模塊。
[0084]消息處理單元監聽到消息處理邏輯表達式變更(例如增加、刪除、修改)後，更新消息處理邏輯表達式內存結構體中的內容，如果成功修改，更新程序會發送一個確認消息到同步模塊，如修改未成功，也會發送失敗消息到同步模塊，同時附加錯誤原因等。
[0085]基於上述實施例，在本發明的一些實施例中，本發明實施例提供的實現分布式消息處理的系統實施例還可以包括:
[0086]結果輸出模塊，用於將消息處理單元輸出的處理結果發送至對應的結果接收裝置，結果接收裝置包括消息隊列、簡訊發送系統、郵件發送系統、資料庫以及系統模塊。
[0087]消息被處理後會有相應的結果輸出，包括處理結果和接收體，接收體例如簡訊系統、郵件系統、消息隊列、系統模塊、資料庫等等。
[0088]這樣，本發明實施例中通過提供一套表達式系統，抽象各種消息處理邏輯和實現方式，可以將處理邏輯中的規則項、函數、宏變量以及相互之間的關係組成消息處理邏輯表達式，也即消息的處理邏輯可以使用表達式系統中的內容組合實現，具有靈活性；通過實時接收用戶利用表達式系統建立的消息處理邏輯表達式，並同步到消息處理模塊，在系統不需要重啟的情況下，用戶在線修改消息處理邏輯表達式就能夠實現消息處理邏輯和流程的變更。
[0089]本發明實施例將常見數據處理抽象為一套表達式系統，能夠自由組合和實現邏輯，同時易於擴展；消息的處理邏輯都可使用消息處理邏輯表達式組合實現，消息處理邏輯可以隨時變更；可以在系統運行中無縫變更和新增消息處理邏輯，不用重啟系統，從而實現了一個通用的、高響應、靈活擴展的分布式消息處理系統。
[0090]參見圖2所示，是本發明實施例中提供的實現分布式消息處理系統的數據流圖。實現分布式消息處理可以分為消息處理邏輯表達式變更流程以及消息處理流程。
[0091]A、消息處理邏輯表達式變更流程:
[0092]Al、用戶通過設置模塊界面修改(增加、刪除、修改、查詢)消息處理邏輯表達式。
[0093]A2、表達式模塊提供規則項、函數、運算符、表達式、宏變量、觸發器。用戶修改或新增的消息處理邏輯表達式由表達式模塊提供組合、驗證、存儲等。
[0094]A3、消息處理邏輯表達式創建完成後，並且通過表達式模塊的合法性驗證，合法消息處理邏輯表達式會通過設置模塊傳送到同步模塊。
[0095]A4、同步模塊使用socket通信方式同步到分布式的各消息處理模塊中的消息初選單元以及綜合處理單元。消息同步之後，各消息處理模塊內部自帶的更新程序修改、合併、更新消息處理邏輯表達式。完成之後返回確認消息，同步模塊把各消息處理模塊同步結果返回給設置模塊，設置模塊通知用戶消息處理邏輯表達式是否成功變更和同步，用戶可以根據結果選擇合適的操作。
[0096]B、消息處理流程:
[0097]B1、外部待處理的消息通過接口傳入消息處理模塊。(在分布式消息處理模塊中，可以存在多個消息初選單元，取決於系統消息並發量和系統處理能力)。
[0098]B2、消息初選單元使用消息處理邏輯表達式，初次匹配數據，篩選出符合消息處理邏輯表達式的消息，送入消息隊列中。初選模塊只是做消息的初步篩選，真正的處理流程和處理邏輯由綜合處理單元完成。採用這種分層處理的設計是因為在大量的消息中，可能存在大量不符合消息處理邏輯表達式的消息，由消息初選單元忽略這些消息，可以降低後續消息處理的壓力。如果不分層，消息處理集中在一個處理模塊完成，會造成單模塊壓力過大，處理並發量有限，同時，分布式消息處理系統可能需要綜合處理消息，相互關聯的數據分布在不同的節點上時候造成處理困難等。
[0099]B3、綜合處理單元從消息隊列中取出篩選後的數據(可並發多進程、多線程處理)，對消息對應的消息處理邏輯表達式進行拆分出來，將複雜表達式分解為各個表達式最小單元，分解後的表達式子項都有對應的處理邏輯。按照實現各表達式子項的處理邏輯和處理函數逐項處理，直到整個表達式處理結束，一條消息處理完畢。
[0100]B4、綜合處理單元對消息處理完成後，需要輸出消息處理結果，可以放入結果隊列，也可以持久化到資料庫或文件中。
[0101]B5、郵件或簡訊發送程序從結果隊列中取出結果，發送郵件或簡訊等。還可以擴展其他的結果輸出方式。
[0102]相應的，參見圖3所示，是本發明實施例提供的實現分布式消息處理的方法實施例一的流程圖，可以包括以下步驟:
[0103]步驟301:設置模塊接收用戶利用表達式模塊建立的消息處理邏輯表達式，並將消息處理邏輯表達式發送給同步模塊；表達式模塊中包括有規則項、函數、運算符、表達式、宏變量以及觸發器；消息處理邏輯表達式由規則項、函數、運算符、表達式、宏變量中的一種或多種組成；觸發器用於當所需處理的消息滿足消息處理邏輯表達式時觸發對應的處理操作。
[0104]步驟302:同步模塊將接收到的消息處理邏輯表達式實時發送給消息處理模塊。
[0105]步驟303:消息處理模塊根據同步模塊實時發送的消息處理邏輯表達式，更新自身保存的消息處理邏輯表達式；接收所需處理的消息，讀取最新的消息處理邏輯表達式，按照最新的消息處理邏輯表達式對應的消息處理邏輯對所需處理的消息進行處理。
[0106]在本發明的一些實施例中，按照最新的消息處理邏輯表達式對應的消息處理邏輯對所需處理的消息進行處理的具體實現可以包括:
[0107]對所需處理的消息進行是否符合最新的消息處理邏輯表達式的篩選，將篩選通過的消息放入消息隊列；
[0108]從消息隊列讀取篩選通過的消息，按照最新的消息處理邏輯表達式對應的消息處理邏輯對篩選通過的消息進行處理。
[0109]在本發明的一些實施例中，按照最新的消息處理邏輯表達式對應的消息處理邏輯對篩選通過的消息進行處理的具體實現可以包括:
[0110]將最新的消息處理邏輯表達式拆分為表達式最小單元；
[0111]按照表達式最小單元對應的消息處理邏輯對篩選通過的消息進行處理。
[0112]在本發明的一些實施例中，本發明實施例提供的實現分布式消息處理的方法實施例還可以包括:
[0113]結果輸出模塊將消息處理單元輸出的處理結果發送至對應的結果接收裝置，結果接收裝置包括消息隊列、簡訊發送系統、郵件發送系統、資料庫以及系統模塊。
[0114]在本發明的一些實施例中，在消息處理單元更新自身保存的消息處理邏輯表達式之後，本發明實施例提供的實現分布式消息處理的方法實施例還可以包括:
[0115]消息處理單元判斷更新是否成功，如果是，向同步模塊發送同步成功信號，以使同步模塊將同步成功信號發送給設置模塊，如果否，向同步模塊發送失敗信號，以使同步模塊將同步失敗信號發送給設置模塊，使設置模塊重新將接收用戶利用表達式模塊建立的消息處理邏輯表達式發送給同步模塊。
[0116]本發明實施例提供的實現分布式消息處理的方法實施例還可以包括:
[0117]表達式模塊判斷用戶建立的消息處理邏輯表達式是否滿足表達式合法性規則，如果是，觸發設置模塊將消息處理邏輯表達式發送給同步模塊，如果否，通過設置模塊提示用戶重新建立消息處理邏輯表達式。
[0118]參見圖4所示，是本發明實施例提供的實現分布式消息處理的方法實施例二的流程圖，可以包括以下步驟:
[0119]步驟401:設置模塊接收用戶利用表達式模塊建立的消息處理邏輯表達式。[0120]步驟402:表達式模塊判斷用戶建立的消息處理邏輯表達式是否滿足表達式合法性規則，如果是，進入步驟403，如果否，進入步驟404。
[0121]步驟403:設置模塊將消息處理邏輯表達式發送給同步模塊，並進入步驟405。
[0122]步驟404:通過設置模塊提示用戶重新建立消息處理邏輯表達式，返回步驟401。
[0123]步驟405:同步模塊將接收到的消息處理邏輯表達式實時發送給消息處理模塊。
[0124]步驟406:消息處理模塊根據同步模塊實時發送的消息處理邏輯表達式，更新自身保存的消息處理邏輯表達式。
[0125]步驟407:消息處理單元判斷更新是否成功，如果是，進入步驟408，如果否，執行步驟409。
[0126]步驟408:向同步模塊發送同步成功信號，以使同步模塊將同步成功信號發送給設置模塊，並進入步驟410。[0127]步驟409:向同步模塊發送失敗信號，以使同步模塊將同步失敗信號發送給設置模塊，返回步驟403。
[0128]步驟410:消息處理模塊接收所需處理的消息，讀取最新的消息處理邏輯表達式，按照最新的消息處理邏輯表達式對應的消息處理邏輯對所需處理的消息進行處理。
[0129]步驟411:結果輸出模塊將消息處理單元輸出的處理結果發送至對應的結果接收裝置，結果接收裝置包括消息隊列、簡訊發送系統、郵件發送系統、資料庫以及系統模塊。
[0130]這樣，本發明實施例中提供一套表達式系統，抽象各種消息處理邏輯和實現方式，可以將處理邏輯中的規則項、函數、宏變量以及相互之間的關係組成消息處理邏輯表達式，也即消息的處理邏輯可以使用表達式系統中的內容組合實現，具有靈活性；通過實時接收用戶利用表達式系統建立的消息處理邏輯表達式，並同步到消息處理模塊，在系統不需要重啟的情況下，用戶在線修改消息處理邏輯表達式就能夠實現消息處理邏輯和流程的變更。在本發明實施例中消息的處理邏輯都可以使用邏輯表達式靈活組合實現，且消息處理邏輯可以隨時變更，從而實現了一個通用的、高響應、靈活擴展的分布式消息處理系統。
[0131]需要說明的是，本說明書中各個實施例採用遞進的方式描述，每個實施例重點說明的都是與其他實施例的不同之處，各個實施例之間相同相似部分互相參見即可。對於實施例公開的系統或裝置而言，由於其與實施例公開的方法相對應，所以描述的比較簡單，相關之處參見方法部分說明即可。
[0132]還需要說明的是，在本文中，諸如第一和第二等之類的關係術語僅僅用來將一個實體或者操作與另一個實體或操作區分開來，而不一定要求或者暗示這些實體或操作之間存在任何這種實際的關係或者順序。而且，術語「包括」、「包含」或者其任何其他變體意在涵蓋非排他性的包含，從而使得包括一系列要素的過程、方法、物品或者設備不僅包括那些要素，而且還包括沒有明確列出的其他要素，或者是還包括為這種過程、方法、物品或者設備所固有的要素。在沒有更多限制的情況下，由語句「包括一個……」限定的要素，並不排除在包括所述要素的過程、方法、物品或者設備中還存在另外的相同要素。
[0133]結合本文中所公開的實施例描述的方法或算法的步驟可以直接用硬體、處理器執行的軟體模塊，或者二者的結合來實施。軟體模塊可以置於隨機存儲器(RAM)、內存、只讀存儲器(ROM)、電可編程ROM、電可擦除可編程ROM、寄存器、硬碟、可移動磁碟、CD-ROM、或【技術領域】內所公知的任意其它形式的存儲介質中。[0134]對所公開的實施例的上述說明，使本領域專業技術人員能夠實現或使用本發明。對這些實施例的多種修改對本領域的專業技術人員來說將是顯而易見的，本文中所定義的一般原理可以在不脫離本發明的精神或範圍的情況下，在其它實施例中實現。因此，本發明將不會被限制於本文所示的這些實施例，而是要符合與本文所公開的原理和新穎特點相一致的最寬的範圍。
【權利要求】
1.一種實現分布式消息處理的系統，其特徵在於，所述系統包括: 表達式模塊、設置模塊、同步模塊以及至少一個消息處理模塊； 所述表達式模塊中包括有規則項、函數、運算符、表達式、宏變量以及觸發器；所述觸發器用於當所需處理的消息滿足消息處理邏輯表達式時觸發對應的處理操作； 所述設置模塊，用於接收用戶利用所述表達式模塊建立的消息處理邏輯表達式，並將所述消息處理邏輯表達式發送給同步模塊；所述消息處理邏輯表達式由所述規則項、所述函數、所述運算符、所述表達式、所述宏變量中的一種或多種組成； 所述同步模塊，用於將接收到的所述消息處理邏輯表達式實時發送給所述消息處理模塊； 消息處理模塊，用於根據所述同步模塊實時發送的消息處理邏輯表達式，更新自身保存的消息處理邏輯表達式；接收所需處理的消息，讀取最新的消息處理邏輯表達式，按照所述最新的消息處理邏輯表達式對應的消息處理邏輯對所述所需處理的消息進行處理。
2.根據權利要求1所述的系統，其特徵在於，所述系統還包括: 結果輸出模塊，用於將所述消息處理單元輸出的處理結果發送至對應的結果接收裝置，所述結果接收裝置包括消息隊列、簡訊發送系統、郵件發送系統、資料庫以及系統模塊。
3.根據權利要 求1所述的系統，其特徵在於，所述消息處理模塊包括: 更新單元，用於根據所述同步模塊實時發送的消息處理邏輯表達式，更新自身保存的消息處理邏輯表達式； 消息初選單元，用於接收所需處理的消息，讀取最新的消息處理邏輯表達式，對所述所需處理的消息進行是否符合所述最新的消息處理邏輯表達式的篩選，將篩選通過的消息放入消息隊列； 綜合處理單元，用於從所述消息隊列讀取所述篩選通過的消息，讀取最新的消息處理邏輯表達式；按照所述最新的消息處理邏輯表達式對應的消息處理邏輯對所述篩選通過的消息進行處理。
4.根據權利要求3所述的系統，其特徵在於，所述綜合處理單元包括: 第一讀取子單元，用於從所述消息隊列讀取所述篩選通過的消息； 第二讀取子單元，用於讀取最新的消息處理邏輯表達式； 拆分子單元，用於將所述最新的消息處理邏輯表達式拆分為表達式最小單元； 處理子單元，用於按照所述表達式最小單元對應的消息處理邏輯對所述篩選通過的消息進行處理。
5.根據權利要求3所述的系統，其特徵在於，所述消息處理單元還包括: 確認更新單元，用於在更新自身保存的消息處理邏輯表達式之後，判斷更新是否成功，如果是，向所述同步模塊發送同步成功信號，以使所述同步模塊將所述同步成功信號發送給所述設置模塊，如果否，向所述同步模塊發送失敗信號，以使所述同步模塊將所述同步失敗信號發送給所述設置模塊，使所述設置模塊重新將接收用戶利用所述表達式模塊建立的消息處理邏輯表達式發送給所述同步模塊。
6.根據權利要求1所述的系統，其特徵在於，所述表達式模塊還用於: 判斷用戶建立的消息處理邏輯表達式是否滿足表達式合法性規則，如果是，觸發所述設置模塊將所述消息處理邏輯表達式發送給同步模塊，如果否，通過所述設置模塊提示用戶重新建立消息處理邏輯表達式。
7.一種實現分布式消息處理的方法，其特徵在於，所述方法包括: 設置模塊接收用戶利用表達式模塊建立的消息處理邏輯表達式，並將所述消息處理邏輯表達式發送給同步模塊；所述表達式模塊中包括有規則項、函數、運算符、表達式、宏變量以及觸發器；所述消息處理邏輯表達式由所述規則項、所述函數、所述運算符、所述表達式、所述宏變量中的一種或多種組成；所述觸發器用於當所需處理的消息滿足消息處理邏輯表達式時觸發對應的處理操作； 所述同步模塊將接收到的所述消息處理邏輯表達式實時發送給所述消息處理模塊； 所述消息處理模塊根據所述同步模塊實時發送的消息處理邏輯表達式，更新自身保存的消息處理邏輯表達式；接收所需處理的消息，讀取最新的消息處理邏輯表達式，按照所述最新的消息處理邏輯表達式對應的消息處理邏輯對所述所需處理的消息進行處理。
8.根據權利要求7所述的方法，其特徵在於，所述方法還包括: 結果輸出模塊將所述消息處理單元輸出的處理結果發送至對應的結果接收裝置，所述結果接收裝置包括消息隊列、簡訊發送系統、郵件發送系統、資料庫以及系統模塊。
9.根據權利要求7所述的方法，其特徵在於，所述按照所述最新的消息處理邏輯表達式對應的消息處理邏輯對所述所需處理的消息進行處理，包括: 對所述所需處理的消息進行是否符合所述最新的消息處理邏輯表達式的篩選，將篩選通過的消息放入消息隊列； 從所述消息隊列讀取所述篩選通過的消息，按照所述最新的消息處理邏輯表達式對應的消息處理邏輯對所述篩選通過的消息進行處理。
10.根據權利要求9所述的方法，其特徵在於，所述按照所述最新的消息處理邏輯表達式對應的消息處理邏輯對所述篩選通過的消息進行處理，包括: 將所述最新的消息處理邏輯表達式拆分為表達式最小單元； 按照所述表達式最小單元對應的消息處理邏輯對所述篩選通過的消息進行處理。
11.根據權利要求7所述的方法，其特徵在於，在所述消息處理單元更新自身保存的消息處理邏輯表達式之後，所述方法還包括: 所述消息處理單元判斷更新是否成功，如果是，向所述同步模塊發送同步成功信號，以使所述同步模塊將所述同步成功信號發送給所述設置模塊，如果否，向所述同步模塊發送失敗信號，以使所述同步模塊將所述同步失敗信號發送給所述設置模塊，使所述設置模塊重新將接收用戶利用所述表達式模塊建立的消息處理邏輯表達式發送給所述同步模塊。
12.根據權利要求7所述的方法，其特徵在於，所述方法還包括: 所述表達式模塊判斷用戶建立的消息處理邏輯表達式是否滿足表達式合法性規則，如果是，觸發所述設置模塊將所述消息處理邏輯表達式發送給同步模塊，如果否，通過所述設置模塊提示用戶重新建立消息處理邏輯表達式。
【文檔編號】G06F9/50GK103914349SQ201410174706
【公開日】2014年7月9日 申請日期:2014年4月28日 優先權日:2014年4月28日
【發明者】王帥 申請人:北京搜狐新媒體信息技術有限公司