NetMQ介绍
NetMQ 是 ZeroMQ的C#移植版本。
一、ZeroMQ
ZeroMQ（Ø）是一个轻量级的消息内核，它是对标准socket接口的扩展。它提供了一种异步消息队列，多消息模式，消息过滤（订阅），对多种传输协议的无缝访问。
ZeroMQ是基于消息队列的多线程网络库，其对套接字类型、连接处理、帧、甚至路由的底层细节进行抽象，提供跨越多种传输协议的套接字。ZeroMQ是网络通信中新的一层，介于应用层和传输层之间（按照TCP/IP划分），其是一个可伸缩层，可并行运行，分散在分布式系统间。
ZeroMQ几乎所有的I/O操作都是异步的，主线程不会被阻塞。ZeroMQ会根据用户调用zmq_init函数时传入的接口参数，创建对应数量的I/O Thread。
ZeroMQ不是单独的服务或者程序，仅仅是一套组件，其封装了网络通信、消息队列、线程调度等功能，向上层提供简洁的API，应用程序通过加载库文件，调用API函数来实现高性能网络通信。
特点
如果说ZeroMQ最突出的三个特点是什么？ 答案是 快，很快，非常快。
不仅仅是速度，就连使用起来ZeroMQ和其他的MQ产品比起来差别太大，以至于有种疑问，ZeroMQ还算的上队列产品麽？
通常的MQ，你需要一台运行在服务器上的MQ服务端，它需要启动一个或多个进程来维护队列中的数据并做持久化以防止宕机后数据丢失，当然还必须要有有个监控组件保证能尽快的发现队列的阻塞延迟处理速率等运行状况。然后你至少要有一个客户端通过订阅发布或者请求响应的方式与队列服务端交换数据。
ZeroMQ则完全不同，它不需要独立部署，反之它仅提供一个DLL，你可以将它Host到任何运行的进程中去，IIS、Windows Servers、Window Form都可以。你只需要告诉运行模式和通信端口，就能马上运行起来。 像不像WCF ？
另外一个不得不说的特点是ZeroMQ彻底放弃了持久化，虽然你可以设置一个磁盘Swarp区域用以临时存放队列数据，但一旦工作线程结束，所有数据都将丢失。
最后说一说ZeroMQ的多对多的优势，一般的MQ产品如果要实现多个队列服务和多个客户端的自由连接就不得不额外的做自己的实现。但ZeroMQ天生就具备这个能力，甚至在可以任何时候添加和取消服务端和客户端的任一个。
消息模型
ZeroMQ将消息通信分成4种模型，分别是
- 一对一结对模型（Exclusive-Pair）：一个TCP Connection，但是TCP Server只能接受一个连接，数据可以双向流动
- 请求回应模型（Request-Reply）：跟一对一结对模型的区别在于请求端可以是1~N个，该模型主要用于远程调用及任务分配等
- 发布订阅模型（Publish-Subscribe）：发布端单向分发数据，且不关心是否把全部信息发送给订阅端。如果发布端开始发布信息时，订阅端尚未连接上来，则这些信息会被直接丢弃。订阅端未连接导致信息丢失的问题，可以通过与请求回应模型组合来解决。订阅端只负责接收，而不能反馈，且在订阅端消费速度慢于发布端的情况下，会在订阅端堆积数据。该模型主要用于数据分发。天气预报、微博明星粉丝可以应用这种经典模型。
- 推拉模型（Push-Pull）：Server端作为Push端，而Client端作为Pull端，如果有多个Client端同时连接到Server端，则Server端会在内部做一个负载均衡，采用平均分配的算法，将所有消息均衡发布到Client端上。与发布订阅模型相比，推拉模型在没有消费者的情况下，发布的消息不会被消耗掉；在消费者能力不够的情况下，能够提供多消费者并行消费解决方案。该模型主要用于多任务并行。
这4种模型总结出了通用的网络通信模型，在实际中可以根据应用需要，组合其中的2种或多种模型来形成自己的解决方案。
请求回应模型
发布订阅模型
推拉模型
传输协议
ZeroMQ支持四类传输协议。每种传输协议由地址字符串来定义，该字符串由两部分组成：transport://endpoint。传输（transport） 部分指定了所使用的底层传输协议，端点（endpoint） 部分的格式则随着使用的协议而有所不同，具体如下：
- TCP (tcp://hostname:port): 在主机之间进行通讯
- INROC (inproc://name): 在同一进程的线程之间进行通讯（线程间）
- IPC (ipc:///tmp/filename): 同一主机的进程之间进行通讯
- PGM (pgm://interface;address:port 和 epgm://interface;address:port): 多播通讯
二、NetMQ
NetMQ 也是一个社区开源项目，网站在Github上 https://github.com/zeromq/netmq, 可以通过Nuget包获取。
下面以NetMQ的一对一模式为例，引入3.3.0.7 NetMQ
服务端：
class Program
{
static void Main(string[] args)
{
using (NetMQContext context = NetMQContext.Create())
{
Server(context);
}
} static void Server(NetMQContext context)
{
using (NetMQSocket serverSocket = context.CreateResponseSocket())
{
serverSocket.Bind("tcp://*:5555");
Console.WriteLine("Waiting for connetion...");
while (true)
{
string message = serverSocket.ReceiveString();
Console.WriteLine("Receive message {0}", message); string sendMessage = "Hello " + message;
serverSocket.Send(sendMessage);
Console.WriteLine("Send message {0}", sendMessage); if (message == "exit")
{
break;
}
}
}
} }
客户端：
class Program
{
static void Main(string[] args)
{
using (NetMQContext context = NetMQContext.Create())
{
Client(context);
}
}
static void Client(NetMQContext context)
{
using (NetMQSocket clientSocket = context.CreateRequestSocket())
{
clientSocket.Connect("tcp://127.0.0.1:5555"); while (true)
{
Console.WriteLine("Please enter your message:");
string message = Console.ReadLine();
clientSocket.Send(message); string answer = clientSocket.ReceiveString();
Console.WriteLine("Answer from server: {0}", answer); if (message == "exit")
{
break;
}
}
}
} }
更多参考：通过 C# 使用 ZeroMQ （一） ZeroMQ 通讯模式 pdf格式：通过C#使用ZeroMQ （n452）
性能测试
//千万级别的数据入列测试。对4种数据大小（200B、2K、10K、20K）对比测试
class TestPerformance
{
static string msg200B = "";
static string msg2K = "";
static string msg10K = "";
static string msg20K = "";
//存储 运行的时间
static List<long> msg200BResult = new List<long>() { };
static List<long> msg2KResult = new List<long>() { };
static List<long> msg10KResult = new List<long>() { };
static List<long> msg20KResult = new List<long>() { }; /// <summary>
/// 初始化
/// </summary>
static void Init()
{
StringBuilder temp = new StringBuilder();
Random ran = new Random(Guid.NewGuid().GetHashCode());
for (int i = ; i < * ; i++)
{
temp.Append((char)ran.Next(, ));
if (i == ) //一个Char两个字节；
msg200B = temp.ToString();
if (i == )
msg2K = temp.ToString();
if (i == * )
msg10K = temp.ToString();
}
msg20K = temp.ToString();
} /// <summary>
/// 输出结果
/// </summary>
public static void MainTest()
{
Init();
//10次测试取平均值
for (int i = ; i < ; i++)
{
//long deq = 0;
msg200BResult.Add(InQueue(msg200B));
msg2KResult.Add(InQueue(msg2K));
msg10KResult.Add(InQueue(msg10K));
msg20KResult.Add(InQueue(msg20K));
}
Console.WriteLine("200B inqueue 1000W, ElapsedMilliseconds:{0}", msg200BResult.Average());
Console.WriteLine("msg2K inqueue 1000W, ElapsedMilliseconds:{0}", msg2KResult.Average());
Console.WriteLine("msg10K inqueue 1000W, ElapsedMilliseconds:{0}", msg10KResult.Average());
Console.WriteLine("msg20K inqueue 1000W, ElapsedMilliseconds:{0}", msg20KResult.Average());
Console.ReadLine();
} /// <summary>
/// 往队列发布数据
/// </summary>
/// <param name="msg"></param>
/// <returns></returns>
static long InQueue(string msg)
{
ManualResetEvent rest = new ManualResetEvent(false);
using (NetMQContext context = NetMQContext.Create())
using (var pub = context.CreatePublisherSocket())
{
pub.Bind("tcp://127.0.0.1:9991"); long enElapse = ;
ThreadPool.QueueUserWorkItem((o) =>
{
Stopwatch sw = new Stopwatch();
Random ran = new Random(Guid.NewGuid().GetHashCode());
sw.Start();
//long enCount = 0;
for (int i = ; i < * ;i++) //1000W ;
{
if (i % == )
Console.Write("."); pub.SendMore("AAA");
pub.Send(msg + ran.Next(-, ));//追加随机数避免字符串消息内容相同；
}
sw.Stop();
Console.WriteLine("\n Has InQueue 1000W Mesage.Length={1} , ElapsedMilliseconds:{0}\n", sw.ElapsedMilliseconds, msg.Length);
enElapse = sw.ElapsedMilliseconds;
rest.Set();
});
rest.WaitOne();
return enElapse;
} }
}
其他几种模式
各个构件逐个介绍下：
1. RequestSocket：经典的请求Socket构件，一般和ResponseSocket一起组合成请求应答模式。
2. ResponseSocket：请求应答中的应答方，中间可以加入XPublishSocket,RouterSocket等扩展最终到达RequestSocket。
3. RouterSocket、DealerSocket： 当需要保证请求应答模式中可扩展性时需要在两者之间添加一个中间方隔离两端的耦合。这时候就需要RouterSocket+DealerSocket组 合。RouterSocket负责连接RequestSocket，DealerSocket则负责Response的一头
4. PublisherSocket：发布订阅中的发布方。注意由于ZeroMQ的高效，注意尽量让订阅方先启动，保证不丢失消息。
5. SubscriberSocket：发布订阅模式中的订阅方，注意由于发布订阅模式实际是在订阅方做消息筛选的，所有实际上订阅方将接收所有的发布消息再更加自己的订阅清理不需要的。
6. XSubscriberSocket、XPublisherSocket：可能您的发布订阅又是会需要跨网络的广播，这时候您需要在另一个网络中有一个代理，XSubscriberSocket + XPublisherSocket就是为此而生的，XSubscriberSocket负责承上，XPublisherSocke负责承上。
7. PairSocket：当你的一个任务需要跨线程、跨进程甚至跨服务器时就会用到PairSocket模式，它可以在自己任务启动线程指向第一步的队列，然后等待最后一步所在的队列返回结果即可，开始和结束队列直接可以有多个步骤队列，以流水线的方式连接再一起工作。
8. PushSocket：当你不满足于PariSocket只能单线管道模式之下时，你会用到推拉模式，这种模式允许你在任任务流水线的任一环节做并行处理，并在并行后的下一环节归拢整理结果。
9. Pullsocket：推拉模式中的拉的一方。
请求应答模式：
private static void Main(string[] args)
{
using (NetMQContext ctx = NetMQContext.Create())
{
ThreadPool.QueueUserWorkItem((o) =>
{
using (var server = ctx.CreateResponseSocket())
{
server.Bind("tcp://127.0.0.1:5556");
//server.Monitor("", NetMQ.zmq.SocketEvent.All);//必要时可以将队列延迟，阻塞等事件发往Monitor队列；
while (true)
{
string m1 = server.ReceiveString();
Console.WriteLine("From Client: {0}", m1);
server.Send("Hi Back");
}
}
});
using (var client = ctx.CreateRequestSocket())
{
client.Connect("tcp://127.0.0.1:5556");
client.Send("Hello");
string m2 = client.ReceiveString();
Console.WriteLine("From Server: {0}", m2);
Thread.Sleep();
client.Send("Word");
m2 = client.ReceiveString();
Console.WriteLine("From Server: {0}", m2);
Console.ReadLine();
}
Console.Read();
}
}
发布订阅模式：
//发布端；
using (NetMQContext context = NetMQContext.Create())
using (var pub = context.CreatePublisherSocket())
{
pub.Bind("tcp://127.0.0.1:9991");
ThreadPool.QueueUserWorkItem((o) =>
{
Stopwatch sw = new Stopwatch();
Random ran = new Random(Guid.NewGuid().GetHashCode());
sw.Start(); for(int i = ; i < ; i++) //1000W ;
{
if (enCount % == )
Console.Write(".");
//pub.SendMore("AAA");//必要时可以给消息加个名称为“AAA”的信封，这样订阅端可以有选择的接受消息；
pub.Send(msg + ran.Next(-, ));//追加随机数放在消息内容字符串一样；
}
sw.Stop();
Console.WriteLine("\n Has InQueue 1000W Mesage, ElapsedMilliseconds:{0}\n", sw.ElapsedMilliseconds);
}); //订阅端
using (NetMQContext context = NetMQContext.Create())
using (var sub = context.CreateSubscriberSocket())
{
sub.Connect("tcp://127.0.0.1:9991");
sub.Subscribe("");//空字符串表示订阅所有，仅订阅“AAA”：sub.Subscribe("AAA");这时第一次ReceiveString()将返回“AAA”，之后才是真正的消息。
while (true)
{
var msg = sub.ReceiveString();//接收消息；
Console.WriteLine("msg:{0}", msg);
}
}
推拉模式：
static void Main(string[] args)
{
using (var ctx = NetMQContext.Create())
{
ThreadPool.QueueUserWorkItem((o) =>
{
//ventilator
using (var ventilator = ctx.CreatePushSocket())
{
ventilator.Bind("tcp://127.0.0.1:9992");
Thread.Sleep();
ventilator.SendMore("A");
ventilator.Send("#InputInfo#");
//sink
using (var sink = ctx.CreatePullSocket())
{
sink.Bind("tcp://127.0.0.1:9993");
while (true)
{
var result = sink.Receive();
Console.WriteLine(Formate(result));
}
}
}
});
//worker;
ThreadPool.QueueUserWorkItem((o) =>
{
using (var ctxWorker = NetMQContext.Create())
{
var recv = ctxWorker.CreatePullSocket();
recv.Connect("tcp://127.0.0.1:9992");
var send = ctxWorker.CreatePushSocket();
send.Connect("tcp://127.0.0.1:9993");
while (true)
{
var input = recv.Receive();
Console.WriteLine("Input:{0}", Formate(input));
Thread.Sleep();// do work;
send.Send(string.Format("Worker {2} Input:{0},Output:{1}", Formate(input), "*****", id));
}
}
});
Console.Read();
}
}
static string Formate(byte[] input)
{
return System.Text.Encoding.Default.GetString(input);
}
参考：初识 ZeroMQ
最新文章
- Rocksdb Compaction原理
- MAC下安装与配置MySQL
- mysqldump 逻辑备份的正确姿势
- Ninject.MVC 知识点记录
- sql 关联查询
- Building and setting up QT environment for BeagleBone
- 在Linux下不使用密码远程登陆其他Linux
- 第四届CCF大数据学术会议征文通知
- 基于css制作轮播图的部分效果
- Libgdx开发ios游戏
- java 非缓冲与缓冲数据读取比较
- 根据矩阵变化实现基于 HTML5 的 WebGL 3D 自动布局
- Spring boot集成swagger2
- Oracle day04 DML_事务_序列_视图_数据类型_DDL
- C# .NET MD5 HASH
- 【M2】软件工程终期总结报告——前端设计总结
- 【github&&git】1、github中的watch、star、fork的作用
- iview,用render函数渲染
- poj1942 Paths on a Grid（无mod大组合数）
- Python 迭代对象、迭代器、生成器