返回首页 | 申博娱乐APP下载

合作共赢、快速高效、优质的网站建设提供商

更多精品源码-尽在织梦模板-www.moke8.com

C#中关于ActiveMQ的使用详解

时间:2017-09-22 编辑:admin

ActiveMQ是个好东东,不用多说。ActiveMQ供给多种语言支撑,如Java, C, C++, C#, Ruby, Perl, Python, PHP等。因为我在windows下开发GUI,比较关心C++和C#,其间C#的ActiveMQ很简略,Apache供给NMS(.Net Messaging Service)支撑.Net开发,只需如下几个过程即能树立简略的完成。C++的使用相对费事些,稍后写文章介绍。

1、去ActiveMQ官方网站下载最新版的ActiveMQ,我之前下的是5.3.1,5.3.2现在也现已出来了。

2、去ActiveMQ官方网站下载最新版的Apache.NMS,需求下载Apache.NMS和Apache.NMS.ActiveMQ两个bin包,如果对源码感兴趣,也可下载src包。这儿要提示一下,如果下载1.2.0版别的NMS.ActiveMQ,Apache.NMS.ActiveMQ.dll在实际使用中有个bug,即中止ActiveMQ使用时会抛WaitOne函数反常,检查src包中的源码发现是因为Apache.NMS.ActiveMQ-1.2.0-src\src\main\csharp\Transport\InactivityMonitor.cs中的如下代码形成的,修正一下源码从头编译即可。看了一下最新版1.3.0现已修正了这个bug,因而下载最新版即可。

private void StopMonitorThreads()   
        {   
            lock(monitor)   
            {   
                if(monitorStarted.CompareAndSet(true, false))   
                {   
                    AutoResetEvent shutdownEvent = new AutoResetEvent(false);   
                    // Attempt to wait for the Timers to shutdown, but don't wait   
                    // forever, if they don't shutdown after two seconds, just quit.   
                    this.readCheckTimer.Dispose(shutdownEvent);   
                    shutdownEvent.WaitOne(TimeSpan.FromMilliseconds(2000));   
                    this.writeCheckTimer.Dispose(shutdownEvent);   
                    shutdownEvent.WaitOne(TimeSpan.FromMilliseconds(2000));   
                                                    //WaitOne的界说:public virtual bool WaitOne(TimeSpan timeout,bool exitContext)   
                    this.asyncTasks.Shutdown();   
                    this.asyncTasks = null;   
                    this.asyncWriteTask = null;   
                    this.asyncErrorTask = null;   
                }   
            }   
        }  
     private void StopMonitorThreads() 
        { 
            lock(monitor) 
            { 
                if(monitorStarted.CompareAndSet(true, false)) 
                { 
                    AutoResetEvent shutdownEvent = new AutoResetEvent(false);

                    // Attempt to wait for the Timers to shutdown, but don't wait 
                    // forever, if they don't shutdown after two seconds, just quit. 
                    this.readCheckTimer.Dispose(shutdownEvent); 
                    shutdownEvent.WaitOne(TimeSpan.FromMilliseconds(2000)); 
                    this.writeCheckTimer.Dispose(shutdownEvent); 
                    shutdownEvent.WaitOne(TimeSpan.FromMilliseconds(2000)); 
                                                    //WaitOne的界说:public virtual bool WaitOne(TimeSpan timeout,bool exitContext) 
                    this.asyncTasks.Shutdown(); 
                    this.asyncTasks = null; 
                    this.asyncWriteTask = null; 
                    this.asyncErrorTask = null; 
                } 
            } 
        }

3、运转ActiveMQ,找到ActiveMQ解压后的bin文件夹:...\apache-activemq-5.3.1\bin,履行activemq.bat批处理文件即可发动ActiveMQ服务器,默许端口为61616,这可在配置文件中修正。

4、写C#程序完成ActiveMQ的简略使用。新建C#工程(一个Producter项目和一个Consumer项目),WinForm或Console程序均可,这儿建的是Console工程,添加对Apache.NMS.dll和Apache.NMS.ActiveMQ.dll的引证,然后即可编写完成代码了,简略的Producer和Consumer完成代码如下:

producer:

using System;   
using System.Collections.Generic;   
using System.Text;   
using Apache.NMS;   
using Apache.NMS.ActiveMQ;   
using System.IO;   
using System.Xml.Serialization;   
using System.Runtime.Serialization.Formatters.Binary;   
namespace Publish   
{   
    class Program   
    {   
        static void Main(string[] args)   
        {   
            try  
            {   
                //Create the Connection Factory   
                IConnectionFactory factory = new ConnectionFactory("tcp://localhost:61616/");   
                using (IConnection connection = factory.CreateConnection())   
                {   
                    //Create the Session   
                    using (ISession session = connection.CreateSession())   
                    {   
                        //Create the Producer for the topic/queue   
                        IMessageProducer prod = session.CreateProducer(   
                            new Apache.NMS.ActiveMQ.Commands.ActiveMQTopic("testing"));   
                        //Send Messages   
                        int i = 0;   
                        while (!Console.KeyAvailable)   
                        {   
                            ITextMessage msg = prod.CreateTextMessage();   
                            msg.Text = i.ToString();   
                            Console.WriteLine("Sending: " + i.ToString());   
                            prod.Send(msg, Apache.NMS.MsgDeliveryMode.NonPersistent, Apache.NMS.MsgPriority.Normal, TimeSpan.MinValue);   
                            System.Threading.Thread.Sleep(5000);   
                            i++;   
                        }   
                    }   
                }   
                Console.ReadLine();   
           }   
            catch (System.Exception e)   
            {   
                Console.WriteLine("{0}",e.Message);   
                Console.ReadLine();   
            }   
        }   
    }   
}

consumer:

using System;   
using System.Collections.Generic;   
using System.Text;   
using Apache.NMS;   
using Apache.NMS.ActiveMQ;   
using System.IO;   
using System.Xml.Serialization;   
using System.Runtime.Serialization.Formatters.Binary;   
namespace Subscribe   
{   
    class Program   
    {   
        static void Main(string[] args)   
        {   
            try  
            {   
                //Create the Connection factory   
                IConnectionFactory factory = new ConnectionFactory("tcp://localhost:61616/");   
                //Create the connection   
                using (IConnection connection = factory.CreateConnection())   
                {   
                    connection.ClientId = "testing listener";   
                    connection.Start();   
                    //Create the Session   
                    using (ISession session = connection.CreateSession())   
                    {   
                        //Create the Consumer   
                        IMessageConsumer consumer = session.CreateDurableConsumer(new Apache.NMS.ActiveMQ.Commands.ActiveMQTopic("testing"), "testing listener", null, false);   
                        consumer.Listener += new MessageListener(consumer_Listener);   
                        Console.ReadLine();   
                    }   
                    connection.Stop();   
                    connection.Close();   
                }   
            }   
            catch (System.Exception e)   
            {   
                Console.WriteLine(e.Message);   
            }   
        }   
        static void consumer_Listener(IMessage message)   
        {   
            try  
            {   
                ITextMessage msg = (ITextMessage)message;   
                Console.WriteLine("Receive: " + msg.Text);   
           }   
            catch (System.Exception e)   
            {   
                Console.WriteLine(e.Message);   
            }   
        }   
    }   
}

程序完成的功用:生产者producer树立名为testing的主题,并每隔5秒向该主题发送音讯,消费者consumer订阅了testing主题,因而只需生产者发送testing主题的音讯到ActiveMQ服务器,服务器就将该音讯发送给订阅了testing主题的消费者。

编译生成producer.exe和consumer.exe,并履行两个exe,即可看到音讯的发送与接收了。

这个比方是建的主题(Topic),ActiveMQ还支撑另一种方法:Queue,即P2P,两者有什么差异呢?差异在于,Topic是播送,即如果某个Topic被多个消费者订阅,那么只需有音讯抵达服务器,服务器就将该音讯发给悉数的消费者;而Queue是点到点,即一个音讯只能发给一个消费者,如果某个Queue被多个消费者订阅,没有特殊情况的话音讯会一个一个地轮番发给不同的消费者,比方:

msg1-->consumer A

msg2-->consumer B

msg3-->consumer C

msg4-->consumer A

msg5-->consumer B

msg6-->consumer C

特殊情况是指:ActiveMQ支撑过滤机制,即生产者能够设置音讯的特点(Properties),该特点与消费者端的Selector对应,只要消费者设置的selector与音讯的Properties匹配,音讯才会发给该消费者。Topic和Queue都支撑Selector。

Properties和Selector该怎么设置呢?请看如下代码:

producer:

public void SetProperties()
{
ITextMessage msg = prod.CreateTextMessage();   
                            msg.Text = i.ToString();   
                            msg.Properties.SetString("myFilter", "test1");   
                            Console.WriteLine("Sending: " + i.ToString());   
                            prod.Send(msg, Apache.NMS.MsgDeliveryMode.NonPersistent, Apache.NMS.MsgPriority.Normal, TimeSpan.MinValue);  
ITextMessage msg = prod.CreateTextMessage(); 
                            msg.Text = i.ToString(); 
                            msg.Properties.SetString("myFilter", "test1"); 
                            Console.WriteLine("Sending: " + i.ToString()); 
                            prod.Send(msg, Apache.NMS.MsgDeliveryMode.NonPersistent, Apache.NMS.MsgPriority.Normal, TimeSpan.MinValue);

}

consumer:

public void SetSelector()
{
//生成consumer时经过参数设置Selector   
IMessageConsumer consumer = session.CreateConsumer(new Apache.NMS.ActiveMQ.Commands.ActiveMQQueue("testing"), "myFilter='test1'");  
//生成consumer时经过参数设置Selector 
IMessageConsumer consumer = session.CreateConsumer(new Apache.NMS.ActiveMQ.Commands.ActiveMQQueue("testing"), "myFilter='test1'");
}

以上就是C#中关于ActiveMQ的使用详解的具体内容,更多请重视其它相关文章!


浏览:

网站建设

流程

    网站建设流程