NetMQ发布订阅C#示例

NetMQ (ZeroMQ to .Net),ØMQ号称史上最快中间件。它对socket通信进行了封装,使得我们不需要写socket函数调用就能完成复杂的网络通信。和一般意义上的消息队列产品不同的是,它没有消息队列服务器,而更像是一个网络通信库。从网络通信的角度看,它处于会话层之上,应用层之下。【ZeroMQ 官网】:http://zeromq.org

ØMQ有4个基本通信模型:分别是一对一结对模型(Exclusive-Pair)、请求回应模型(Request-Reply)、发布订阅模型(Publish-Subscribe)、推拉模型(Push-Pull)。

Request-reply pattern 请求-回复模型

  • 这种模型主要用于从客户端向一个或多个服务实例发送请求,然后等待紧接着对于每个请求的回复
  • 里面又具体分了ZMQ_REQ ZMQ_REP ZMQ_DEALER ZMQ_ROUTER
  • REQ 发送完消息后,必须接收一个回应消息后,才能发送新的消息
  • REP当接收消息时,都会返回一个消息

Publish-subscribe pattern 发布-订阅模式

  • 这种模式主要用于1对多的数据发布(一个发布者,多个订阅者)
  • 里面又具体分了ZMQ_PUB ZMQ_SUB
  • PUB发送消息给所有的SUB。如果此时SUB没有启动,下次启动时会漏掉该消息

Pipeline pattern 管道模式

  • 这种模式主要用于发布数据到由管道排列的节点上面,数据总是沿着管道流动。每个管道阶段连接了至少一个节点
  • 里面又具体分了ZMQ_PUSH ZMQ_PULL
  • 一个1对N队列的实现,PUSH将数据放入队列,PULL从队列中不取出数据。数据会负载均衡的发送给每一个PULL

Exclusive pair pattern 独立对模式

  • peer to peer 模式。主要用于进程内部线程间通信
  • 里面又具体分了ZMQ_PAIR
  • 线程间1-to-1队列的实现,采用了lock free实现,所以速度很快

下面是订阅发布的示例代码:

发布服务端:

  public static class NetMQPub
    {
        readonly static ManualResetEvent _terminateEvent = new ManualResetEvent(false);
        /// <summary>
        /// NetMQ 发布模式
        /// </summary>
        public static void Start()
        {
            string[] wethers = new string[5] {"晴朗","多云","阴天","小雨","暴雪" };

            //CTRL+C 退出程序
            Console.CancelKeyPress += Console_CancelKeyPress;
            Console.WriteLine("发布多个地区天气预报:");

            using (var context = NetMQContext.Create())
            {
                using (var publisher = context.CreatePublisherSocket())
                {
                    publisher.Bind("tcp://127.0.0.1:5556");

                    var rng = new Random();
                    string msg;
                    int sleeptime = 10;

                    while (_terminateEvent.WaitOne(0) == false)
                    {
                        //随机生成天气数据
                        int zipcode = rng.Next(0, 99);
                        int temperature = rng.Next(-50, 50);
                        int wetherId = rng.Next(0, 4);

                        msg = string.Format("{0} {1} {2}", zipcode, temperature, wethers[wetherId]);
                        publisher.Send(msg,Encoding.UTF8, zmq.SendReceiveOptions.DontWait);

                        Console.WriteLine(msg);
                        Thread.Sleep(sleeptime);
                    }
                }
            }
        }

        static void Console_CancelKeyPress(object sender, ConsoleCancelEventArgs e)
        {
            Console.WriteLine("exit...");
            _terminateEvent.Set();
        }
    }

  订阅客户端,可启动多个实例来模拟接收天气信息:

  public static class NetMQSub
    {
        public delegate void GetDataHandler(string message);
        public static event GetDataHandler OnGetData;

        /// <summary>
        /// NetMQ 订阅模式
        /// </summary>
        public static void Start()
        {
            var rng = new Random();
            int zipcode = rng.Next(0, 99);
            Console.WriteLine("接收本地天气预报 {0}...", zipcode);

            OnGetData += new GetDataHandler(ProcessData);

            using (var context = NetMQContext.Create())
            using (var subscriber = context.CreateSubscriberSocket())
            {
                subscriber.Connect("tcp://127.0.0.1:5556");
                subscriber.Subscribe(zipcode.ToString(CultureInfo.InvariantCulture));

                while(true)
                {
                    string results = subscriber.ReceiveString(Encoding.UTF8);
                    Console.Write(".");

                    string[] split = results.Split(new[] { ‘ ‘ }, StringSplitOptions.RemoveEmptyEntries);

                    int zip = int.Parse(split[0]);
                    if (zip == zipcode)
                    {
                        OnGetData(results);
                    }
                }
            }
        }

        public static void ProcessData(string msg)
        {
            Console.WriteLine("天气情况:" + msg);
        }
    }

  

时间: 2024-10-10 10:57:21

NetMQ发布订阅C#示例的相关文章

C# Redis系列(三)-Redis发布订阅及客户端编程

发布订阅模型 Redis中的发布订阅 客户端编程示例 0.3版本Hredis 发布订阅模型 在应用级其作用是为了减少依赖关系,通常也叫观察者模式.主要是把耦合点单独抽离出来作为第三方,隔离易变化的发送方和接收方. 发送方:只负责向第三方发送消息.(杂志社把读者杂志交给邮局) 接收方:被动接收消息.(1:向邮局订阅读者杂志,2:门口去接邮过来的杂志) 第三方作用是:存储订阅杂志的接收方,并在杂志过来时送给接收方. (邮局) C#示例,发送方把杂志放到邮局里面: if (QA.AddBug()) E

JavaScript发布/订阅实例

原文链接: Pub/Sub JavaScript Object原文日期: 2014年6月11日翻译日期: 2014年6月13日 翻译人员: 铁锚 高效AJAX网站的三大杀器: 事件代理, 浏览历史管理, 以及高效应用级 发布/订阅通信机制. 本博客的原文站点 同时使用了这三种技术,本文中作者将分享其中最简单的一个: 该网站使用的 一个微型 发布/订阅模块. 如果你不了解 发布/订阅 模式,那么可以将其类比为 你发表了一篇博文,所有人都可以订阅你的博客, 也类似于广播电台的工作方式: 有一个站台进

MQTT的学习之Mosquitto发布-订阅(2)

在<MQTT的学习之Mosquitto安装&使用(1)>一文末尾,我已经模拟了发布-订阅模式,只是那时在服务器直接模拟的,并不是java代码模拟的.下面贴出Java代码 1.首先引入依赖包: <!-- mosquitto依赖 --> <dependency> <groupId>org.eclipse.paho</groupId> <artifactId>org.eclipse.paho.client.mqttv3</ar

redis的发布订阅模式

概要 redis的每个server实例都维护着一个保存服务器状态的redisServer结构 struct redisServer { /* Pubsub */ // 字典,键为频道,值为链表 // 链表中保存了所有订阅某个频道的客户端 // 新客户端总是被添加到链表的表尾 dict *pubsub_channels;  /* Map channels to list of subscribed clients */ // 这个链表记录了客户端订阅的所有模式的名字 list *pubsub_pa

ActiveMQ发布-订阅消息模式

一.订阅杂志我们很多人都订过杂志,其过程很简单.只要告诉邮局我们所要订的杂志名.投递的地址,付了钱就OK.出版社定期会将出版的杂志交给邮局,邮局会根据订阅的列表,将杂志送达消费者手中.这样我们就可以看到每一期精彩的杂志了. 仔细思考一下订杂志的过程,我们会发现这样几个特点:1.消费者订杂志不需要直接找出版社:2.出版社只需要把杂志交给邮局:3.邮局将杂志送达消费者.邮局在整个过程中扮演了非常重要的中转作用,在出版社和消费者相互不需要知道对方的情况下,邮局完成了杂志的投递. 二. 发布-订阅消息模

发布/订阅消息传送模型

1.发布/订阅模型概览 发布/订阅(publish-and-subscribe)模型通常被简写为pub/sub模型.在这个模型中,消息生产者成为发布者(publisher),而消息消费者则称为订阅者(subscribe).在点对点模型中,是将消息发送到一个队列中,而发布/订阅模型则是将消息发布给一个主题.发布/订阅模型最重要的特性如下: 消息通过一个称为主题的虚拟通道进行交换. 每条消息都会传送给称为订阅者的多个消息消费者.订阅者有许多类型,包括持久性.非持久性和动态性. 发布者通常不会知道.也

Redis教程03——Redis 发布/订阅(Pub/Sub)

Pub/Sub 订阅,取消订阅和发布实现了发布/订阅消息范式(引自wikipedia),发送者(发布者)不是计划发送消息给特定的接收者(订阅者).而是发布的消息分到不同的频道,不需要知道什么样的订阅者订阅.订阅者对一个或多个频道感兴趣,只需接收感兴趣的消息,不需要知道什么样的发布者发布的.这种发布者和订阅者的解耦合可以带来更大的扩展性和更加动态的网络拓扑. F为了订阅foo和bar,客户端发出一个订阅的频道名称: SUBSCRIBE foo bar 其他客户端发到这些频道的消息将会被推送到所有订

RabbitMQ(三) ——发布订阅

RabbitMQ(三) --发布订阅 (转载请附上本文链接--linhxx) 一.概述 RabbitMQ的发布订阅(Publish/Subscribe),其将生产者和消费者进一步解耦,生产者生产消息后,交付给交换机,消费者上线后,主动主动去队列中取数据进行处理.该模式也符合上一节工作队列中的ack.预取等规则. 发布订阅模式如下图所示: 二.交换机(exchange) 生产者生产完消息之后,都是将消息通过channel交给交换机,即生产者并不直接和队列联系.在没有定义交换机的时候,RabbitM

redis(3)发布订阅

一.发布/订阅模式 在软件工程里面,发布/订阅是一种消息模式,这种模式旨在将消息发送者和消息接收者解耦.发送者不需要关心将消息发送给谁,接收者也不需要知道消息的发送者是谁.发送者将消息发布以后就结束动作,接收者可以订阅自己感兴趣的消息. 除了发布/订阅模式还有一种和它很类似的,消息队列,是一种典型的面向消息中间件的系统.许多消息系统都会同时支持发布/订阅和消息队列模型,例如Java Message Service(JMS) 参见:维基百科 二.redis的发布/订阅 我们从一个简单的示例开始,首