ActiveMQ中的Destination高級特性(一)

----------------------------------------------------------------------------------------

Destination高級特性----->Composite Destinations
　　組合隊列Composite Destinations ： 允許用一個虛擬的destination代表多個destinations,這樣就可以通過composite destinations在一個操作中同時向多個queue/topic發送消息。
　　有兩種實現方式：
　　　　第一種:在客戶端編碼實現
　　　　第二種：在activemq.xml配置文件中實現

　　第一種：在客戶端編碼實現
　　　　在composite destinations中，多個destination之間采用","分隔。如下：這里有2個destination  "my-queue1"和"my-queue2"

    Session session = connection.createSession(Boolean.TRUE, Session.AUTO_ACKNOWLEDGE);
    Destination destination = session.createQueue("my-queue1,my-queue2");

消息發送程序:JmsSend.java

import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.Destination;
import javax.jms.MessageProducer;
import javax.jms.Session;
import javax.jms.TextMessage;

import org.apache.activemq.ActiveMQConnectionFactory;

public class JmsSend {
    public static void main(String[] args) throws Exception {
        ConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://192.168.1.81:61616");
        Connection connection = connectionFactory.createConnection();
        connection.start();
        
Session session = connection.createSession(Boolean.TRUE, Session.AUTO_ACKNOWLEDGE); 17 Destination destination = session.createQueue("my-queue1,my-queue2"); 18         
        MessageProducer producer = session.createProducer(destination);
        for(int i = 0;i < 5;i++){
            TextMessage message = session.createTextMessage("messageAAA --->" + i);
            //通過生產者發出消息
            producer.send(message);
        }
        session.commit();
        session.close();
        connection.close();
    }
}

運行消息發送程序后，去訪問http://192.168.1.81:8161/admin/queues.jsp，可以看到如下：

第一個消費者程序:JmsReceiver.java,  從"my-queue1"這個destination中消費消息

import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.Destination;
import javax.jms.MessageConsumer;
import javax.jms.Session;
import javax.jms.TextMessage;

import org.apache.activemq.ActiveMQConnectionFactory;

public class JmsReceiver {
    public static void main(String[] args) throws Exception {
        ConnectionFactory cf = new ActiveMQConnectionFactory("tcp://192.168.1.81:61616");
        Connection connection =  cf.createConnection();
        connection.start();
        
Session session = connection.createSession(Boolean.TRUE, Session.AUTO_ACKNOWLEDGE); 17 Destination destination = session.createQueue("my-queue1"); 18         MessageConsumer consumer = session.createConsumer(destination);
        int i = 0;
        while(i < 5){
            Thread.sleep(1000);
            i++;
            TextMessage message = (TextMessage)consumer.receive();
            session.commit();
            System.out.println("111接收到的消息是:"+message.getText());
        }
        session.close();
        connection.close();
    }
}

運行結果:

再去訪問http://192.168.1.81:8161/admin/queues.jsp，可以看到如下：

第二個消費者程序:JmsReceiver2.java, 從"my-queue2"這個destination中消費消息

import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.Destination;
import javax.jms.MessageConsumer;
import javax.jms.Session;
import javax.jms.TextMessage;

import org.apache.activemq.ActiveMQConnectionFactory;

public class JmsReceiver2 {
    public static void main(String[] args) throws Exception {
        ConnectionFactory cf = new ActiveMQConnectionFactory("tcp://192.168.1.81:61616");
        Connection connection =  cf.createConnection();
        connection.start();
        
      Session session = connection.createSession(Boolean.TRUE, Session.AUTO_ACKNOWLEDGE); 17       Destination destination = session.createQueue("my-queue2"); 18         MessageConsumer consumer = session.createConsumer(destination);
        int i = 0;
        while(i < 5){
            Thread.sleep(1000);
            i++;
            TextMessage message = (TextMessage)consumer.receive();
            session.commit();
            System.out.println("222接收到的消息是:"+message.getText());
        }
        session.close();
        connection.close();
    }
}

 運行結果:

 再去訪問http://192.168.1.81:8161/admin/queues.jsp，可以看到如下：

 

　　第二種：在activemq.xml配置文件中實現

消息發送程序，JmsSend.java

import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.Destination;
import javax.jms.MessageProducer;
import javax.jms.Session;
import javax.jms.TextMessage;

import org.apache.activemq.ActiveMQConnectionFactory;

public class JmsSend {
    public static void main(String[] args) throws Exception {
        ConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://192.168.1.81:61616");
        Connection connection = connectionFactory.createConnection();
        connection.start();
        
Session session = connection.createSession(Boolean.TRUE, Session.AUTO_ACKNOWLEDGE); 17 Destination destination = session.createQueue("MY.QUEUE"); 18         
        MessageProducer producer = session.createProducer(destination);
        for(int i = 0;i < 5;i++){
            TextMessage message = session.createTextMessage("messageAAA --->" + i);
            //通過生產者發出消息
            producer.send(message);
        }
        session.commit();
        session.close();
        connection.close();
    }
}

運行后，去訪問http://192.168.1.81:8161/admin/queues.jsp，可以看到如下：

消費者程序1和2  與上面的代碼一樣，這里就不重復了。

 Destination高級特性----->Configure Startup Destinations

　　在啟動ActiveMQ的時候如果需要創建Destination的話，可以在activemq.xml中配置

 啟動activemq后去訪問，http://192.168.1.81:8161/admin

 Destination高級特性----->Delete Inactive Destinations

　　在ActiveMQ的queue在不使用之后，可以通過web控制台或者JMX方式來刪除掉，當然，也可以通過配置，使得broker可以自動探測到無用的隊列並刪除掉，回收響應資源。

說明：
　　schedulePeriodForDestinationPurge: 設置多長時間檢查一次，這里是10秒。
　　inactiveTimoutBeforeGC: 設置當Destination為空后，多長時間被刪除，這里是30秒。
　　gcInactiveDestinations:設置刪除掉不活動的隊列，默認為false

 Destination高級特性----->wildcars(通配符)

　　Wildcars用來支持名字分層體系，它不是JMS規范的一部分，是ActiveMQ的擴展。

　　ActiveMQ支持以下三種wildcars：
　　　　.     用於作為路徑上名字間的分隔符
　　　　*    用於匹配路徑上的任何名字
　　　　>   用於遞歸地匹配任何以這個名字開始的destination