public void ActiveSend() { IConnectionFactory factory = new ConnectionFactory("tcp://localhost:61616"); //通过工厂建立连接 using (IConnection connection = factory.CreateConnection()) { try { //通过连接创建Session会话 using (ISession session = connection.CreateSession()) { //通过会话创建生产者,方法里面new出来的是MQ中的Queue using (IMessageProducer prod = session.CreateProducer(new Apache.NMS.ActiveMQ.Commands.ActiveMQQueue("firstQueue"))) { //创建一个发送的消息对象 ITextMessage message = prod.CreateTextMessage(); while (true) { try { //给这个对象赋实际的消息 message.Text = Console.ReadLine(); //设置消息对象的属性,这个很重要哦,是Queue的过滤条件,也是P2P消息的唯一指定属性 message.Properties.SetString("filter", "demo"); //生产者把消息发送出去,几个枚举参数MsgDeliveryMode是否长链,MsgPriority消息优先级别,发送最小单位,当然还有其他重载 prod.Send(message, MsgDeliveryMode.NonPersistent, MsgPriority.Normal, TimeSpan.MinValue); Console.WriteLine("消息发送成功"); } catch (Exception ex) { Console.WriteLine("消息发送失败!" + ex); } } } } } catch (Exception e) { Console.WriteLine("创建发送MQ失败失败" + e); } } }
public void ActiveReceived() { //创建连接工厂 IConnectionFactory factory = new ConnectionFactory("tcp://localhost:61616"); try { //通过工厂构建连接 using (IConnection connection = factory.CreateConnection()) { //这个是连接的客户端名称标识 connection.ClientId = "firstQueueListener"; //启动连接,监听的话要主动启动连接 connection.Start(); //通过连接创建一个会话 using (ISession session = connection.CreateSession()) { //通过会话创建一个消费者,这里就是Queue这种会话类型的监听参数设置 using (IMessageConsumer consumer = session.CreateConsumer(new Apache.NMS.ActiveMQ.Commands.ActiveMQQueue("firstQueue"), "filter=‘demo‘")) { ITextMessage message; while (true) { try { message = (ITextMessage)consumer.Receive(); Console.WriteLine("Receive msg:" + message.Text); } catch (Exception ex) { Console.WriteLine("接收消息失败!" + ex); } } } } } } catch (Exception ex) { Console.WriteLine("启动接收失败!" + ex); } }
测试结果可以去http://localhost:8161/admin查看,登陆用户密码默认均为admin
完整测试代码
https://pan.baidu.com/s/1NjKtg77cYTRbswxl5cO67A
原文:https://www.cnblogs.com/xieyang07/p/10404040.html