基于多線程worker的高可靠數(shù)據(jù)分發(fā)服務(wù)模型

鄭嘉成，張亞楠

摘要：針對(duì)傳統(tǒng)數(shù)據(jù)分發(fā)服務(wù)中，由慢訂閱者與慢連接問(wèn)題引起的高丟包率問(wèn)題，提出多線程數(shù)據(jù)分發(fā)模型，降低丟包率。采用多線程技術(shù)，基于actor模型，引入多線程worker實(shí)現(xiàn)訂閱者并行接收數(shù)據(jù)，改進(jìn)訂閱者無(wú)狀態(tài)消息返回發(fā)布者可能引起的慢連接問(wèn)題，建立了基于多線程worker的高可靠數(shù)據(jù)分發(fā)服務(wù)模型，有效降低了數(shù)據(jù)傳輸過(guò)程中的丟包率。不僅通過(guò)與ZeroMQ（zero message queue，零消息隊(duì)列）進(jìn)行對(duì)比試驗(yàn)，表明該模型在接收不同長(zhǎng)度報(bào)文情況下，能有效降低丟包率。而且通過(guò)對(duì)模型進(jìn)行壓力測(cè)試，驗(yàn)證其降低丟包率的有效性。

關(guān)鍵詞：數(shù)據(jù)分發(fā)服務(wù);發(fā)布訂閱;丟包率;多線程;慢訂閱者

DOI：10.15938/j.jhust.2022.01.003

中圖分類(lèi)號(hào)： TP393? ? ? ?文獻(xiàn)標(biāo)志碼： A? ? ? ? ?文章編號(hào)： 1007-2683（2022）01-0014-09

Highly Reliable Data Distribution Service Model Based

on Multithreaded Worker

ZHENG Jiacheng，ZHANG Yanan

（1School of Computer Science and Technology， Harbin University of Science and Technology， Harbin 150080，China;

2School of Software and Microelectronics， Harbin University of Science and Technology， Harbin 150080，China）

Abstract：A multithreaded data distribution model is proposed to reduce the packet loss rate caused by slow subscribers and slow connections in traditional data distribution services. Based on the actor model， the multithreaded worker is introduced to enable subscribers to receive data in parallel and to improve the slow connection problem caused by lack of status message from subscribers to publishers， and a highly reliable data distribution service model based on the multithreaded worker is established to reduce the packet loss rate during data transmission effectively. A contrast experiment ZeroMQ （zero message queue） shows that the model can effectively reduce the packet loss rate when receiving messages of different lengths. The effectiveness of the model in reducing packet loss is also verified through stress tests.

Keywords：data distribution service; publish and subscribe; packet loss rate; multithreading; slow subscriber

0引言

數(shù)據(jù)分發(fā)服務(wù)（data distribution service，DDS）是一種網(wǎng)絡(luò)數(shù)據(jù)通信的核心技術(shù)，滿(mǎn)足分布式實(shí)時(shí)應(yīng)用的通信需求，促使數(shù)據(jù)實(shí)時(shí)、高效地分發(fā)[1]。目前DDS數(shù)據(jù)分發(fā)服務(wù)在仿真領(lǐng)域的應(yīng)用十分廣泛，例如戰(zhàn)場(chǎng)指揮[2-4]，雷達(dá)仿真與聲納仿真[5-9]，在軍事領(lǐng)域尤其突出。與此同時(shí)，民用領(lǐng)域也逐漸開(kāi)始采用數(shù)據(jù)分發(fā)服務(wù)，例如船舶管理[10-11]，飛行器協(xié)同[12]以及電力系統(tǒng)管理[13-14]。隨著應(yīng)用場(chǎng)景的不斷擴(kuò)展，許多實(shí)際工作都因數(shù)據(jù)傳輸丟包率高，仿真效果不佳而停滯不前，例如強(qiáng)化學(xué)習(xí)對(duì)于仿真精度以及傳輸效率要求都比較高，高丟包對(duì)于多智能體間數(shù)據(jù)交換會(huì)產(chǎn)生極大的負(fù)面影響，在很大程度上降低模型評(píng)估的正確性，目前許多研究都致力于降低數(shù)據(jù)分發(fā)服務(wù)的高丟包問(wèn)題。

目前對(duì)于定位丟包位置[15]和丟包率建模[16]的技術(shù)已經(jīng)存在，但對(duì)于丟包問(wèn)題本身還難以有效解決。通過(guò)上行增強(qiáng)技術(shù)[17]優(yōu)化降低上行丟包率的方法，在適用場(chǎng)景上略顯不足。利用關(guān)聯(lián)分析法對(duì)MR數(shù)據(jù)的深度挖掘[18]，借此實(shí)現(xiàn)精準(zhǔn)優(yōu)化，但該方法僅能從宏觀上降低丟包率，微觀上的丟包問(wèn)題并沒(méi)有得到解決。調(diào)整發(fā)送接收隊(duì)列長(zhǎng)度以及引入心跳機(jī)制的處理方法[19-20]，在面對(duì)不同節(jié)點(diǎn)硬件時(shí)，存在通用性較低的問(wèn)題。

本文通過(guò)改進(jìn)零消息隊(duì)列（zero message queue，ZMQ）發(fā)布訂閱模式，引入actor模型的思想，提出一種基于多線程worker的訂閱者，實(shí)現(xiàn)多線程協(xié)同數(shù)據(jù)接收，引入輪詢(xún)池任務(wù)調(diào)度與收集器概念，改進(jìn)傳統(tǒng)發(fā)布訂閱模式，建立一種高可靠的數(shù)據(jù)分發(fā)服務(wù)模型，有效降低數(shù)據(jù)分發(fā)過(guò)程中的丟包率。相比于從宏觀上降低丟包率，本模型從微觀上分析并解決了引起丟包率高的慢訂閱者問(wèn)題和慢連接問(wèn)題;且本模型通用性較高，面對(duì)不同節(jié)點(diǎn)條件都能得到較好的應(yīng)用。

1問(wèn)題提出與分析

1.1高丟包率

使用ZMQ發(fā)布訂閱模式進(jìn)行數(shù)據(jù)發(fā)布時(shí)，發(fā)布者節(jié)點(diǎn)可能會(huì)因慢訂閱者或慢連接問(wèn)題造成高丟包率。因此，如何解決傳統(tǒng)發(fā)布訂閱模式下丟包率高的問(wèn)題，是本文關(guān)注的重點(diǎn)。

1.2慢訂閱者問(wèn)題

當(dāng)發(fā)布者與訂閱者雙方已經(jīng)建立連接，此時(shí)主要的丟包原因在于訂閱者接收?qǐng)?bào)文速度無(wú)法跟上發(fā)布者發(fā)送報(bào)文速度。訂閱者接收?qǐng)?bào)文后，報(bào)文需要經(jīng)過(guò)業(yè)務(wù)流處理，當(dāng)業(yè)務(wù)流線程時(shí)間開(kāi)銷(xiāo)較大時(shí)，會(huì)造成訂閱者進(jìn)程中recv線程阻塞，阻塞結(jié)束后再次偵聽(tīng)發(fā)布者報(bào)文。ZMQ底層采用邊緣觸發(fā)機(jī)制，當(dāng)訂閱業(yè)務(wù)（SUB）流線程阻塞后多個(gè)報(bào)文到達(dá)時(shí)，觸發(fā)器只會(huì)觸發(fā)一次，報(bào)文中僅有實(shí)際觸發(fā)的報(bào)文會(huì)被讀取，未觸發(fā)的報(bào)文則被丟棄。圖1為SUB進(jìn)程示意圖。

解決慢訂閱者有兩種方法。一種是將報(bào)文貯存在發(fā)布者緩沖區(qū)中，但在高吞吐量應(yīng)用場(chǎng)景下，發(fā)布者緩沖區(qū)由于堆積數(shù)據(jù)會(huì)導(dǎo)致內(nèi)存溢出，尤其是面對(duì)多個(gè)訂閱者的情況，磁盤(pán)緩沖更容易溢出。另外一種是將報(bào)文在訂閱者緩沖區(qū)中貯存，這也是ZMQ的默認(rèn)行為。該方法在瞬間消息量很大的應(yīng)用場(chǎng)景下十分適用，訂閱者可能只是暫時(shí)跟不上發(fā)布者速度，最終會(huì)趕上進(jìn)度，然而，在連續(xù)接收大消息量的場(chǎng)景下，訂閱者會(huì)因緩沖區(qū)溢出而崩潰，該方法并沒(méi)有解決慢訂閱者問(wèn)題本身。

1.3慢連接問(wèn)題

訂閱者與發(fā)布者間進(jìn)行數(shù)據(jù)傳輸，需要建立TCP連接，訂閱者通過(guò)硬編碼端點(diǎn)字符串向發(fā)布者建立連接，這一過(guò)程涉及TCP三次握手。對(duì)于訂閱者來(lái)說(shuō)，該握手過(guò)程通常會(huì)花費(fèi)數(shù)毫秒時(shí)間;對(duì)于發(fā)布者來(lái)說(shuō)，啟動(dòng)發(fā)布者接收開(kāi)始命令后就開(kāi)始將原始數(shù)據(jù)封裝成報(bào)文進(jìn)行發(fā)送。PUB進(jìn)程與SUB進(jìn)程并行執(zhí)行，ZMQ在后臺(tái)進(jìn)行異步I/O傳輸，由于發(fā)布者發(fā)送消息在時(shí)間上領(lǐng)先于訂閱者建立連接，因此對(duì)于發(fā)布者來(lái)說(shuō)，此時(shí)會(huì)形成無(wú)訂閱者接收消息的情況。zmqPUB套接字出于減輕系統(tǒng)負(fù)載的目的，會(huì)在無(wú)接收者期間自動(dòng)丟棄所有報(bào)文，基于該

處理機(jī)制，發(fā)布者在訂閱者建立連接期間發(fā)送的報(bào)文會(huì)全部丟失。即使嘗試在啟動(dòng)訂閱者之后再啟動(dòng)發(fā)布者，也無(wú)法規(guī)避TCP三次握手所需要的數(shù)毫秒，最終還是會(huì)導(dǎo)致發(fā)布者丟棄在建立連接之前發(fā)送的所有報(bào)文。圖2為慢連接問(wèn)題示意圖。

針對(duì)慢連接問(wèn)題，有兩種解決方法。方法一是發(fā)布者延遲發(fā)布，在發(fā)布者啟動(dòng)后阻塞其PUB進(jìn)程，SUB進(jìn)程在PUB進(jìn)程阻塞期間建立連接，實(shí)現(xiàn)進(jìn)程同步。該方法的不足之處在于無(wú)法準(zhǔn)確設(shè)置PUB進(jìn)程阻塞時(shí)間長(zhǎng)短，阻塞時(shí)間設(shè)置過(guò)長(zhǎng)會(huì)造成不必要的時(shí)間開(kāi)銷(xiāo)，阻塞時(shí)間設(shè)置過(guò)短則不能完全同步收發(fā)雙方進(jìn)程，無(wú)法徹底解決慢連接問(wèn)題。方法二將發(fā)布者數(shù)據(jù)流看作無(wú)限流，沒(méi)有所謂的起點(diǎn)與終點(diǎn)，從訂閱者建立連接后接收到的第一條報(bào)文作為起點(diǎn)，忽略建立連接時(shí)被丟棄的報(bào)文。該方法并沒(méi)有解決慢連接問(wèn)題，且適用的場(chǎng)景較少。

2設(shè)計(jì)與實(shí)現(xiàn)

傳統(tǒng)訂閱者消息接收線程與業(yè)務(wù)流線程串行執(zhí)行，引起慢訂閱者問(wèn)題;本模型訂閱者通過(guò)多線程worker實(shí)現(xiàn)，多個(gè)worker間并行執(zhí)行消息接收線程與業(yè)務(wù)流線程。通過(guò)共享內(nèi)存的inproc協(xié)議作為worker與訂閱者間通信的橋梁，以提高模型執(zhí)行效率。多線程worker訂閱者將傳統(tǒng)模型訂閱者緩沖區(qū)在邏輯上劃分為若干分區(qū)，通過(guò)多個(gè)worker線程分別讀取邏輯上劃分的分區(qū)，達(dá)到優(yōu)于單一線程處理整個(gè)緩沖區(qū)的效果，提高訂閱者消息接收能力。

在此基礎(chǔ)上，為了實(shí)現(xiàn)worker間負(fù)載均衡，本模型訂閱者中新增并實(shí)現(xiàn)了輪詢(xún)池作為worker狀態(tài)輪詢(xún)接口，負(fù)責(zé)有序管理多個(gè)緩沖分區(qū)，通過(guò)雙向異步偵聽(tīng)和輪詢(xún)算法實(shí)現(xiàn)worker負(fù)載均衡，防止同一個(gè)worker處理過(guò)多消息，提高發(fā)布訂閱模型并發(fā)性;另外，在傳統(tǒng)訂閱者的基礎(chǔ)上，設(shè)計(jì)并實(shí)現(xiàn)用于消息匯總的訂閱者收集器，不同worker將各自處理的消息結(jié)果進(jìn)行異步匯總，通過(guò)公平隊(duì)列這樣的無(wú)鎖算法實(shí)現(xiàn)訂閱者消息接收能力的進(jìn)一步提升。

傳統(tǒng)發(fā)布訂閱模型中，發(fā)布者與訂閱者之間只能進(jìn)行單向通信，訂閱者對(duì)發(fā)布者而言不可見(jiàn)，引起慢連接問(wèn)題。本模型通過(guò)架設(shè)雙向通路實(shí)現(xiàn)發(fā)布者與訂閱者間雙向通信，基于雙向通信達(dá)到發(fā)布者與訂閱者間狀態(tài)同步，控制發(fā)布者進(jìn)程與訂閱者進(jìn)程同步執(zhí)行，防止慢連接問(wèn)題引起的丟包現(xiàn)象。

2.1模型設(shè)計(jì)

基于多線程worker的高可靠數(shù)據(jù)分發(fā)服務(wù)模型，主要包含3個(gè)部分的內(nèi)容：訂閱者與worker之間關(guān)系結(jié)構(gòu)的設(shè)計(jì)、worker之間的消息分配設(shè)計(jì)、worker資源釋放時(shí)機(jī)的控制。①訂閱者與worker之間關(guān)系結(jié)構(gòu)的設(shè)計(jì)，該設(shè)計(jì)涉及訂閱者與worker間是否解耦，不同worker間能否相互通信，對(duì)于發(fā)揮模型并發(fā)性來(lái)說(shuō)非常關(guān)鍵。訂閱者啟動(dòng)時(shí)，通過(guò)其上下文subscriber_context創(chuàng)建多個(gè)worker實(shí)例，創(chuàng)建的worker為訂閱者服務(wù)，由此構(gòu)成一個(gè)能夠多線程并行處理消息的訂閱者。設(shè)計(jì)良好消息分配機(jī)制的worker可以有序管理訂閱者緩沖區(qū)的不同分區(qū)，對(duì)于充分發(fā)揮模型的的并發(fā)性，降低worker之間耦合性，都具有重要意義。②worker之間的消息分配設(shè)計(jì)，由于worker是消息驅(qū)動(dòng)的，為同一訂閱者服務(wù)的多個(gè)worker之間相互解耦，不直接進(jìn)行消息通信，取而代之的是由訂閱者創(chuàng)建輪詢(xún)池為worker進(jìn)行消息分配。此外，消息分配是否合理決定了worker任務(wù)分發(fā)是否高效，基于輪詢(xún)池的負(fù)載均衡分配方式與數(shù)據(jù)分發(fā)服務(wù)業(yè)務(wù)場(chǎng)景相契合，采用輪詢(xún)算法進(jìn)行消息分配的輪詢(xún)池有利于提高worker接收消息利用率，有利于充分發(fā)揮多線程編程的并發(fā)效率。③控制worker資源釋放時(shí)機(jī)，多線程worker數(shù)據(jù)分發(fā)服務(wù)模型中，訂閱者可以動(dòng)態(tài)控制worker資源釋放時(shí)機(jī)，這一點(diǎn)對(duì)于數(shù)據(jù)分發(fā)系統(tǒng)來(lái)說(shuō)至關(guān)重要。如果將數(shù)據(jù)分發(fā)到某個(gè)worker處進(jìn)行接收，那么系統(tǒng)必須等待該消息的接收結(jié)果，該等待過(guò)程包括消息的接收、處理和匯總。當(dāng)不再需要worker接收消息時(shí)，訂閱者可以適時(shí)釋放worker資源，達(dá)到節(jié)約系統(tǒng)資源的目的，釋放worker資源時(shí)采用Python語(yǔ)言經(jīng)典的資源回收機(jī)制，當(dāng)worker引用計(jì)數(shù)為零時(shí)，會(huì)被視為垃圾進(jìn)行處理，回收worker所占用的內(nèi)存空間，由于該操作涉及較大的危險(xiǎn)性，因此采用Cpython解釋器提供的自動(dòng)垃圾回收機(jī)制，防止由于內(nèi)存回收造成可能的內(nèi)存溢出問(wèn)題。圖3是多線程worker數(shù)據(jù)分發(fā)模型消息架構(gòu)。

多線程worker數(shù)據(jù)分發(fā)服務(wù)模型的消息數(shù)據(jù)流向如圖3所示，分為3個(gè)部分：發(fā)布者、worker和收集器，分別表示數(shù)據(jù)源端的發(fā)布者，數(shù)據(jù)實(shí)際接收的worker端和數(shù)據(jù)處理結(jié)果匯總的收集器端，發(fā)布者、worker和收集器都有具體名稱(chēng)、地址以及端口號(hào)。對(duì)于需要跨機(jī)器通信的情況，可以利用硬編碼端點(diǎn)字符串，即IP地址和端口號(hào)，對(duì)于用戶(hù)來(lái)說(shuō)，發(fā)布端作為相對(duì)固定的端點(diǎn)，其地址和端口是透明的，通過(guò)連接發(fā)布端地址和端口，與應(yīng)用程序進(jìn)行消息通信。其中消息數(shù)據(jù)在各部分的具體過(guò)程如下，圖4為訂閱者創(chuàng)建worker時(shí)序圖。步驟一：上層應(yīng)用產(chǎn)生原始數(shù)據(jù)，向發(fā)布者zmqsocket（zmqPUB）發(fā)送selfstart（）命令，此時(shí)發(fā)布者等待訂閱者connect（）方法的返回。步驟二：訂閱者通過(guò)self__init（）__方法初始化，在訂閱者上下文subscriber_context中創(chuàng)建若干worker，并通過(guò)Pollerregister（）方法將worker注冊(cè)進(jìn)輪詢(xún)池，每一個(gè)worker管理一塊訂閱者緩沖分區(qū)。同時(shí)，對(duì)發(fā)布者建立連接。步驟三：發(fā)布者將原始數(shù)據(jù)進(jìn)行封裝，以報(bào)文的形式發(fā)送給訂閱者，訂閱者在調(diào)度上下文中將消息分配給worker處理。步驟四：worker接收到消息后，在調(diào)度上下文中完成業(yè)務(wù)邏輯處理，并將處理完成的結(jié)果封裝成ZMQ消息發(fā)送給收集器。步驟五：收到終止信號(hào)后，收集器調(diào)用worker的終止事件回調(diào)，對(duì)worker進(jìn)行停止操作，訂閱者釋放worker資源。

2.2輪詢(xún)池設(shè)計(jì)與實(shí)現(xiàn)

為了實(shí)現(xiàn)消息快速高效的接收，防止在一個(gè)worker處集中過(guò)多的數(shù)據(jù)，導(dǎo)致該worker負(fù)載過(guò)大，盡可能降低worker隊(duì)列中消息的等待時(shí)長(zhǎng)是至關(guān)重要的。新增訂閱者輪詢(xún)池，主要負(fù)責(zé)訂閱者消息分配，偵聽(tīng)發(fā)布者是否發(fā)送新消息以及分配最后一個(gè)工作的worker為其進(jìn)行處理。圖5為輪詢(xún)池的數(shù)據(jù)結(jié)構(gòu)。

輪詢(xún)池是一個(gè)狀態(tài)輪詢(xún)接口，映射Python內(nèi)置輪詢(xún)機(jī)制，通過(guò)selfsockets列表保存輪詢(xún)對(duì)象與輪詢(xún)事件的二元組（worker_x， event_x），利用self_map哈希表保存輪詢(xún)對(duì)象在self_map中的索引值，提高輪詢(xún)時(shí)的查詢(xún)速度。

在訂閱者上下文中創(chuàng)建worker的同時(shí)，在輪詢(xún)池中注冊(cè)worker。訂閱者主線程接收消息時(shí)，輪詢(xún)池對(duì)同一線程中的不同輪詢(xún)對(duì)象進(jìn)行雙向異步偵聽(tīng)，偵聽(tīng)對(duì)象為已注冊(cè)的ZMQ套接字，注冊(cè)時(shí)采用ZMQ套接字引用與特定事件對(duì)的方式，當(dāng)偵聽(tīng)到特定事件發(fā)生在對(duì)應(yīng)ZMQ套接字上時(shí)，發(fā)送回調(diào)激活套接字。上游偵聽(tīng)發(fā)布者，為避免輪詢(xún)循環(huán)內(nèi)的任何阻塞調(diào)用，采用poll（）方法檢查是否有可用數(shù)據(jù)，如果偵聽(tīng)到上游有可用數(shù)據(jù)進(jìn)入，則彈出下一個(gè)worker套接字對(duì)其進(jìn)行業(yè)務(wù)處理;下游偵聽(tīng)worker，當(dāng)輪詢(xún)池偵聽(tīng)到可用數(shù)據(jù)時(shí)，刷新worker套接字recv（）端口，采用輪詢(xún)算法分配worker套接字處理消息。采用輪詢(xún)作為實(shí)現(xiàn)多線程worker負(fù)載均衡的算法。輪詢(xún)算法適用于訂閱者訂閱一種主題時(shí)，多個(gè)worker處理的消息的業(yè)務(wù)邏輯完全相同，可以近似地認(rèn)為每一個(gè)worker處理一條消息的開(kāi)銷(xiāo)相同，圖6為輪詢(xún)池消息分配示意圖。

2.3多線程WORKER設(shè)計(jì)與實(shí)現(xiàn)

為實(shí)現(xiàn)數(shù)據(jù)并行處理，訂閱者部署多線程worker負(fù)責(zé)處理業(yè)務(wù)邏輯。將節(jié)點(diǎn)概念擴(kuò)展，節(jié)點(diǎn)可以是主機(jī)、進(jìn)程甚至是線程，worker節(jié)點(diǎn)通過(guò)inproc協(xié)議（進(jìn)程內(nèi)連接）與訂閱者線程建立連接，采用inproc協(xié)議作為訂閱者與worker之間通信的橋梁，inproc協(xié)議是一種在單個(gè)ZMQ上下文線程間通過(guò)共享內(nèi)存實(shí)現(xiàn)消息傳遞的協(xié)議，可以顯著提高多線程worker數(shù)據(jù)傳輸效率。為了提高模型的并發(fā)性，對(duì)

多線程worker隔離性與一致性提出了要求。對(duì)于隔離性，worker間應(yīng)避免使用鎖和信號(hào)量等技術(shù)來(lái)實(shí)現(xiàn)并發(fā)隔離，在不使用時(shí)間開(kāi)銷(xiāo)較大的方案的前提下，需要防止同一報(bào)文重復(fù)被多個(gè)worker業(yè)務(wù)流處理，降低處理效率?；诖丝紤]，多線程worker通過(guò)輪詢(xún)池消息分配實(shí)現(xiàn)并發(fā)編程隔離性，輪詢(xún)算法作為一種無(wú)狀態(tài)調(diào)度算法，確保每條消息只會(huì)由一個(gè)worker進(jìn)行處理，且ZMQ并發(fā)編程本身采用無(wú)鎖算法，從而在保證性能的前提下實(shí)現(xiàn)多線程worker隔離性。一致性方面，worker接收到報(bào)文后對(duì)報(bào)文解包，進(jìn)行相應(yīng)業(yè)務(wù)流處理后，將處理結(jié)果封裝成ZMQ報(bào)文，發(fā)送至收集器匯總，收集器接收匯總消息，激活回調(diào)函數(shù)，更新統(tǒng)計(jì)信息;worker通過(guò)zmqPULL套接字發(fā)送匯總消息，與收集器之間通過(guò)inproc協(xié)議建立快速的進(jìn)程內(nèi)連接，多個(gè)worker間采用公平隊(duì)列算法實(shí)現(xiàn)消息匯總，采用該調(diào)度算法的優(yōu)點(diǎn)在于不會(huì)因高負(fù)載worker而阻塞其他worker匯總消息，可以在保證消息匯總一致性的同時(shí)，進(jìn)一步提高多線程worker性能。worker類(lèi)圖如圖7所示，包括zmqcontextContext定義上下文，zmqsocketSocket定義通信套接字等。

worker通過(guò)訂閱者上下文sub_context初始化后，與輪詢(xún)池建立進(jìn)程內(nèi)連接，完成操作后通過(guò)workerrecv（）方法異步等待消息分配，輪詢(xún)池偵測(cè)到消息到來(lái)后，會(huì)將消息分配給worker，worker對(duì)接收到的消息進(jìn)行業(yè)務(wù)流處理，通過(guò)ZMQ消息的形式發(fā)送結(jié)果給收集器，每處理一條消息，判斷一次當(dāng)前消息發(fā)送是否發(fā)生已經(jīng)結(jié)束，如果當(dāng)前狀態(tài)不再有新的消息傳入，則通過(guò)Python GC機(jī)制實(shí)現(xiàn)資源回收，節(jié)約開(kāi)銷(xiāo)。worker流程圖如圖8所示。

2.4狀態(tài)同步

受制于傳統(tǒng)發(fā)布訂閱模式單向數(shù)據(jù)傳輸?shù)乃枷?，訂閱者可以接收?lái)自發(fā)布者的消息，但發(fā)布者無(wú)法獲取訂閱者的狀態(tài)，雙方建立連接時(shí)無(wú)法做到狀態(tài)同步。因此，需要改進(jìn)傳統(tǒng)單向數(shù)據(jù)傳輸模式，使發(fā)布者可以獲取訂閱者的狀態(tài)信息，通過(guò)數(shù)據(jù)雙向傳輸達(dá)到狀態(tài)同步的目的。在發(fā)布者接收到上層應(yīng)用開(kāi)始命令后，通過(guò)新增REP套接字阻塞等待訂閱者狀態(tài)報(bào)文;訂閱者建立連接后，通過(guò)新增REQ套接字向發(fā)布者發(fā)送狀態(tài)報(bào)文。REP套接字接收狀態(tài)報(bào)文，回調(diào)激活發(fā)布者，實(shí)現(xiàn)發(fā)布訂閱雙方狀態(tài)同步，解決慢連接問(wèn)題。圖9為訂閱者流程圖。

3測(cè)試與分析

3.2測(cè)試方案

通過(guò)測(cè)試原生ZMQ發(fā)布訂閱模式與基于多線程worker的ZMQ發(fā)送不同長(zhǎng)度報(bào)文、固定個(gè)數(shù)報(bào)文以及進(jìn)行壓力測(cè)試時(shí)的丟包率。本次測(cè)試的網(wǎng)絡(luò)環(huán)境為3684Mb帶寬。用于發(fā)送報(bào)文的計(jì)算機(jī)配置如下：Win10操作系統(tǒng)，運(yùn)行內(nèi)存16G，內(nèi)存500G，CPU為AMD r5，六核，主頻為21-40GHz;用于接收?qǐng)?bào)文的計(jì)算機(jī)配置如下：Win10系統(tǒng)，運(yùn)行內(nèi)存12G，內(nèi)存500G，CPU為Intel CORE i5，四核，主頻為230GHz。

本次測(cè)試分為兩項(xiàng)基礎(chǔ)測(cè)試與兩項(xiàng)壓力測(cè)試，第一項(xiàng)基礎(chǔ)測(cè)試內(nèi)容為發(fā)送端發(fā)送一百萬(wàn)條不同大小報(bào)文，報(bào)文大小分別為64、128、256、512、1024、2048、4096、8192Byte;第二項(xiàng)基礎(chǔ)測(cè)試內(nèi)容為發(fā)送端發(fā)送總字節(jié)數(shù)為一百萬(wàn)Bytes的報(bào)文，每條報(bào)文的大小分別為64、128、256、512、1024、2048、4096、8192Byte。第一項(xiàng)壓力測(cè)試內(nèi)容為在不同額外負(fù)載下，發(fā)送一百萬(wàn)條消息，即測(cè)試性能時(shí)，同時(shí)以3M/s、6M/s和30M/s的速率發(fā)送其他流量，以非測(cè)試報(bào)文擠占帶寬;第二項(xiàng)壓力測(cè)試內(nèi)容為長(zhǎng)時(shí)間連續(xù)發(fā)送報(bào)文，發(fā)送時(shí)長(zhǎng)分別為1～10 min。

上述基礎(chǔ)測(cè)試中發(fā)送端均不設(shè)置發(fā)送間隔，并控制發(fā)送端分別以1M/s、2M/s和10M/s的速率進(jìn)行發(fā)送，測(cè)試其丟包率。上述壓力測(cè)試不設(shè)置發(fā)送間隔與發(fā)送速率，測(cè)試其丟包率。

3.2測(cè)試結(jié)果

基礎(chǔ)測(cè)試與壓力測(cè)試結(jié)果如圖10～17所示。圖10為ZMQ基礎(chǔ)測(cè)試1丟包率，分別測(cè)試發(fā)送一百萬(wàn)條報(bào)文，報(bào)文長(zhǎng)度分別為64、128、256、512、1024、2048、4096、8192Byte時(shí)的丟包率，發(fā)送速率分別為1M/s、2M/s和10M/s，當(dāng)報(bào)文大小增長(zhǎng)到512Byte時(shí)，丟包率有明顯提升，達(dá)到了6745%;圖14和圖16分別為ZMQ壓力測(cè)試1和壓力測(cè)試2的丟包率，可以觀察到在512Byte時(shí)丟包率陡增，分別達(dá)到了6899%和7296%。該現(xiàn)象是由于報(bào)文過(guò)大導(dǎo)致訂閱者業(yè)務(wù)流阻塞時(shí)間過(guò)長(zhǎng)，期間緩沖區(qū)溢出以致于大部分報(bào)文被丟棄。

從基礎(chǔ)測(cè)試和壓力測(cè)試結(jié)果中可知，原生ZMQ在發(fā)布512Byte以下大小報(bào)文時(shí)，可以保持較低的丟包率，即使是在帶寬被擠占和長(zhǎng)時(shí)間發(fā)送的情況下，發(fā)布小型報(bào)文的丟包率都較低;而當(dāng)報(bào)文大小增長(zhǎng)到512Byte及以上時(shí)，丟包率發(fā)生陡增，說(shuō)明原生ZMQ對(duì)于大報(bào)文的處理能力較低。本文提出的多線程worker數(shù)據(jù)分發(fā)模型通過(guò)worker并行讀取數(shù)據(jù)，提高訂閱者緩沖區(qū)的利用率，增強(qiáng)訂閱者面對(duì)大型報(bào)文時(shí)的處理能力，在保持高效處理小型報(bào)文的基礎(chǔ)上，提高了訂閱者處理大型報(bào)文的能力，測(cè)試結(jié)果表明，本模型在處理不同大小報(bào)文的壓力測(cè)試下，都能保持5%以下的丟包率。

4結(jié)論

本文針對(duì)傳統(tǒng)數(shù)據(jù)分發(fā)模型高丟包問(wèn)題，提出了一種基于多線程worker的高可靠數(shù)據(jù)分發(fā)服務(wù)模型，解決傳統(tǒng)模型中慢訂閱者與慢連接問(wèn)題。在發(fā)送不同長(zhǎng)度報(bào)文和相同總長(zhǎng)度報(bào)文的基礎(chǔ)測(cè)試中，接收長(zhǎng)度小于512Byte的短報(bào)文時(shí)丟包率降低2211%;接收長(zhǎng)度大于等于512Byte的長(zhǎng)報(bào)文時(shí)丟包率降低8737%。在非測(cè)試報(bào)文擠占帶寬壓力測(cè)試中，本模型降低丟包率5920%，在長(zhǎng)時(shí)間連續(xù)發(fā)送報(bào)文壓力測(cè)試中，本模型降低丟包率5565%。實(shí)驗(yàn)表明，本文提出的多線程數(shù)據(jù)分發(fā)模型在發(fā)布訂閱長(zhǎng)度在8192Byte以?xún)?nèi)報(bào)文時(shí)可以顯著降低丟包率，該模型是一種有效的模型。

基于多線程的數(shù)據(jù)分發(fā)模型丟包率低，可以用于改善強(qiáng)化學(xué)習(xí)中多智能體間數(shù)據(jù)交互，使得模型評(píng)估更加精準(zhǔn)，仿真效果更加完善。雖然該模型可以有效降低報(bào)文丟包率，但在發(fā)布者高頻發(fā)送報(bào)文的情況下，丟包率仍然較高，因此仍然需要進(jìn)一步研究高頻發(fā)送報(bào)文丟包的原因和解決方案。

參 考 文 獻(xiàn)：

[1]郝玲玲，傅妍芳.發(fā)布/訂閱模型DDS應(yīng)用研究及其性能評(píng)價(jià)[J].西安工業(yè)大學(xué)學(xué)報(bào)，2020，40（3）：290.

HAO Lingling， FU Yanfang. Application and Performance Evaluation of DDS Based on Publish/Subscribe Model [J]. Journal of Xi′an Technological University，2020，40（3）：290.

[2]朱子杰，汪敏，李榮寬，等.基于DDS戰(zhàn)術(shù)服務(wù)框架的實(shí)現(xiàn)機(jī)制[J].指揮信息系統(tǒng)與術(shù)，2020，11（4）：52.

ZHU Zijie， WANG Min， LI Rongkuan， et al. Implementation Mechanism for Tactical Service Framework Based on DDS [J]. Command Information System and Technology，2020，11（4）：52.

[3]雷媛元，焦璐，王銳，等.基于數(shù)據(jù)分發(fā)服務(wù)的通用仿真框架技術(shù)[J].計(jì)算機(jī)應(yīng)用，2020，40（S1）：146.

LEI Yuanyuan， JIAO Lu， WANG Rui， et al. General Simulation Framework Based on Data Distribution Service[J]. Journal of Computer Applications，2020， 40（S1）：146.

[4]劉家雨，王永生，劉愛(ài)東.基于DDS的防空武器信息交互系統(tǒng)[J].指揮控制與仿真，2021，43（2）：127.

LIU Jiayu， WANG Yongsheng， LIU Aidong. Air Defense Weapon Information Exchange System Based on DDS [J]. Command Control & Simulation，2021，43（2）：127.

[5]山壽，王鵬，聶瑤佳.基于任務(wù)容器的多源數(shù)據(jù)實(shí)時(shí)監(jiān)控技術(shù)研究[J].計(jì)算機(jī)測(cè)量與控制，2020，28（3）：98.

SHAN Shou， WANG Peng， NIE Yaojia. Research on Multisource Data Realtime Monitoring System Based on Task Container [J]. Computer Measurement & Control，2020，28（3）：98.

[6]孫曉冬.基于數(shù)據(jù)分發(fā)服務(wù)的雷達(dá)仿真系統(tǒng)設(shè)計(jì)[J].科技視界，2019（18）：6.

SUN Xiaodong， Radar Simulation System Design Based on Data Distribution Service [J]. Science & Technology Vision，2019（18）：6.

[7]山壽，郝明哲，孫偉.基于數(shù)據(jù)分發(fā)服務(wù)和WPF技術(shù)的試飛實(shí)時(shí)監(jiān)控系統(tǒng)設(shè)計(jì)[J].計(jì)算機(jī)測(cè)量與控制，2020，28（3）：119.

SHAN Shou， HAO Mingzhe， SUN Wei. Design of Realtime Monitoring System for Flight Test Based on DDS and WPF Technology [J]. Computer Measurement & Control，2020，28（3）：119.

[8]畢曉龍.基于數(shù)據(jù)分發(fā)服務(wù)的聲吶仿真系統(tǒng)的研究[J].艦船電子工程，2019，39（12）：191.

BI Xiaolong. Research on Sonar Simulation System Based on Data Distribution Service [J]. Ship Electronic Engineering，2019，39（12）：191.

[9]CHA J H， KIM D S. Design and Implementation of a Realtime Monitoring Tool for Data Distribution Service[J]. IEIE Transactions on Smart Processing & Computing，2018，7（4）：264.

[10]劉元斌，占日新.航電系統(tǒng)仿真環(huán)境通信架構(gòu)研究[J].中國(guó)新技術(shù)新產(chǎn)品，2021（3）：16.

LIU Yuanbin， ZHAN Rixin. Research on Energy Electronics System Simulation Environment Communication Architecture [J]. China New Technology New Product，2021（3）：16.

[11]王坤，房玉吉，馮源，等.基于Qt和OpenDDS的船舶電力模擬訓(xùn)練系統(tǒng)指令處理方法[J].船海工程，2018，47（6）：50.

WAN GKun， FANG Yuji， FENG Yuan， et al.? Command Processing of the Ship′s Power Simulation Training System Based on QT and OpenDDS [J]. Ship & Ocean Engineering，2018，47（6）：50.

[12]杜越洋，趙盾，閆智超.基于數(shù)據(jù)分發(fā)服務(wù)的無(wú)人機(jī)任務(wù)載荷綜合仿真平臺(tái)研究[J].無(wú)人系統(tǒng)技術(shù)，2021，4（1）：79.

DU Yueyang， ZHAO Dun， YAN Zhichao. Research on DDSbased Integrated Simulation Platform for UAV Mission Payload [J]. Unmanned Systems Technology，2021，4（1）：79.

[13]TAREK A.Youssef，Mohammad Mahmoudian Esfahani，Osama Mohammed. DataCentric Communication Framework for Multicast IEC 61850 Routable GOOSE Messages over the WAN in Modern Power Systems[J]. Applied Sciences，2020，10（3）：848.

[14]史佳雯，汪洋，張庚，等.面向電力通信網(wǎng)的分布式仿真系統(tǒng)[J].計(jì)算機(jī)工程與應(yīng)用，2019，55（19）：246.

SHI Jiawen， WANG Yang， ZHANG Geng， et al. Distributed Simulation System for Power Communication Network [J]. Computer Engineering and Applications， 2019，55（19）：246.

[15]金志平，梁志標(biāo)，劉勝華，等.自適應(yīng)動(dòng)態(tài)視頻流傳輸策略研究[J].網(wǎng)絡(luò)安全技術(shù)與應(yīng)用，2021（5）：57.

JIN Zhiping， LIANG Zhibiao， LIU Shenghua， et al. Research on Adaptive Dynamic Video Flow Transmission Strategy [J]. Network Security Technology & Application，2021（5）：57.

[16]韓旭，趙國(guó)榮，王康.基于線性編碼補(bǔ)償方法的非固定丟包率下的分布式融合估計(jì)器[J].北京航空航天大學(xué)學(xué)報(bào)，2020，46（6）：1229.

HAN Xu， ZHAO Guorong， WANG Kang. A Decentralized Fusion Estimator Using Linear Coding Compensation Method with Nonfixed Dropout Rates [J]. Journal of Beijing University of Aeronautics and Astronautics，2020，46（6）：1229.

[17]孫學(xué)軍.降低上行丟包率提升VoLTE語(yǔ)音質(zhì)量的研究[J].中小企業(yè)管理與科技（上旬刊），2019（7）：182.

SUN Xuejun. Research on Reducing Uplink Packet Loss Rate and Improving the Quality of VoLTE Speech [J]. Management & Technology of SME， 2019（7）：182.

[18]王建，王康，劉方森.基于MR的VoLTE高丟包優(yōu)化分析方法研究[J].電信技術(shù)，2019（6）：19.

WANG Jian， WANG Kang， LIU Fangsen. Research on MRbased Volte High Packet Loss Packet Optimization Analysis Method [J]. Telecommunications Technology， 2019（6）：19.

[19]李春雷，高峰，顏運(yùn)強(qiáng).基于Actor模型的軟總線設(shè)計(jì)[J].計(jì)算機(jī)工程，2019，45（5）：77.

LI Chunlei， GAO Feng， YAN Yunqiang. Design of Soft Bus Based on Actor Model [J]. Computer Engineering，2019，45（5）：77.

[20]童佳鋒.基于DDS的軟件接口測(cè)試方法研究[J].電子測(cè)試，2021（9）：113.

TONG Jiafeng. Research on DDSbased Software Interface Testing Method [J]. Electronic Test， 2021（9）：113.

（編輯：王萍）