工业物联网或系统集成中应用消息队列（ActiveMQ，C#的demo）的场景全面分析

2022-01-06 03:33:11

目录

工业物联网或系统集成中应用消息队列（ActiveMQ）的场景全面分析... 1

前言... 1

第一章终端/交互场景... 3

1.1 终端设备... 3

1.2 通讯机制... 3

第二章 ActvieMQ应用场景... 4

2.1 发布/订阅（Publish/Subscribe）... 4

2.2 生产者/消费者（Producer/Consumer）... 7

2.3 请求/应答（Request/Response）... 10

第三章假定场景分析... 16

3.1 通讯层... 16

3.2 数据业务层... 16

3.3 综述... 16

前言

互联网技术已经发展的很成熟了，各种开源的代码、框架和解决方案等。鉴于互联网技术的通用性，势必向其他领域延展。不管是工业4.0，还是互联网+ 工业，互联网技术向工业领域传导也是必然的。

所以，对于工业方面的应用场景的技术储备和技术线路调研也是日常工作很重要的一部分，为公司的横向和纵向发展提供技术平台和保障，当然也取决于领导的视野。

第一章终端/交互场景

任何技术都是为业务服务，而业务是有特定的应用场景。离开了实现环境去谈技术是没有实际意义的，解决实际问题而又能保证相当长时间内的稳定性是我们努力实现的目标。同时要从多个角度来考虑问题，以及做出平衡。

1.1 终端设备

（1）终端种类：嵌入式硬件/传感器、PC机（监测站、大型监控设备等）、手机终端等。

（2）交互方式：单向交互，数据上传，可能服务端会有返回确认信息，证明数据已经收到了；双向交互，服务端不仅仅会返回确认信息，同时还要主动下发给指定终端命令信息，例如：控制硬件设备机械动作命令、修改硬件设备参数命令、以及补传相关数据信息命令等等。

（3）设备管理：这里指的设备管理是说设备的状态，包括两个方面：设备IO状态和设备通讯状态。设备IO状态包括：IO打开和IO关闭。设备通讯状态包括：通讯中断、通讯干扰和通讯正常。为了判断故障，这里的逻辑关系是：IO打开的时候不一定代表通讯正常；IO关闭不一定代表通讯中断；通讯中断不一定代表IO关闭；通讯干扰不一定代表IO打开。

（4）数据完整性：允许数据缺失，一般在原来数据基础上的增量数据是可以允许丢失的；不允许数据缺失，一般脉冲数据是不允许数据丢失的。

1.2 通讯机制

（1）主动请求数据：服务器端主动下发命令给终端，让谁上传数据、上传什么数据都由服务器端决定。

（2）被动接收数据：服务器端被动接收终端上传的数据，根据数据信息进行数据处理，以及返回确认信息。

第二章 ActvieMQ应用场景

消息队列比较多，本文以ActiveMQ为例进行介绍，全部代码实现C#为主，主要考虑到常见的应用模式。事例代码下载：http://pan.baidu.com/s/1qXZ1sU4。

2.1 发布/订阅（Publish/Subscribe）

一个信息发布者在某一个主题上发布消息，所有订阅该主题的订阅都会收到相同的消息，这种模式是一对多的关系，如下图：

发布端代码：

static void Main(string[] args)
        {
            try
            {
                IConnectionFactory factory = new ConnectionFactory("tcp://localhost:61616/");
                using (IConnection connection = factory.CreateConnection())
                {
                    using (ISession session = connection.CreateSession())
                    {
                        IMessageProducer prod = session.CreateProducer(new Apache.NMS.ActiveMQ.Commands.ActiveMQTopic("Topic"));

                        string text = Console.ReadLine();
                        while (text!="exit")
                        {
                            ITextMessage msg = prod.CreateTextMessage();
                            msg.Text = text;
                            prod.Send(msg, Apache.NMS.MsgDeliveryMode.NonPersistent, Apache.NMS.MsgPriority.Normal, TimeSpan.MinValue);
                            Console.WriteLine("Sending: " + text);
                            System.Threading.Thread.Sleep(2000);
                        }
                    }
                }
                Console.ReadLine();
            }
            catch (System.Exception e)
            {
                Console.WriteLine("{0}", e.Message);
                Console.ReadLine();
            }
        }

订阅端代码：

static void Main(string[] args)
        {
            try
            {
                IConnectionFactory factory = new ConnectionFactory("tcp://localhost:61616/");
                using (IConnection connection = factory.CreateConnection())
                {
                    connection.ClientId = "testing listener1";
                    connection.Start();

                    using (ISession session = connection.CreateSession())
                    {
                        IMessageConsumer consumer = session.CreateDurableConsumer(new Apache.NMS.ActiveMQ.Commands.ActiveMQTopic("Topic"), "testing listener1", null, false);
                        consumer.Listener += new MessageListener(consumer_Listener);
                        Console.ReadLine();
                    }
                    connection.Stop();
                    connection.Close();
                }
            }
            catch (System.Exception e)
            {
                Console.WriteLine(e.Message);
                Console.ReadLine();
            }
        }

        static void consumer_Listener(IMessage message)
        {
            try
            {
                ITextMessage msg = (ITextMessage)message;
                Console.WriteLine("Receive: " + msg.Text);
            }
            catch (System.Exception e)
            {
                Console.WriteLine(e.Message);
            }
        }

2.2 生产者/消费者（Producer/Consumer）

生产者生产了一块香皂，消费者购买了该块香皂，使用完了，就在这个世界上消息了，生产者和消费者之间的关系存在一种偶然性，这是一对一的关系，如下图：

生产端代码：

 static void Main(string[] args)
        {
            try
            {
                IConnectionFactory factory = new ConnectionFactory("tcp://localhost:61616/");
                using (IConnection connection = factory.CreateConnection())
                {
                    using (ISession session = connection.CreateSession())
                    {
                        IMessageProducer prod = session.CreateProducer(new Apache.NMS.ActiveMQ.Commands.ActiveMQQueue("Queue"));

                        string text = Console.ReadLine();
                        while (text != "exit")
                        {
                            ITextMessage msg = prod.CreateTextMessage();
                            msg.Text = text;
                            prod.Send(msg, Apache.NMS.MsgDeliveryMode.NonPersistent, Apache.NMS.MsgPriority.Normal, TimeSpan.MinValue);
                            Console.WriteLine("Sending: " + text);
                            System.Threading.Thread.Sleep(2000);
                        }
                    }
                }
                Console.ReadLine();
            }
            catch (System.Exception e)
            {
                Console.WriteLine("{0}", e.Message);
                Console.ReadLine();
            }
        }

消费端代码：

static void Main(string[] args)
        {
            try
            {
                IConnectionFactory factory = new ConnectionFactory("tcp://localhost:61616/");
                using (IConnection connection = factory.CreateConnection())
                {
                    //connection.ClientId = "testing listener2";
                    connection.Start(); 
                    using (ISession session = connection.CreateSession())
                    {
                        IMessageConsumer consumer = session.CreateConsumer(new Apache.NMS.ActiveMQ.Commands.ActiveMQQueue("Queue"));
                        consumer.Listener += new MessageListener(consumer_Listener);
                        Console.ReadLine();
                    }
                    connection.Stop();
                    connection.Close();
                }
            }
            catch (System.Exception e)
            {
                Console.WriteLine(e.Message);
                Console.ReadLine();
            }
        }

        static void consumer_Listener(IMessage message)
        {
            try
            {
                ITextMessage msg = (ITextMessage)message;
                Console.WriteLine("Receive: " + msg.Text);
            }
            catch (System.Exception e)
            {
                Console.WriteLine(e.Message);
            }
        }

2.3 请求/应答（Request/Response）

请求-应答的通信方式应用很普遍，客户端向服务端上传实时数据或参数，服务端处理完之后，要返回确认信息，这种交互关系如下图：

客户端代码：

static void Main(string[] args)
        {
            IConnectionFactory factory = new ConnectionFactory("tcp://localhost:61616/");
            try
            {
                using (IConnection connection = factory.CreateConnection())
                {
                    connection.Start();
                    using (ISession session = connection.CreateSession(AcknowledgementMode.AutoAcknowledge))
                    {
                        IDestination destination =  new Apache.NMS.ActiveMQ.Commands.ActiveMQQueue("client.messages");

                        IMessageProducer producer = session.CreateProducer(destination);
                        producer.DeliveryMode=MsgDeliveryMode.NonPersistent;

                        IDestination tempDest = session.CreateTemporaryQueue();
                        IMessageConsumer responseConsumer = session.CreateConsumer(tempDest);
                        responseConsumer.Listener += new MessageListener(consumer_Listener);
                       
                        string text = Console.ReadLine();
                        while (text != "exit")
                        {
                            ITextMessage msg = session.CreateTextMessage();
                            msg.Text = text;
                            msg.NMSReplyTo = tempDest;
                            msg.NMSCorrelationID = DateTime.Now.ToString("yyyyMMddHHmmss");
                            producer.Send(msg, Apache.NMS.MsgDeliveryMode.NonPersistent, Apache.NMS.MsgPriority.Normal, TimeSpan.MinValue);
                            Console.WriteLine("Sending: " + text);
                            System.Threading.Thread.Sleep(2000);
                        }

                        Console.ReadLine();
                    }
                    connection.Stop();
                    connection.Close();
                }
            }
            catch (Exception ex)
            {
                Console.WriteLine(ex.Message);
                Console.ReadLine();
            }
        }

        static void consumer_Listener(IMessage message)
        {
            try
            {
                ITextMessage msg = (ITextMessage)message;
                Console.WriteLine("Receive: " + msg.Text);
            }
            catch (System.Exception e)
            {
                Console.WriteLine(e.Message);
            }
        }

服务端代码：

 private static ISession session;

        private static IMessageProducer replyProducer;
        static void Main(string[] args)
        {
            IConnectionFactory factory = new ConnectionFactory("tcp://localhost:61616/");
            try
            {
                    IConnection connection = factory.CreateConnection();
                    connection.Start();
                    session = connection.CreateSession(AcknowledgementMode.AutoAcknowledge);

                    IDestination adminQueue = new Apache.NMS.ActiveMQ.Commands.ActiveMQQueue("client.messages");
                    replyProducer = session.CreateProducer();
                    replyProducer.DeliveryMode=MsgDeliveryMode.NonPersistent;

                    IMessageConsumer consumer = session.CreateConsumer(adminQueue);
                    consumer.Listener += new MessageListener(consumer_Listener);
                Console.ReadLine();
            }
            catch (Exception ex)
            {
                Console.WriteLine(ex.Message);
                Console.ReadLine();
            }
        }

        static void consumer_Listener(IMessage message)
        {
            try
            {
                ITextMessage response = session.CreateTextMessage();
                if (message is ITextMessage) {
                    ITextMessage txtMsg = (ITextMessage)message;
                    string messageText = txtMsg.Text;
                    response.Text = messageText;

                    Console.WriteLine("Receive:" + messageText);
                }

                response.NMSCorrelationID=message.NMSCorrelationID;

                replyProducer.Send(message.NMSReplyTo, response);
            }
            catch (Exception e)
            {
                Console.WriteLine(e.Message);

            }
        }

第三章假定场景分析

我们以系统建设过程中的构架来分析消息队列在两个层面的问题，通讯层和数据业务层。

3.1 通讯层

通讯层是否可以用消息队列（ActiveMQ）？这个问题取决于两方面：1、如果终端设备有嵌入式硬件，甚至还是C51开发的，那么在系统集成和物联的过程中，就涉及到兼容性的问题。显然和消息队列进行对接是一件头痛的事，用C51写一个对接的驱动不是不可能，但是要评估工作量和稳定性。2、服务端与指定某个终端双向交互频繁的情况，特别是服务端实时发送设备校准命令的情况，这种情况消息队列是不如通讯框架的。

3.2 数据业务层

服务端接收到数据后，完全可以使用消息队列的生产者和消费者模式处理数据，对系统的业务进行解耦。

下发命令也可以通过消息队列，这样可以统一控制端的接口，再由通讯框架下发到指定的终端。

3.3 综述

综合考虑，建议在通讯层使用通讯框架，对于设备的IO状态和通讯状态能够及时反应，通讯效率也是能够得到保障的；对于数据业务层，建议不要放在通讯框架内部进行处理，可以使用消息队列，配合通讯框架使用。

整体架构图如下：

文章得到了群友支持：

码农公寓

第一章 终端/交互场景