当前位置：首页 > CN2资讯 > 正文内容

windows server 2016 消息队列安装 server2012消息队列

11小时前CN2资讯

环境准备

1.安装jdk1.7+

2.下载新版ActiveMQ

http://activemq.apache.org/

3.启动activemq服务

4.启动成功后的界面是

5.启动成功后

浏览器访问http://localhost:8161/admin，默认用户名和密码admin/admin，管理员界面如下：

6.如果你想修改用户名和密码的话

在conf/jetty-realm.properties中修改即可，重启服务才能生效。

7.如果需要修改端口可以在jetty文件中修改，重启服务才能生效。

发送消息的基本步骤

(1)、创建连接使用的工厂类JMS ConnectionFactory

(2)、使用管理对象JMS ConnectionFactory建立连接Connection，并启动

(3)、使用连接Connection 建立会话Session

(4)、使用会话Session和管理对象Destination创建消息生产者MessageSender

(5)、使用消息生产者MessageSender发送消息

消息接收者从JMS接受消息的步骤

(1)、创建连接使用的工厂类JMS ConnectionFactory

(2)、使用管理对象JMS ConnectionFactory建立连接Connection，并启动

(3)、使用连接Connection 建立会话Session

(4)、使用会话Session和管理对象Destination创建消息接收者MessageReceiver

(5)、使用消息接收者MessageReceiver接受消息，需要用setMessageListener将MessageListener接口绑定到MessageReceiver消息接收者必须实现了MessageListener接口，需要定义onMessage事件方法。

C#实战代码

1.ActiveMQ官方网站下载最新版的C#驱动

Apache.NMS：http://activemq.apache.org/nms/nms-api-downloads.html

Apache.NMS.ActiveMQ：http://activemq.apache.org/nms/activemq-downloads.html

2.引用以下两个dll

Apache.NMS.ActiveMQ.dll Apache.NMS.dll

config中添加配置

在ActiveMQ的conf目录的activemq.xml中的节点<broker></broker>里面添加账号密码，重启服务才能生效

<plugins> <simpleAuthenticationPlugin> <users> <authenticationUser username="admin" password="manager" groups="users,admins"/> </users> </simpleAuthenticationPlugin> </plugins>

生产者代码

public void ActiveMQProducer() { string ActiveMQUrl = System.Configuration.ConfigurationManager.AppSettings["ActiveMQUrl"]; IConnectionFactory factory = new ConnectionFactory(ActiveMQUrl); //通过工厂建立连接 using (IConnection connection = factory.CreateConnection("admin", "manager")) { //通过连接创建Session会话 using (ISession session = connection.CreateSession()) { //通过会话创建生产者，方法里面new出来的是MQ中的Queue ActiveMQQueue quene = new ActiveMQQueue("audioQueue");//点到点模式队列模式 //ActiveMQTopic quene = new ActiveMQTopic("testTopic");//广播形式发布订阅模式发送的消息如果没有消费者接收就会没了 IMessageProducer prod = session.CreateProducer(quene); //创建一个发送的消息对象 ITextMessage message = prod.CreateTextMessage(); //给这个对象赋实际的消息 //任务实体模型 ActiveMQInfo activeMQInfo = new ActiveMQInfo(); activeMQInfo.TaskID = Guid.NewGuid().ToString(); activeMQInfo.TaskName = "订单处理"; activeMQInfo.FileName = Guid.NewGuid().ToString(); activeMQInfo.TaskDateTime = DateTime.Now; message.Text = JsonConvert.SerializeObject(activeMQInfo); //设置消息对象的属性，这个很重要哦，是Queue的过滤条件，也是P2P消息的唯一指定属性 message.Properties.SetString("filter", "getduration"); //生产者把消息发送出去，几个枚举参数MsgDeliveryMode是否长链，MsgPriority消息优先级别，发送最小单位，当然还有其他重载 prod.Send(message, MsgDeliveryMode.NonPersistent, MsgPriority.Normal, TimeSpan.MinValue); } } }

消费者代码

public void ActiveMQConsumer() { try { string ActiveMQUrl = System.Configuration.ConfigurationManager.AppSettings["ActiveMQUrl"]; //创建连接工厂 IConnectionFactory factory = new ConnectionFactory(ActiveMQUrl); //通过工厂构建连接 IConnection connection = factory.CreateConnection("admin", "manager"); //这个是连接的客户端名称标识 connection.ClientId = "GuardQueueListener"; //启动连接，监听的话要主动启动连接 connection.Start(); //logger.Error("消息队列启动"); //通过连接创建一个会话 ISession session = connection.CreateSession(); //通过会话创建一个消费者，这里就是Queue这种会话类型的监听参数设置 ActiveMQQueue quene = new ActiveMQQueue("audioQueue");//队列模式 //ActiveMQTopic quene = new ActiveMQTopic("testTopic");//发布订阅模式发送的消息如果没有消费者接收就会没了 IMessageConsumer consumerAudio = session.CreateConsumer(quene,"filter='getduration'"); //注册监听事件 consumerAudio.Listener += new MessageListener(consumerMediaAudio_Listener); } catch (Exception ex) { //记录异常信息 } } private void consumerMediaAudio_Listener(IMessage message) { ITextMessage iTextMessage = (ITextMessage)message; string msg = iTextMessage.Text; //逻辑处理代码 //ActiveMQInfo activeMQInfo = JsonConvert.DeserializeObject<ActiveMQInfo>(msg); }

ActiveMQInfo实体类

public class ActiveMQInfo { public string TaskID { get; set; } public string TaskName { get; set; } public string FileName { get; set; } public DateTime TaskDateTime { get; set; } }