使用Active MQ在.net和java系統之間通信

ActiveMQ 是Apache出品，最流行的，能力強勁的開源消息總線。ActiveMQ 是一個完全支持JMS1.1和J2EE 1.4規范的 JMS Provider實現

一.特性列表

⒈ 多種語言和協議編寫客戶端。語言: Java,C,C++,C#,Ruby,Perl,Python,PHP。應用協議： OpenWire,Stomp REST,WS Notification,XMPP,AMQP
⒉ 完全支持JMS1.1和J2EE 1.4規范 （持久化，XA消息，事務)
⒊ 對Spring的支持，ActiveMQ可以很容易內嵌到使用Spring的系統里面去，而且也支持Spring2.0的特性
⒋ 通過了常見J2EE服務器（如 Geronimo,JBoss 4,GlassFish,WebLogic)的測試，其中通過JCA 1.5 resource adaptors的配置，可以讓ActiveMQ可以自動的部署到任何兼容J2EE 1.4 商業服務器上
⒌ 支持多種傳送協議：in-VM,TCP,SSL,NIO,UDP,JGroups,JXTA
⒍ 支持通過JDBC和journal提供高速的消息持久化
⒎ 從設計上保證了高性能的集群，客戶端-服務器，點對點
⒏ 支持Ajax
⒐ 支持與Axis的整合
⒑ 可以很容易得調用內嵌JMS provider，進行測試

二、Jms規范里的兩種message傳輸方式Topic和Queue，兩者的對比如下：

Topic
Queue

概要
Publish Subscribe messaging 發布訂閱消息
Point-to-Point 點對點

有無狀態
topic數據默認不落地，是無狀態的。

Queue數據默認會在mq服務器上以文件形式保存，比如Active MQ一般保存在$AMQ_HOME\data\kr-store\data下面。也可以配置成DB存儲。

完整性保障
並不保證publisher發布的每條數據，Subscriber都能接受到。
Queue保證每條數據都能被receiver接收。

消息是否會丟失
一般來說publisher發布消息到某一個topic時，只有正在監聽該topic地址的sub能夠接收到消息；如果沒有sub在監聽，該topic就丟失了。
Sender發送消息到目標Queue，receiver可以異步接收這個Queue上的消息。Queue上的消息如果暫時沒有receiver來取，也不會丟失。

消息發布接收策略
一對多的消息發布接收策略，監聽同一個topic地址的多個sub都能收到publisher發送的消息。Sub接收完通知mq服務器
一對一的消息發布接收策略，一個sender發送的消息，只能有一個receiver接收。receiver接收完后，通知mq服務器已接收，mq服務器對queue里的消息采取刪除或其他操作。

三、安裝

1.首先去http://activemq.apache.org/download.html 下載最新版本

2.解壓后，目錄如下：

+bin (windows下面的bat和unix/linux下面的sh)

+conf (activeMQ配置目錄，包含最基本的activeMQ配置文件）

+data （默認是空的）

+docs (index,replease版本里面沒有文檔，-.-b不知道為啥不帶）

+example （幾個例子

+lib (activemMQ使用到的lib)

-apache-activemq-4.1-incubator.jar (ActiveMQ的binary)

-LICENSE.txt

-NOTICE.txt

-README.txt

-user-guide.html

3.使用bin\activemq.bat啟動服務

四、創建.Net Producer

    class Program
    {
        static IConnectionFactory _factory = null;
        static IConnection _connection = null;
        static ITextMessage _message = null;
        static ITextMessage _message2 = null;
        static void Main(string[] args)
        {
            //創建工廠
            _factory = new ConnectionFactory("tcp://localhost:61616/");
            try
            {
                //創建連接
                using (_connection = _factory.CreateConnection())
                {
                    //創建會話
                    using (ISession session = _connection.CreateSession(AcknowledgementMode.AutoAcknowledge))
                    {
                        //創建一個主題
                        IDestination destination = new Apache.NMS.ActiveMQ.Commands.ActiveMQTopic("topic");

                        //創建生產者
                        IMessageProducer producer = session.CreateProducer(destination);

                        int counter = 0;
                        Console.WriteLine("請輸入你要發送數據，然后回車！");

                        while (true)
                        {
                            string msg = Console.ReadLine();
                            if (msg != string.Empty)
                            {
                                Console.BackgroundColor = ConsoleColor.Blue;
                                Console.Beep();
                                counter++;

                                //創建一個文本消息
                                _message = producer.CreateTextMessage("This is .Net AcitveMQ Message....");
                                //_message2 = producer.CreateTextMessage("<Root><First>Test</First><Root>");

                                Console.WriteLine("正在發送第{0}組發送數據...........", counter);
                                //發送消息
                                producer.Send(_message, MsgDeliveryMode.Persistent, MsgPriority.Normal,
                                    TimeSpan.MaxValue);

                                Console.WriteLine("'"+_message.Text+"'已經發送！");
                                //producer.Send(_message2, MsgDeliveryMode.Persistent, MsgPriority.Normal,
                                //    TimeSpan.MinValue);
                                //Console.WriteLine("'" + _message2.Text + "'已經發送！");
                                producer.Send(msg, MsgDeliveryMode.Persistent, MsgPriority.Normal, TimeSpan.MinValue);
                                Console.WriteLine("你輸入的數據'" + msg + "'已經發送！");
                                Console.WriteLine("**************************************************");
                                Console.BackgroundColor = ConsoleColor.Black;
                                Console.WriteLine("請輸入你要發送數據，然后回車！");
                            }
                        }
                    }
                }
            }
            catch (Exception ex)
            {
                Console.WriteLine(ex.ToString());
            }
            Console.ReadLine();
        }
    }

五、創建Java Consumer

import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.Session;
import javax.jms.TextMessage;
import javax.jms.Topic;
import javax.jms.TopicSubscriber;

import org.apache.activemq.ActiveMQConnectionFactory;

public class ActiveMQConsumer {
    public static void main(String[] args) {
        // ConnectionFactory ：連接工廠，JMS 用它創建連接
        ConnectionFactory connectionFactory;
        // Connection ：JMS 客戶端到JMS Provider 的連接
        Connection connection = null;
        // Session： 一個發送或接收消息的線程
        Session session;
        Topic topic;
        // 消費者，消息接收者
        TopicSubscriber subscriber;
        connectionFactory = new ActiveMQConnectionFactory(
                //ActiveMQConnection.DEFAULT_USER,
                //ActiveMQConnection.DEFAULT_PASSWORD,
                "tcp://localhost:61616");
        try {
            // 構造從工廠得到連接對象
            connection = connectionFactory.createConnection();
            connection.setClientID("Customer");
            // 啟動
            connection.start();
            // 獲取操作連接
            session=connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
            topic = session.createTopic("topic");
            subscriber= session.createDurableSubscriber(topic, "Customer", null, false);

            while (true) {
                //TextMessage message = (TextMessage) subscriber.receive(10000);
                //TextMessage message = (TextMessage) subscriber.receiveNoWait();
                TextMessage message= (TextMessage) subscriber.receive();
                if (null != message) {
                    System.out.println("收到消息" + message.getText());

                } else {
                    break;
                }
            }
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            try {
                if (null != connection)
                    connection.close();
            } catch (Throwable ignore) {
            }
        }
    }
}

六、參考資料

http://heisetoufa.iteye.com/blog/1908335

http://www.cnblogs.com/sunwei2012/archive/2010/03/22/1691919.html

http://www.cnblogs.com/tenghoo/archive/2009/11/23/activeMQ_client.html

http://www.cnblogs.com/CopyPaster/archive/2012/04/27/2473179.html

http://blog.csdn.net/gongqingkui/article/details/8615777

http://blog.csdn.net/mypurse/article/details/6265073