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的运用详解的细致内容，更多请关注ki4网别的相干文章！
