一 、發送端

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

ActiveMQ學習筆記（二、p2p方式實現）