activemq读取剩余消息队列中消息的数量

Wesley13
• 阅读 621

先上原文链接:

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很简单,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实现代码如下:

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该如何设置呢?请看如下代码:

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);

producer:

//生成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

点赞
收藏
评论区
推荐文章
blmius blmius
2年前
MySQL:[Err] 1292 - Incorrect datetime value: ‘0000-00-00 00:00:00‘ for column ‘CREATE_TIME‘ at row 1
文章目录问题用navicat导入数据时,报错:原因这是因为当前的MySQL不支持datetime为0的情况。解决修改sql\mode:sql\mode:SQLMode定义了MySQL应支持的SQL语法、数据校验等,这样可以更容易地在不同的环境中使用MySQL。全局s
Jacquelyn38 Jacquelyn38
2年前
2020年前端实用代码段,为你的工作保驾护航
有空的时候,自己总结了几个代码段,在开发中也经常使用,谢谢。1、使用解构获取json数据let jsonData  id: 1,status: "OK",data: 'a', 'b';let  id, status, data: number   jsonData;console.log(id, status, number )
皕杰报表之UUID
​在我们用皕杰报表工具设计填报报表时,如何在新增行里自动增加id呢?能新增整数排序id吗?目前可以在新增行里自动增加id,但只能用uuid函数增加UUID编码,不能新增整数排序id。uuid函数说明:获取一个UUID,可以在填报表中用来创建数据ID语法:uuid()或uuid(sep)参数说明:sep布尔值,生成的uuid中是否包含分隔符'',缺省为
Stella981 Stella981
2年前
KVM调整cpu和内存
一.修改kvm虚拟机的配置1、virsheditcentos7找到“memory”和“vcpu”标签,将<namecentos7</name<uuid2220a6d1a36a4fbb8523e078b3dfe795</uuid
Wesley13 Wesley13
2年前
MQ 消息中间件梳理
!(https://oscimg.oschina.net/oscnet/c44a8479ca04449e89ed30c37058117a.jpg)_来源:_blog.csdn.net/qq\_29676623/article/details/85108070为什么使用消息队列?消息队列的优点和缺点?kafka、activemq
Stella981 Stella981
2年前
Spring Boot日志集成
!(https://oscimg.oschina.net/oscnet/1bde8e8d00e848be8b84e9d1d44c9e5c.jpg)SpringBoot日志框架SpringBoot支持JavaUtilLogging,Log4j2,Lockback作为日志框架,如果你使用star
Wesley13 Wesley13
2年前
ActiveMQ 简介
ActiveMQ是什么?•  ActiveMQ是Apache开发的一款开源消息中间件;•  支持JMS1.1协议和J2EE1.4规范;•  支持语言Java,C,C,C,Ruby,Perl,Python,PHP;•  并且对Spring支持友好,可以很方便的集成;•  并且支持多种传送协议:inVM,TCP,SSL,NIO,UD
Stella981 Stella981
2年前
Django中Admin中的一些参数配置
设置在列表中显示的字段,id为django模型默认的主键list_display('id','name','sex','profession','email','qq','phone','status','create_time')设置在列表可编辑字段list_editable
Wesley13 Wesley13
2年前
MySQL部分从库上面因为大量的临时表tmp_table造成慢查询
背景描述Time:20190124T00:08:14.70572408:00User@Host:@Id:Schema:sentrymetaLast_errno:0Killed:0Query_time:0.315758Lock_
Python进阶者 Python进阶者
2个月前
Excel中这日期老是出来00:00:00,怎么用Pandas把这个去除
大家好,我是皮皮。一、前言前几天在Python白银交流群【上海新年人】问了一个Pandas数据筛选的问题。问题如下:这日期老是出来00:00:00,怎么把这个去除。二、实现过程后来【论草莓如何成为冻干莓】给了一个思路和代码如下:pd.toexcel之前把这