ActiveMQ在C#中的应用_bodybo的博客

相关文章推荐

完美的皮带 · （转载）jsp实现购物车 - ...· 1 周前 ·

鬼畜的可乐 · Java邮件发送解决ssl ...· 1 周前 ·

玩命的猴子 · Camera ...· 6 天前 ·

重情义的冲锋衣 · JSR 356 (Java API for ...· 5 天前 ·

快乐的哑铃 · 南京地铁14号线最新规划曝光· 4 月前 ·

安静的松鼠 · Amazon Live· 1 年前 ·

3026862795 · 中国文学理论批评史_百度百科· 4 年前 ·

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，网址： http://activemq.apache.org/download.html 。我之前下的是5.3.1，5.3.2现在也已经出来了。

2、去ActiveMQ官方网站下载最新版的Apache.NMS，网址： http://activemq.apache.org/nms/download.html ，需要下载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; 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： <textarea class="c-sharp" name="code">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) //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); Console.ReadLine(); catch (System.Exception e) Console.WriteLine("{0}",e.Message); Console.ReadLine(); consumer: <textarea class="c-sharp" name="code">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) //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) 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： <textarea class="c-sharp" name="code">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：

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

------------------------------------------------------------------------------------------------------------------------------------------------------------

介于许多朋友对Demo程序的需求，上传一个之前自己写的WinForm Demo吧，下载地址在这里： http://download.csdn.net/detail/bodybo/4498147

截图如下：

MQ发送程序

MQ接收程序

关于 ActiveMQ 在 C# 中的应用，https://blog.csdn.net/bodybo/article/details/5647968这篇文章已经有介绍，但是这篇文章介绍的是 ActiveMQ 推送消息的模式，我碰到的场景是当前分布式节点从消息队列中读取任务，并在当前节点开启多线程进行处理。如果在本地进行缓存，则不利于负载均衡，且 服务器 宕机会造成任务丢失，所以我采用了主动从消息队列读取的... 最近公司的一个项目要使用 ActiveMQ ，客户端使用 C# ，.NET Framework4.7.2。在网上找了教程，引入了两个包分别是：Apache.NMS(v2.0.0) 和Apache.NMS. ActiveMQ (v1.8.0) 连接 ActiveMQ 的代码片段如下： using System; using Apache.NMS; using Apache.NMS. ActiveMQ ; using Apache.NMS.Util; Uri connecturi = new Uri(".. ActiveMQ 应用 1 ActiveMQ 常用 API 简介下述 API 都是接口类型,由定义在 javax.jms 包中 .是 JMS 标准接口定义.1.1ConnectionFactory链接工厂, 用于创建链接的工厂类型.1.2Connection链接. 用于建立访问 ActiveMQ 连接的类型, 由链接工厂创建.1.3 Session 会话, 一次持久有效有状态的访问. 由链接创建.1.4D... 1、修改 activemq .xml的配置文件 <broker xmlns="http:// activemq .apache.org/schema/core" brokerName="AMQ-DEMO" useJmx="true" dataDirectory="${ activemq .data}"> </broker> <managementContext> <managementContext createConnector="true" jmxDomainNam 最近开始学习 activemq 系统的使用，关于如何在 activemq 中时实获取一个主题/队列(topic/queue)的订阅/消费者数量，让我花了挺多时间才搞明白: 在 activemq 中，可以通过订阅主题或队列对应的消息管理主题,从而获取咨询消息(Advisory Message), 每当新增/减少订阅者时都会收到相关的咨询消息。 Advisory Message 中就包含了一些我们想要的数据(包含在... /** 启动 activeMQ 服务 */ public static void main(String[] args) throws Exception { RunServer rs = new RunServer(); 先上原文链接：http://blog.csdn.net/bodybo/article/details/5647968 ActiveMQ 在 C# 中的应用 ActiveMQ 是个好东东，不必多说。 ActiveMQ 提供多种语言支持，如Java, C, C++, C# , Ruby, Perl, Python, PHP等。由于我在windows下开发GUI，比较关心C++和 C# ，其中 C# 的 ActiveMQ 很简单...