題圖來自https://www.pexels.com/

?概述

在傳統企業級開發中，消息隊列機制已經成為一種非常常見的技術實現手段，而基于NetMQ則看起來有點像一朵“奇葩”，看起來從名字似乎是一個消息隊列（Message Quene），但事實上更多的卻是一個類似于socket機制的消息庫。它雖然提供了消息隊列的能力，但又與傳統消息隊列中間件如kafka、rabbitmq等有一定的區別。

不過，不管它是啥，它提供的一些類似于消息隊列的機制，使得開發者能夠快速在項目中使用起來，例如類似于發布訂閱模式、推拉模式等機制，接入簡便，功能也挺強大。而且當如果我們要實現消息加密時，還可能通過一些簡單的操作實現，例如我們可以選擇對內容進行Rsa加密，或者也許還有其他的實現方法？

TL;DR:本文首先介紹NETMQ及其常用的使用模式，進而討論如何基于NETMQ實現消息的加密傳輸機制。

NetMq簡介和基本特性

ZeroMQ

NETMQ是一種輕量級的消息隊列組件，是著名的ZeroMQ的重要成員。2010年，AMQP的最初設計者Pieter Hintjens帶領其團隊退出了該開源項目，并發起成立了ZeroMQ這個新的消息庫，并發展至今。Pieter Hintjens 后由于膽管癌復發，于2016年接受了安樂死。

在ZeroMQ的官方網站中，其介紹到ZeroMQ看起來似乎是一個消息隊列框架，實際上更像一個并發處理框架。它除了提供了多種消息隊列機制（如Pub-Sub、Pull-Push、Dealer、XPub-XSub機制）外，更是為開發者提供了跨多種傳輸能力的套接字，它不僅適用于進程間的消息傳輸，也同樣適用于進程內、TCP和多播的傳輸機制，基于其提供的框架，開發者能快速的實現原子消息的傳輸能力。ZeroMQ的輕量級體現在其框架靈活簡單，性能優異，無需依賴外部組件，即可輕松實現優秀的性能。它也支持異步I/O的傳輸機制，可為多核應用程序提供擴展，且能成為集群部署的核心傳輸組件。

ZeroMQ提供了多種語言實現，參見其官方網站，包括C語言，C#，Java等主流后端語言，都支持良好，同樣，也支持包括Go、Node.JS等最近比較熱門的新興語言。ZeroMQ自然也支持不同語言間的數據傳輸，使其可以成為跨語言傳輸的一種消息協議。

ZeroMQ的Zero，代表一種極簡文化，可以代表零代理層（與Mqtt等隊列機制不同，ZeroMQ提供的是一種無代理層的隊列機制），零延遲，零成本和零管理。ZeroMQ致力于打造極簡的通信組件，通過消除組件的復雜性來提升其功能應用效果。

NETMQ和ClrZmq

對于C#開發者來說，可以使用NetMQ和ClrZmq兩種不同的方式來獲得ZeroMQ的魔力，前者是基于C#語言原生實現的ZeroMQ通信協議，后者則是通過C#調用基于C語言實現 的Libzmq庫來使用。

相對而言，前者可能更受歡迎。NETMQ也同樣繼承了ZeroMQ的優雅性能和輕量化，開發者可通過Nuget下載NetMQ的的組件，通過幾行代碼就可以集成消息隊列和套接字傳輸能力。如圖所示，NetMQ獲得了約175w的下載量，算是一個比較受歡迎的基礎組件。

而同樣在nuget上，ClrZmq的下載量則遠遠少于NetMQ，僅僅8w多的下載量，可能說明它只是一種小眾框架吧。值得一提的是，ClrZmq需要根據構建平臺來選擇不同的架構。

NETMQ的組成部分

截止本文撰寫時，NETMQ的版本為4.0.1.6，作為輕量級組件的一個評判標準，依賴項復雜度也是個重要指標，而NetMQ只依賴了AsyncIO、NaCI.NET、System.ServiceModel.Primitives、System.Threading.Tasks.Extension、System.ValueTuple五個組件，算是名副其實，此處重點介紹兩個非System開頭的組件。

AsyncIO：該組件是一個高性能的異步的消息套接字庫，事實上在Nuget上，該消息庫比NetMQ更受歡迎，基于該組件，可減少套接字開發的成本。

NaCI：該組件是一個加密組件，實現了包括Curve25519x、Salsa20、Poly1305加密算法。Curve25519是一種橢球曲線加密算法，被設計用于橢圓曲線迪菲-赫爾曼（ECDH）密鑰交換方法。Salsa20是一種流加密算法。Poly1305是一種消息認證碼，可用于檢測消息的完整性和驗證消息的真實性，現常在網絡安全協議（SSL/TLS）中與salsa20或ChaCha20流密碼結合使用。這三種算法都是由密碼專家丹尼爾·J·伯恩斯坦設計的加密算法。

這兩個組件都是由NETMQ的創建者somdoron^[1]（Doron Somech）創建，并引入到NETMQ中。

官方網站

NETMQ官方網站地址為https://netmq.readthedocs.io/，該網站提供了較為完整的學習示例，開發者可參考該示例快速學會該組件的用法。

常見模式實現

NETMQ提供了多種消息通信機制，例如發布訂閱模式，推拉模式，

發布訂閱模式（Pulish-Subscriber Pattern）

簡介

發布-訂閱是一種消息傳遞模式，其中消息的發送者（稱為發布者）不會將消息編程為直接發送給特定的接收者（稱為訂閱者）。發布的消息按照主題進行特征化，作為發布者事先不用知道可能有哪些訂閱者（如果有）。

類似地，訂閱者可訂閱多個主題，也可只訂閱一個主題。訂閱者也同樣無需關注發布者是否真實存在，不過由于ZeroMQ本身沒有代理層，且需要綁定服務端端口，事實上看起來似乎必須給定發布者。但由于ZeroMQ本身也可以作為一種微服務架構的基礎設施^[2]，實際上也可以通過一些機制，例如消息代理，地址代理，DNS網關如ZeroConf，Gossip協議等機制，將發布者隱藏在消息網關背后，從而使得訂閱者無需關注發布者具體在哪里。

代碼示例

該需要首先創建一個發布者，并通過主題的形式發布消息。

class Program {private static string _address = "";static void Main(string[] args){Console.WriteLine("Hello World!");_address = "tcp://127.0.0.1:5556"; //設置端口var task = Task.Factory.StartNew(async() =>{await BeginPublisherAsync();});var taskSubScriber = Task.Factory.StartNew(() =>{ BeginSubscriberSocket();}); while(Console.ReadKey().Key!=ConsoleKey.Escape);}/// <summary>/// 啟動消息發布/// </summary>/// <returns></returns>private static async Task BeginPublisherAsync(){using (var publisher = new PublisherSocket()){publisher.Bind(_address); //綁定端口while (true){publisher.SendMoreFrame("DotNET技術圈") // Topic.SendFrame("test"); // Messageawait Task.Delay(TimeSpan.FromSeconds(1));}}}/// <summary>/// 訂閱消息/// </summary>private static void BeginSubscriberSocket(){using (var subscriber = new SubscriberSocket()){subscriber.Connect(_address);subscriber.SubscribeToAnyTopic();while (true){var topic = subscriber.ReceiveFrameString();var msg = subscriber.ReceiveFrameString();Console.WriteLine("收到消息: {0} {1}", topic, msg);}}} }

在上述代碼中，發布者綁定了tcp://127.0.0.1:5556端口，并通過同步阻塞的方式，發布主題為Topic的消息內容。也可以指定主機的固定ip地址來進行消息發布，還能通過inproc://inproc-demo的方式進行進程內通信。

using var subscriber = new SubscriberSocket()subscriber.Connect("tcp://127.0.0.1:5556");subscriber.Subscribe("TopicA"); //訂閱到TopicA主題，也可通過SubscribeToAnyTopic訂閱所有主題，也可通過UnSubcribe取消訂閱相關主題while (true) {var topic = subscriber.ReceiveFrameString();var msg = subscriber.ReceiveFrameString();Console.WriteLine("From Publisher: {0} {1}", topic, msg); }

請求響應模式（Request-Response Pattern)

請求響應模式也是NETMQ眾多消息模式中最為簡單的一種模式，這種模式實際上有點像http協議，可通過一問一答的同步阻塞的模式進行消息的應答，當然，發送HTTP請求我們也可以不必接收響應，NETMQ的請求響應模式也同樣如此。

示意圖

private static void BeginResponseSocket(){using var responseSocket = new ResponseSocket(_address);string request=responseSocket.ReceiveFrameString();responseSocket.SendFrame("Hello DotNET技術圈");} private static async Task BeginRequestSocketAsync() {using var requestSocket = new RequestSocket();requestSocket.Connect(_address);while (requestSocket.TrySignalOK()){try{requestSocket.TrySendFrame("Hallo I am DotNET技術圈碼農");requestSocket.TrySendFrame("Hallo I am DotNET技術圈碼農"); ---這里會引發錯誤。。}catch(Exception ex){Console.Out.WriteLine(ex);}await Task.Delay(1000);} }

由于該模式的同步阻塞特性，如果同時發送兩條消息，可能會觸發NETMQ重復發送異常，如：

推拉模式

推拉模式與我們傳統意義上理解的類似于手機推送的模式有一些區別，ZeroMQ中說該模式主要將消息下發到提供了一組Push-Pull的套接字，實現消息下發。

值得一提的是，即便的同為ZeroMQ模式下不同語言的版本，對于相同模式的說明，文字描述也不盡相同，例如，在NetMQ的開發者文檔中，

Well a?PushSocket?is normally used to push to a?PullSocket, whilst the?PullSocket?will pull from a?PushSocket. Sounds obvious right!

PushSocket 負責把消息推給PullSocket，同樣PullSocket負責從PushSocket 拉消息。

這樣的說明似乎什么都說了，但又似乎啥都沒說，看看其他語言的實現，例如基于Python的PyZmq中，其描述為這樣：

Push and Pull sockets let you distribute messages to multiple workers, arranged in a pipeline. A Push socket will distribute sent messages to its Pull clients evenly. This is equivalent to producer/consumer model but the results computed by consumer are not sent upstream but downstream to another pull/consumer socket.

推拉模式允許你基于通過管道的機制實現消息分發給多個工作者。單個PushSocket分發會將消息均勻的分發給其Pull客戶端。這樣的操作等效于生產者-消費者模型，但消費者計算的結果不是向上發送，而是向下游發送到另一個拉/消費者套接字。

兩種不同的實現，在描述上區別還是顯著不同，通過兩者的對比，我們可以這樣理解：Push-Pull模式下，兩者都可以互為服務端或客戶端，但無論如何，其消息都是單向傳輸的。消息總是沿著管道向下流動，沿著我們設計的方向傳輸，實現消息在不同節點間的負載均衡。

例如，可以實現如下的效果，通過一個Ventilator來生產數據，通過多個Pull來拉取數據，進而實現數據向下流動，可以參考NetMq官方文檔來實現該代碼。

基于推拉模式，可以設計非常負責的業務模型，例如類似于MapReduce的數據處理器^[3]就是一個這樣的教學工具。（當然，該工具只是一個演示ZeroMQ模式實現的分布式計算的Demo，可能不適合作為生產用途）。

代碼示意

本示例中，僅僅簡單介紹Push-Pull的用法，暫不涉及復雜的模式。

private static async Task BeginPushSocketAsync() {using var pushSocket = new PushSocket(_address);while (true){pushSocket.SendFrame("Hello Clients");await Task.Delay(1000);} }private static async Task BeginPullSocketAsync() {using var pullSocket = new PullSocket(_address);while (true){string message = pullSocket.ReceiveFrameString();Console.WriteLine(message);await Task.Delay(1000);} }

netmq加密傳輸機制實現

當我們使用NetMQ進行消息傳輸時，上述示例均沒有對消息進行任何加密處理，這種策略可能導致一些不可控的安全性風險，例如在開發基于NetMQ的聊天室功能時，發布的信息若未采取任何加密措施，事實上可能意味著消息是以廣播的形式對外發布，從而會造成某些隱私信息泄漏。或者，如果你需要向外Publish某些消息，未授權的訂閱者訂閱了你的數據，雖然可能數據中不包含直接的隱私數據，但同樣可能會引起你的不適。

因此，從安全性的角度來說，無論你計劃基于NetMQ實現何種場景，事實上可能都得考慮以盡可能安全的形式“發布”你的消息。目前我們可通過三種方式來實現消息的加密傳輸功能。第一種是使用基于Tls協議的NetMQ.Security組件，一種是基于非對稱密鑰算法，如RSA加密算法，還有一種是基于ZeroMQ所提供的兩種加密方式，ECC橢球曲線加密算法和Z85加密算法，以對稱密鑰的方式。

基于Tls的NetMq.Security？

NetMQ.Security^[4]也是由NetMQ的主要貢獻者somdoron開發的組件，目前該組件處于不活躍的狀態，截至目前僅有5次更新，上一次更新依然是4年前，通過一些早期帖子，作者Doron Somech也同樣不認為該組件可以在生產環境下使用^[5]（😯），所以事實上可能不太適合作為專業團隊的技術選型。

目前比較詳細的介紹來自杰哥很忙^[6]，且優秀的杰哥對fork了該組件的代碼^[7]，并開發了許多功能，由于主干倉庫已經塵封太久了，開發者有興趣可以參詳參詳。

使用時，我們可通過Nuget下載由NetMQ官方發布的組件，不過，似乎下載量有點慘淡，那么，此處就不再贅述了。。。。

非對稱密鑰算法-Rsa加密

對于文本來說，使用Rsa這種非對稱算法族進行加密是一種非常常見的選擇，RSA是由羅納德·李維斯特^[8]（Ron Rivest）、阿迪·薩莫爾^[9]（Adi Shamir）和倫納德·阿德曼^[10]（Leonard Adleman）在1977年一起提出的，當時他們三人都在麻省理工學院^[11]工作。RSA 就是他們三人姓氏開頭字母拼在一起組成的。

RSA算法的核心是極大整數做因數分解，換言之，對一極大整數做因式分解越困難，RSA算法越可靠。目前傳統計算機只能破解較為簡單的RSA密鑰，如果使用的密鑰長度足夠長，理論上用RSA加密的信息也很難以被破解。在RSA算法中，密鑰由私鑰和公鑰組成。由私鑰負責對內容進行解密，并用公鑰進行加密。分配公鑰的過程必須足夠安全，若被中間人攻擊，則可能導致公鑰失效。

影響RSA密鑰安全性的要素首先是其密鑰長度，目前推薦的RSA算法公鑰長度為2048位。其次是RSA密鑰的填充模式，共有三種填充模式，RSA_PKCS1_PADDING， RSA_PKCS1_OAEP_PADDING， RSA_NO_PADDING。填充技術實現的不好，RSA也不會安全，應盡量選擇最安全的填充模式，例如RSA_PKCS1_PADDING。

原因如下^[12]：

1.RSA加密是確定的，即給定一個密鑰，特定明文總會映射到特定的密文。攻擊者可以根據密文中統計信息獲取明文的一些信息。2.填充技術如果比較弱，那么較小的明文和小型公開指數e將易于受到攻擊。3.RSA有個特性叫做延展性，如果攻擊者可以將一種密文轉換為另一種密文，這種新密文會導致對明文的轉換變得可知，這種特性并沒有解密明文，而是以一種可預測的方式操縱了明文，比如：銀行交易系統中，攻擊者根據新密文，直接去修改原密文中金額的數據，可以在用戶和接受方無法感知的情況下進行修改。

在RSA算法中提供了以下功能提供^[13]：

?密鑰對生成：生成隨機私鑰（通常大小為 1024-4096 位）和相應的公鑰。?加密：使用公鑰加密秘密消息（范圍為 [0...key_length] 的整數），并使用秘密密鑰將其解密。?數字簽名：簽署消息（使用私鑰）并驗證消息簽名（使用公鑰）。?密鑰交換：安全地傳輸一個秘密密鑰，用于以后的加密通信。

RSA 可以使用不同長度的密鑰：1024、2048、3072、4096、8129、16384 甚至更多位的密鑰。3072 位及以上的密鑰長度被認為是安全的。更長的密鑰提供更高的安全性，但消耗更多的計算時間，因此需要在安全性和速度之間進行權衡。很長的 RSA 密鑰（例如 50000 位或 65536 位）對于實際使用來說可能太慢，例如密鑰生成可能需要幾分鐘到幾個小時。

網上也有基于RSA進行NetMQ進行消息加密的示例^[14]，可供參考。其核心流程為，在進行消息發送時，使用RSA公鑰進行加密，

MsgObject sendmsg = EventQueue.Dequeue ( ) ; sendmsg.Content = RSAEncryption.RSAEncrypt(sendmsg.Content); sendmsg.MachineName= msg.MachineName; SendMessageQueue.Enqueue(sendmsg) ;

并在客戶端接收到消息后，對正文進行RSA解密，解密代碼略。

使用對稱密鑰加密算法-Ecc加密算法進行消息加密

RSA算法雖好，但由于私鑰由客戶端管理，公鑰由服務端管理，且RSA必須密鑰位數足夠長才安全，例如2048位，使用這么長的密鑰進行加密時間開銷也令人吃不消的，有沒有一種更簡單、快速的算法來實現呢？

使用AES算法？

我們或許會想到AES算法，例如AES256算法這種“對稱密鑰加密算法^[15]”。在“對稱密鑰加密算法”中，加密和解密使用秘密密鑰或密碼短語（從中派生出密鑰）。

該秘密密鑰用于加密和解密數據，通常是128位或256位，并被稱為“加密密鑰”。有時它以十六進制或 base64 編碼的整數形式給出，或者通過密碼到密鑰派生方案派生，當輸入數據被加密時，它被轉換為加密的密文，當密文被解密時，它被轉換回原始輸入數據。

通常，對稱加密過程使用一系列步驟，涉及不同的加密算法：

密碼到密鑰派生算法（如 Scrypt 或 Argon2）：允許使用密碼而不是密鑰，并使密碼破解變得困難而緩慢。

塊到流密碼轉換算法（塊密碼模式如CBC或CTR ）+消息填充算法如PKCS7 （在某些模式下）：允許使用塊密碼算法（如AES）加密任意大小的數據。

塊密碼算法（如AES ）：使用密鑰安全地加密固定長度的數據塊。

消息認證算法（如HMAC ）：檢查解密后得到的結果是否與加密前的原始消息匹配。

NETMQ的原生解決方案？

不過上述AES加密算法實質上也需要開發者手工處理消息體，存在的內存開銷和時間可能對于用戶來說依然無法接受，或許最好的辦法依然是基于NETMQ框架來入手看看是否有什么“原生”的解決方案。

所幸ZeroMQ在設計之初就已經將安全作為其認為非常重要的一個方面，在這篇博客^[16]中，ZeroMQ提到了其對于安全層的目標，包括：

?它使用起來必須非常簡單，而且不可能出錯。復雜性是密碼學的第一大風險和第一大漏洞。每一個額外的選項都是一種出錯的方式，最終導致一個不安全的系統。?對于實際工作，它必須足夠快。如果安全性使系統變得太慢而無法使用，人們就會將其關閉，因為今天能夠工作的務實需求勝過明天被入侵的風險。?它必須基于標準化協議，以便任何團隊都可以重新實施、獨立驗證并在軟件堆棧之外進行改進。?等等。

并從2013年起，在ZeroMQ版本（4.0.0）中就已經引入了安全架構設計，包括：

?一種新的有線協議ZMTP 3.0^[17]，為所有 ZeroMQ 連接添加了安全握手。?一種新的安全協議CurveZMQ^[18]，它通過 TCP 連接在兩個 ZeroMQ 對等點之間實現“完美的前向安全”。我將在下面解釋 CurveZMQ。?ZMTP 的一組安全機制：NULL、PLAIN 和 CURVE，每個機制都由它們自己的 RFC 描述。NULL 本質上是我們之前所擁有的。PLAIN 允許簡單的用戶名和密碼驗證。CURVE 實現了 CurveZMQ 協議。、?等等。

在ZeroMQ中集成的橢球曲線算法為Curve25519^[19]?，目前，在我們所使用的NetMQ中也同樣集成了該算法。在搞清楚原理后，我們再來使用該算法，發現一切就變得非常簡單明了了。

var serverPair = NetMQ.NetMQCertificate.CreateFromSecretKey(UTF8Encoding.UTF8.GetBytes(”這里是密鑰“));;using var server = new PublishSocket();server.Options.CurveServer = true;server.Options.CurveCertificate = serverPair;server.Bind($"tcp://127.0.0.1:55367");using (var server = new SubscriberSocket()){var cert = NetMQ.NetMQCertificate.CreateFromSecretKey(UTF8Encoding.UTF8.GetBytes(”這里是密鑰“));var curveServerCertificate = serverPair;var clientCertificate = new NetMQCertificate(); ---這里是客戶端密鑰，server.Options.CurveServerCertificate = curveServerCertificate; ---這里是使用服務端密鑰server.Options.CurveCertificate = clientCertificate; ---這里是客戶端密鑰}

結語

本文對NetMQ進行了簡單的概述，包括其常見模式和加密傳輸機制，開發者若有興趣，可通過NetMQ官網獲得更多學習資料。如果開發者加密算法感興趣，還可以通過這個網站(https://cryptobook.nakov.com）讀到許多有關加密的基礎知識。

References

[1]?somdoron:?https://github.com/somdoron
[2]?微服務架構的基礎設施:?https://zguide.zeromq.org/docs/chapter8/
[3]?類似于MapReduce的數據處理器:?https://github.com/sdiehl/kaylee
[4]?NetMQ.Security:?https://github.com/NetMQ/NetMQ.Security
[5]?同樣不認為該組件可以在生產環境下使用:?https://groups.google.com/g/netmq-dev/c/3tcsLvxUWgc
[6]?杰哥很忙:?https://www.cnblogs.com/Jack-Blog/p/9015783.html
[7]?該組件的代碼:?https://github.com/GuojieLin/NetMQ.Security
[8]?羅納德·李維斯特:?https://zh.wikipedia.org/wiki/羅納德·李維斯特
[9]?阿迪·薩莫爾:?https://zh.wikipedia.org/wiki/阿迪·薩莫爾
[10]?倫納德·阿德曼:?https://zh.wikipedia.org/wiki/倫納德·阿德曼
[11]?麻省理工學院:?https://zh.wikipedia.org/wiki/麻省理工學院
[12]?原因如下:?https://blog.csdn.net/makenothing/article/details/88429511
[13]?以下功能提供:?https://cryptobook.nakov.com/asymmetric-key-ciphers/the-rsa-cryptosystem-concepts
[14]?消息加密的示例:?https://blog.actorsfit.in/a?ID=01400-85ae6267-6c93-41e3-b06b-5d9792a422ba/
[15]?對稱密鑰加密算法:?https://cryptobook.nakov.com/symmetric-key-ciphers
[16]?這篇博客:?https://jaxenter.com/using-zeromq-security-part-1-119346.html
[17]?ZMTP 3.0:?http://zmtp.org/
[18]?CurveZMQ:?http://curvezmq.org/
[19]?Curve25519:?http://cr.yp.to/ecdh.html

總結

以上是生活随笔為你收集整理的浅议NetMQ常见模式和消息加密机制的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。