用於傳輸消息的方法和裝置與流程

2023-06-16 00:19:26

本申請涉及計算機技術領域，具體涉及網際網路技術領域，尤其涉及用於傳輸消息的方法和裝置。

背景技術：

MQTT(Message Queuing Telemetry Transport，消息隊列遙測傳輸)是一個已知的即時通訊協議，有可能成為物聯網的重要組成部分。該協議支持所有平臺，幾乎可以把所有聯網物品和外部連接起來，被用來當作傳感器和致動器的通信協議。

現有可用的方案大部分都是單機版本的MQTT服務，單機版本的MQTT服務的缺點是存在單點故障，並且單機的MQTT服務無法支撐更多的用戶連接以及消息發送。目前有些公司是採用的MQTT服務是代理方式實現，只是把MQTT的消息轉發到另外一個消息隊列，這些消息隊列是共享的，同時不支持通配符的路由功能不是一個完整的MQTT服務。同時由於MQTT的訂閱功能不支持分區，這樣會導致訂閱一個主題的消息只能被每個連接獨立處理(每個連接收到的消息都是完全一樣，無法做到每個客戶端收到的消息不一樣的)，就會存在消息擁塞的問題。

技術實現要素：

本申請的目的在於提出一種改進的用於傳輸消息的方法和裝置，來解決以上背景技術部分提到的技術問題。

第一方面，本申請提供了一種用於傳輸消息的方法，其特徵在於，所述方法包括：接收發布客戶端發布的消息；查找訂閱客戶端，其中，所述訂閱客戶端所訂閱的消息的主題和所述發布的消息的主題一致；檢測所查找到的訂閱客戶端上所設置的會話清理標識的值，其中，所述會話清理標識的真值用於標識客戶端斷開連接後、不保存客戶端訂閱的主題所對應的消息，而假值用於標識客戶端斷開連接後、保存客戶端訂閱的主題所對應的消息；響應於所查找到的訂閱客戶端上所設置的會話清理標識的值為假值，將所述消息存儲在分布式系統中，再將存儲在所述分布式系統中的所述消息發送給所查找到的訂閱客戶端。

在一些實施例中，所述方法還包括：響應於所查找到的訂閱客戶端上所設置的會話清理標識的值為真值，將所述消息存儲在所述訂閱客戶端的緩存中，再將存儲在所述緩存中的所述消息發送給所述訂閱客戶端。

在一些實施例中，所述將存儲在所述分布式系統中的所述消息發送給所查找到的訂閱客戶端，包括：通知所述訂閱客戶端從所述分布式系統讀取所述消息；響應於收到所述訂閱客戶端的讀取請求，讀取所述消息並將所讀取的消息發送給所述訂閱客戶端，其中，所述讀取請求包括指示所述訂閱客戶端的已經處理的數據的位置的偏移。

在一些實施例中，所述將所述消息存儲在分布式系統中，包括：記錄所述消息的存儲位置。

在一些實施例中，所述響應於收到所述訂閱客戶端的讀取請求，讀取所述消息並將所讀取的消息發送給所述訂閱客戶端，包括：根據所述消息的存儲位置和所述偏移確定是否需要讀取所述消息；響應於確定需要讀取所述消息，則讀取所述消息並將所讀取的消息發送給所述訂閱客戶端。

在一些實施例中，所述方法還包括：採用單詞查找樹存儲訂閱信息，其中，所述訂閱信息包括：訂閱主題和訂閱所述主題的訂閱客戶端地址。

在一些實施例中，所述分布式系統是kafka系統。

在一些實施例中，所述接收發布客戶端發布的消息還包括：為所述發布的消息按發布時間的先後順序分配消息序列號。

第二方面，本申請提供了一種用於傳輸消息的裝置，其特徵在於，所述裝置包括：接收單元，配置用於接收發布客戶端發布的消息；查找單元，配置用於查找訂閱客戶端，其中，所述訂閱客戶端所訂閱的消息的主題和所述發布的消息的主題一致；檢測所查找到的訂閱客戶端上所設置的會話清理標識的值，其中，所述會話清理標識的真值用於標識客戶端斷開連接後、不保存客戶端訂閱的主題所對應的消息，而假值用於標識客戶端斷開連接後、保存客戶端訂閱的主題所對應的消息；發送單元，配置用於；響應於所查找到的訂閱客戶端上所設置的會話清理標識的值為假值，將所述消息存儲在分布式系統中，再將存儲在所述分布式系統中的所述消息發送給所查找到的訂閱客戶端。

在一些實施例中，所述裝置還包括：緩存單元，配置用於響應於所查找到的訂閱客戶端上所設置的會話清理標識的值為真值，將所述消息存儲在所述訂閱客戶端的緩存中，再將存儲在所述緩存中的所述消息發送給所述訂閱客戶端。

在一些實施例中，所述發送單元進一步配置用於：通知所述訂閱客戶端從所述分布式系統讀取所述消息；響應於收到所述訂閱客戶端的讀取請求，讀取所述消息並將所讀取的消息發送給所述訂閱客戶端，其中，所述讀取請求包括指示所述訂閱客戶端的已經處理的數據的位置的偏移。

在一些實施例中，所述發送單元進一步配置用於：記錄所述消息的存儲位置。

在一些實施例中，所述發送單元進一步配置用於：根據所述消息的存儲位置和所述偏移確定是否需要讀取所述消息；響應於確定需要讀取所述消息，則讀取所述消息並將所讀取的消息發送給所述訂閱客戶端。

在一些實施例中，所述裝置還包括：存儲單元，配置用於採用單詞查找樹存儲訂閱信息，其中，所述訂閱信息包括：訂閱主題和訂閱所述主題的訂閱客戶端地址。

在一些實施例中，所述分布式系統是kafka系統。

在一些實施例中，所述接收單元進一步配置用於：

為所述發布的消息按發布時間的先後順序分配消息序列號。

本申請提供的用於傳輸消息的方法和裝置，通過將發布客戶端發布的消息存儲在分布式系統中再發送給訂閱客戶端，解決了訂閱服務的擁塞問題，保證了MQTT服務的可靠性以及數據備份的需求。

附圖說明

通過閱讀參照以下附圖所作的對非限制性實施例所作的詳細描述，本申請的其它特徵、目的和優點將會變得更明顯：

圖1是本申請可以應用於其中的示例性系統架構圖；

圖2是根據本申請的用於傳輸消息的方法的一個實施例的流程圖；

圖3是根據本申請的用於傳輸消息的方法的一個應用場景的示意圖；

圖4是根據本申請的用於傳輸消息的方法的又一個實施例的流程圖；

圖5是根據本申請的用於傳輸消息的裝置的一個實施例的結構示意圖；

圖6是適於用來實現本申請實施例的伺服器的計算機系統的結構示意圖。

具體實施方式

下面結合附圖和實施例對本申請作進一步的詳細說明。可以理解的是，此處所描述的具體實施例僅僅用於解釋相關發明，而非對該發明的限定。另外還需要說明的是，為了便於描述，附圖中僅示出了與有關發明相關的部分。

需要說明的是，在不衝突的情況下，本申請中的實施例及實施例中的特徵可以相互組合。下面將參考附圖並結合實施例來詳細說明本申請。

圖1示出了可以應用本申請的用於傳輸消息的方法或用於傳輸消息的裝置的實施例的示例性系統架構100。

如圖1所示，系統架構100可以包括客戶端101、102、103、104，網絡105和伺服器106。網絡105用以在客戶端101、102、103、104和伺服器106之間提供通信鏈路的介質。網絡105可以包括各種連接類型，例如有線、無線通信鏈路或者光纖電纜等等。

用戶可以使用客戶端101、102通過網絡105和伺服器106與客戶端103、104交互，以發布或訂閱消息等。客戶端101、102、103、104上安裝有客戶端代理(broker)。

客戶端101、102、103、104可以是具有網絡通信功能的各種電子設備，包括但不限於智慧型手機、冰箱、電視、平板電腦、膝上型便攜計算機和臺式計算機等等。

伺服器106可以是提供各種服務的伺服器，例如對客戶端101、102、103、104發布或訂閱的消息提供轉發功能的後臺雲端伺服器。後臺雲端伺服器可以將從發布客戶端收到的消息轉發給訂閱客戶端。後臺雲端伺服器還可以具備zookeeper功能，其中，zookeeper是一種分布式的、開放源碼的分布式應用程式協調服務。MQTT的broker之間的會話數據都會同步到zookeeper，zookeeper用於協調broker之間各種關係，存儲一些重要的元數據，並且監控broker的狀態，例如與客戶端斷開連接。

需要說明的是，本申請實施例所提供的用於傳輸消息的方法一般由伺服器106執行，相應地，用於傳輸消息的裝置一般設置於伺服器106中。

應該理解，圖1中的終端設備、網絡和伺服器的數目僅僅是示意性的。根據實現需要，可以具有任意數目的終端設備、網絡和伺服器。

繼續參考圖2，示出了根據本申請的用於傳輸消息的方法的一個實施例的流程200。所述的用於傳輸消息的方法，包括以下步驟：

步驟201，接收發布客戶端發布的消息。

在本實施例中，用於傳輸消息的方法運行於其上的電子設備(例如圖1所示的伺服器)可通過有線連接方式或者無線連接方式從用戶利用其進行MQTT服務的客戶端接收發布客戶端發布的消息。

MQTT協議基於發布消息和訂閱主題的原則，多個客戶端連接到一個代理器上並且訂閱他們感興趣的主題，一個客戶端連接到代理器上並且在一個主題上發布消息，其他的客戶端可以訂閱相同的主題並且接收到他們請求的信息。MQTT代理充當了一個簡單的共同的接口供其他設備去連接。主題是由斜槓(/)來劃分層次，就跟文件系統一樣。例如，各種的計算機會發布他們的硬體驅動的溫度信息在如下的主題上，他們計算機和硬體的名字被適當的替換：

「sensors/COMPUTER_NAME/temperature/HARDDRIVE_NAME」。

步驟202，查找訂閱客戶端。

在本實施例中，需要找到一個訂閱的消息的主題和所述發布的消息的主題一致的訂閱客戶端。在MQTT服務中，客戶端可以通過創建一個訂閱來接收消息。訂閱必須要指明主題，並且只能接收到指定主題的消息。根據發布的消息的主題可以找到與該主題的名稱匹配訂閱客戶端。可以在查找主題中使用通配符，MQTT容許「+」和「#」兩個通配符。作為一層的通配符。它可以如下面操作來訂閱上文中提到的所有計算機和硬體驅動的信息：「+sensors/+/temperature/+」。

另外，如主題「a/b/c/d」，以下給出的訂閱是符合的：「+a/b/c/d」，「++/b/c/d」，「+a/+/c/d」，「+a/+/+/d」，「++/+/+/+」。

以下的示例是不符合的：「+a/b/c」，「+b/+/c/d」，「++/+/+」。

「#」可以作為所有剩下的層的通配符。這意味著它必須在主題的最後一個字符。對於「a/b/c/d」主題，以下的示例訂閱是符合的：「a/b/c/d+#」，「+a/#」，「+a/b/#」，「+a/b/c/#」，「++/b/c/#」。

零長度的主題層是有效的，但是會導致一些不明顯的行為。例如，主題「a//topic」會被匹配到一個相反的「a/+/topic」主題上。零長度的主題層可以存在在一個主題字符串的開頭和結尾，所以，「/a/topic」會匹配到「+/a/topic」、「#」或者「/#」主題。「a/topic/」主題會匹配到其他相反的「a/topic/+」或「a/topic/#」主題。

在本實施例的一些可選的實現方式中，採用單詞查找樹存儲訂閱信息，其中，所述訂閱信息包括：訂閱主題和訂閱所述主題的訂閱客戶端地址。MQTT數據信息的路由最大難點在於主題支持通配符，目前支持的通配符包括#和+，#表示匹配所有，+表示匹配特定層級的所有主題，為了解決這些問題，MQTT服務採用tries tree(單詞查找樹或者前綴樹，是一種用於快速檢索的多叉樹結構)的方式保存訂閱的主題信息，tries tree可以提供高效的查詢速度，broker之間的數據傳輸也是採用tries tree的方式保存其他broker需要訂閱的主題，以及需要轉發消息的broker名稱，這樣相對於廣播的方式顯著降低了broker之間消息傳遞數目，有效減少帶寬使用。如果brokerA有客戶端訂閱一個新的主題，通過zookeeper將這個消息推送給其他broker，其他broker會將對應的主題添加到tries tree上，如果有客戶端取消訂閱某個主題，也會通過zookeeper來通知其他broker刪除對應的主題。

步驟203，檢測所查找到的訂閱客戶端上所設置的會話清理標識的值。

在本實施例中，檢測所查找到的訂閱客戶端上所設置的會話清理(clean session)標識的值。其中，所述會話清理標識的真值用於標識客戶端斷開連接後、不保存客戶端訂閱的主題所對應的消息，而假值用於標識客戶端斷開連接後、保存客戶端訂閱的主題所對應的消息。如果會話清理標識的值為真，則在客戶端主動斷開，或者超過預定的時間沒有收到心跳時，清除屬於該客戶端的會話信息。

步驟204，將消息存儲在分布式系統中，再將存儲在分布式系統中的消息發送給所查找到的訂閱客戶端。

在本實施例中，如果訂閱客戶端上所設置的會話清理標識的值為假值，那麼不管該客戶端是否在線，都需要將所有應該發送給這個客戶端的數據保存在分布式系統中。分布式系統(distributed system)是建立在網絡之上的軟體系統。分布式系統具有高度的內聚性和透明性。內聚性是指每一個資料庫分布節點高度自治，有本地的資料庫管理系統。透明性是指每一個資料庫分布節點對用戶的應用來說都是透明的，看不出是本地還是遠程。

在本實施例的一些可選的實現方式中，上述分布式系統是kafka系統(一種已知的、高吞吐量的分布式發布訂閱消息系統，它可以處理消費者規模的網站中的所有動作流數據)。對於某些特定的應用，比如消息推送方式，允許採用redis(一個開源的使用ANSI C語言編寫、支持網絡、可基於內存亦可持久化的日誌型、Key-Value資料庫)，或者Hbase(一個分布式的、面向列的開源資料庫)存儲消息，這種場景是處理消息數目比較少，用來通訊的場景，比如聊天，消息推送。

在本實施例中，將存儲在分布式系統中的消息發送給所查找到的訂閱客戶端。分布式系統中備份了待發送的消息，如果客戶端斷開或者超過預定的時間沒有收到心跳時，可從分布式系統中讀取待發送的消息發給訂閱客戶端。

在本實施例的一些可選的實現方式中，響應於所查找到的訂閱客戶端上所設置的會話清理標識的值為真值，將所述消息存儲在所述訂閱客戶端的緩存中，再將存儲在所述緩存中的所述消息發送給所述訂閱客戶端。如果該客戶端斷開，或者超過預定的時間沒有收到心跳，則清除屬於該客戶端的會話信息，包括緩存裡面還沒有被發送出去的消息。

在本實施例的一些可選的實現方式中，上述將存儲在所述分布式系統中的所述消息發送給所查找到的訂閱客戶端，包括：通知所述訂閱客戶端從所述分布式系統讀取所述消息；響應於收到所述訂閱客戶端的讀取請求，讀取所述消息並將所讀取的消息發送給所述訂閱客戶端，其中，所述讀取請求包括指示所述訂閱客戶端的已經處理的數據的位置的偏移(offset)，「offset」記錄每條日誌的偏移量。

在本實施例的一些可選的實現方式中，記錄所述消息的存儲位置。每次寫入kafka的數據都會返回一個指示消息的存儲位置的偏移。

在本實施例的一些可選的實現方式中，上述響應於收到所述訂閱客戶端的讀取請求，讀取所述消息並將所讀取的消息發送給所述訂閱客戶端，包括：根據所述消息的存儲位置和所述偏移確定是否需要讀取所述消息；響應於確定需要讀取所述消息，則讀取所述消息並將所讀取的消息發送給所述訂閱客戶端。可以通過將發送給這個客戶端的數據保存到kafka的主題，每個客戶端都會分配一個唯一的kafka主題，然後啟動一個任務去從這個唯一的kafka主題讀取數據。由於kafka是採用拉(pull)的方式消費(consume)數據，在本申請中採用事件驅動的方式實現，每次寫入kafka的數據都會返回一個指示消息的存儲位置的偏移，同時會記錄訂閱客戶端消費數據的位置偏移用於在讀取請求中發送，如果kafka最上面的消息的存儲位置的偏移大於訂閱客戶端消費數據的位置偏移，就會有一個任務不停的去讀取數據，如果相等之後就停止，如果有新的數據，就會重新啟動這個任務，kafka讀寫數據的操作都是O(1)的性能，所以即使有很多數據的寫入也不會存在任何性能的問題。

將數據優先寫入kafka後，如果當前訂閱客戶端的消費數據(即已經處理的數據)的位置偏移和kafka中消息的存儲位置的偏移是相等的話，那麼意味當前設備已經消費了kafka隊列裡面的所有數據，所以現在發送給kafka的數據就是最上面的數據了，可以直接發送給對應的客戶端，可以極大的降低延遲。

如果當一個broker出現異常，那麼就無法保證數據發送到對應的客戶端了，特別是會話清理標識的值為假值的客戶端，為了解決這個問題，可以將會話清理標識的值為假值的所有客戶端的信息同步到其他機器，這樣可以通過發送消息的broker來直接將數據寫入到kafka的主題。只要某個broker和zookeeper失去連接，就認為該broker出現異常情況，zookeeper是維護會話信息的中央，如果失去連接，那麼這個broker上面的狀態信息就不準確，該broker會主動斷開所有和他保持連接的客戶端。對於會話清理標識的值為假值的客戶端會連接到其他機器，然後重新構建會話信息，從kafka消費數據，整個過程中沒有丟失任何數據，也不會阻塞其他客戶端的消息發送。

如果一個broker和其他broker及zookeeper都失去連接，那麼這個broker的狀態就不是最新的，無法和其他broker保持一致，我們可以認為該broker是出在一種不正常的狀態，他會自動斷開和其他節點的連接，同時也會斷開和MQTT客戶端的連接。

如果一個broker和其他broker失去連接，但和zookeeper保持連接，則自動關閉成員數最少的分區組，這樣斷開連接的客戶端會重新和最新的網絡保持連接。

繼續參見圖3，圖3是根據本實施例的用於傳輸消息的方法的應用場景的一個示意圖。在圖3的應用場景中，將設備發布的消息通過MQTT代理寫入kafka，然後從kafka訂閱消息，由於kafka是支持分區(partition)的，所以我們可以採用多臺機器來處理不同分區的消息。這樣的話，當消息達到一定數量之後，客戶端不會出現溢出的情況，避免消息丟失，或者磁碟寫滿。

本申請的上述實施例提供的方法，不僅實現了一個完整的分布式MQTT服務，同時還解決MQTT在物聯網場景中的一些限制，比如MQTT的訂閱功能是存在單點不可擴展的問題，比如100萬個客戶端同時往一個主題發消息，那麼訂閱這個主題的客戶端只和MQTT服務有一個連接，這樣的話，單個連接能傳輸的數據上限受到帶寬以及單個機器的處理能力的限制無法橫向擴展，維持，本申請提出將消息存儲到分布式系統中，比如Kafka，每個機器直接往kafka發送消息，Kafka本身是一個可以分布式系統，支持分區，這樣就可以很好的解決擴展性的問題，然後客戶端從kafka訂閱消息處理消息。

進一步參考圖4，其示出了用於傳輸消息的方法的又一個實施例的流程400。該用於傳輸消息的方法的流程400，包括以下步驟：

步驟401，接收發布客戶端發布的消息。

該步驟與步驟201基本相同，因此不再贅述。

步驟402，為發布的消息按發布時間的先後順序分配消息序列號。

在本實施例中，為發布的消息按發布時間的先後順序分配消息序列號，以保證任何客戶端收到的消息順序都是一致的，支持同一客戶端發布消息順序性。保證每個訂閱客戶端收到來自同一個broker上發送的同一個主題的消息的有序性。並且不同broker發送消息順序取決於他們發送到這個broker的時間決定的。

為此我們的設計思想是每個broker都是一個獨立的主伺服器，這個主伺服器為發送給該主伺服器的消息分配有序的消息序列號，然後該主伺服器會將這些有序的消息轉發給其他broker，這樣我們就可以解決單點的擁擠問題，同時也保證同一個設備發送消息的有序性，現實中我們對消息的有序性大部分都是基於一個設備發送消息的有序性來定義的。

步驟403，查找訂閱客戶端。

步驟404，檢測所查找到的訂閱客戶端上所設置的會話清理標識的值。

步驟405，將消息存儲在分布式系統中，再將存儲在分布式系統中的消息發送給所查找到的訂閱客戶端。

步驟403-405與步驟202-204基本相同，因此不再贅述。

從圖4中可以看出，與圖2對應的實施例相比，本實施例中的用於傳輸消息的方法的流程400突出了確保訂閱客戶端收到的消息順序一致性的步驟。未採用統一的計算單元來為每一個消息生成唯一的序列號，因此不會導致大量的請求會被阻塞，也不會出現所有發給同一個主題的數據都會被發往一個特定的broker來分配序列號的情況，這樣就會減少網絡帶寬的消耗，提高系統的可靠性。

進一步參考圖5，作為對上述各圖所示方法的實現，本申請提供了一種用於傳輸消息的裝置的一個實施例，該裝置實施例與圖2所示的方法實施例相對應，該裝置具體可以應用於各種電子設備中。

如圖5所示，本實施例所述的用於傳輸消息的裝置500包括：接收單元501、查找單元502、檢測單元503和發送單元504。其中，接收單元501配置用於接收發布客戶端發布的消息；查找單元502配置用於查找訂閱客戶端，其中，所述訂閱客戶端所訂閱的消息的主題和所述發布的消息的主題一致；檢測單元503配置用於檢測所查找到的訂閱客戶端上所設置的會話清理標識的值，其中，所述會話清理標識的真值用於標識客戶端斷開連接後、不保存客戶端訂閱的主題所對應的消息，而假值用於標識客戶端斷開連接後、保存客戶端訂閱的主題所對應的消息；發送單元504配置用於響應於所查找到的訂閱客戶端上所設置的會話清理標識的值為假值，將所述消息存儲在分布式系統中，再將存儲在所述分布式系統中的所述消息發送給所查找到的訂閱客戶端。

在本實施例中，查找單元502根據接收單元501接收的消息查找訂閱客戶端。

在本實施例中，檢測單元503根據查找單元502查找到的訂閱客戶端的會話清理標識的值確定是否將接收到的消息發給發送單元504處理。

在本實施例的一些可選的實現方式中，裝置500還包括：緩存單元，配置用於響應於所查找到的訂閱客戶端上所設置的會話清理標識的值為真值，將所述消息存儲在所述訂閱客戶端的緩存中，再將存儲在所述緩存中的所述消息發送給所述訂閱客戶端。

在本實施例的一些可選的實現方式中，發送單元504進一步配置用於：通知所述訂閱客戶端從所述分布式系統讀取所述消息；響應於收到所述訂閱客戶端的讀取請求，讀取所述消息並將所讀取的消息發送給所述訂閱客戶端，其中，所述讀取請求包括指示所述訂閱客戶端的已經處理的數據的位置的偏移。

在本實施例的一些可選的實現方式中，發送單元504進一步配置用於：記錄所述消息的存儲位置。

在本實施例的一些可選的實現方式中，發送單元504進一步配置用於：根據所述消息的存儲位置和所述偏移確定是否需要讀取所述消息；響應於確定需要讀取所述消息，則讀取所述消息並將所讀取的消息發送給所述訂閱客戶端。

在本實施例的一些可選的實現方式中，裝置500還包括：存儲單元，配置用於採用單詞查找樹存儲訂閱信息，其中，所述訂閱信息包括：訂閱主題和訂閱所述主題的訂閱客戶端地址。

在本實施例的一些可選的實現方式中，所述分布式系統是kafka系統。

在本實施例的一些可選的實現方式中，接收單元501進一步配置用於：為所述發布的消息按發布時間的先後順序分配消息序列號。

下面參考圖6，其示出了適於用來實現本申請實施例的伺服器的計算機系統600的結構示意圖。

如圖6所示，計算機系統600包括中央處理單元(CPU)601，其可以根據存儲在只讀存儲器(ROM)602中的程序或者從存儲部分608加載到隨機訪問存儲器(RAM)603中的程序而執行各種適當的動作和處理。在RAM 603中，還存儲有系統600操作所需的各種程序和數據。CPU 601、ROM 602以及RAM 603通過總線604彼此相連。輸入/輸出(I/O)接口605也連接至總線604。

以下部件連接至I/O接口605：包括鍵盤、滑鼠等的輸入部分606；包括諸如液晶顯示器(LCD)等以及揚聲器等的輸出部分607；包括硬碟等的存儲部分608；以及包括諸如LAN卡、數據機等的網絡接口卡的通信部分609。通信部分609經由諸如網際網路的網絡執行通信處理。驅動器610也根據需要連接至I/O接口605。可拆卸介質611，諸如磁碟、光碟、磁光碟、半導體存儲器等等，根據需要安裝在驅動器610上，以便於從其上讀出的電腦程式根據需要被安裝入存儲部分608。

特別地，根據本公開的實施例，上文參考流程圖描述的過程可以被實現為計算機軟體程序。例如，本公開的實施例包括一種電腦程式產品，其包括有形地包含在機器可讀介質上的電腦程式，所述電腦程式包含用於執行流程圖所示的方法的程序代碼。在這樣的實施例中，該電腦程式可以通過通信部分609從網絡上被下載和安裝，和/或從可拆卸介質611被安裝。在該電腦程式被中央處理單元(CPU)601執行時，執行本申請的方法中限定的上述功能。

附圖中的流程圖和框圖，圖示了按照本申請各種實施例的系統、方法和電腦程式產品的可能實現的體系架構、功能和操作。在這點上，流程圖或框圖中的每個方框可以代表一個模塊、程序段、或代碼的一部分，所述模塊、程序段、或代碼的一部分包含一個或多個用於實現規定的邏輯功能的可執行指令。也應當注意，在有些作為替換的實現中，方框中所標註的功能也可以以不同於附圖中所標註的順序發生。例如，兩個接連地表示的方框實際上可以基本並行地執行，它們有時也可以按相反的順序執行，這依所涉及的功能而定。也要注意的是，框圖和/或流程圖中的每個方框、以及框圖和/或流程圖中的方框的組合，可以用執行規定的功能或操作的專用的基於硬體的系統來實現，或者可以用專用硬體與計算機指令的組合來實現。

描述於本申請實施例中所涉及到的單元可以通過軟體的方式實現，也可以通過硬體的方式來實現。所描述的單元也可以設置在處理器中，例如，可以描述為：一種處理器包括接收單元、查找單元、檢測單元和發送單元。其中，這些單元的名稱在某種情況下並不構成對該單元本身的限定，例如，接收單元還可以被描述為「接收發布客戶端發布的消息的單元」。

作為另一方面，本申請還提供了一種非易失性計算機存儲介質，該非易失性計算機存儲介質可以是上述實施例中所述裝置中所包含的非易失性計算機存儲介質；也可以是單獨存在，未裝配入客戶端中的非易失性計算機存儲介質。上述非易失性計算機存儲介質存儲有一個或者多個程序，當所述一個或者多個程序被一個設備執行時，使得所述設備：接收發布客戶端發布的消息；查找訂閱客戶端，其中，所述訂閱客戶端所訂閱的消息的主題和所述發布的消息的主題一致；檢測所查找到的訂閱客戶端上所設置的會話清理標識的值，其中，所述會話清理標識的真值用於標識客戶端斷開連接後、不保存客戶端訂閱的主題所對應的消息，而假值用於標識客戶端斷開連接後、保存客戶端訂閱的主題所對應的消息；響應於所查找到的訂閱客戶端上所設置的會話清理標識的值為假值，將所述消息存儲在分布式系統中，再將存儲在所述分布式系統中的所述消息發送給所查找到的訂閱客戶端。

以上描述僅為本申請的較佳實施例以及對所運用技術原理的說明。本領域技術人員應當理解，本申請中所涉及的發明範圍，並不限於上述技術特徵的特定組合而成的技術方案，同時也應涵蓋在不脫離所述發明構思的情況下，由上述技術特徵或其等同特徵進行任意組合而形成的其它技術方案。例如上述特徵與本申請中公開的(但不限於)具有類似功能的技術特徵進行互相替換而形成的技術方案。