首页 > 代码库 > ActiveMQ中的Destination高级特性(一)

ActiveMQ中的Destination高级特性(一)

技术分享

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

Destination高级特性----->Composite Destinations
  组合队列Composite Destinations : 允许用一个虚拟的destination代表多个destinations,这样就可以通过composite destinations在一个操作中同时向多个queue/topic发送消息。
  有两种实现方式:
    第一种:在客户端编码实现
    第二种:在activemq.xml配置文件中实现

  第一种:在客户端编码实现
    在composite destinations中,多个destination之间采用","分隔。如下:这里有2个destination  "my-queue1"和"my-queue2"

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

消息发送程序:JmsSend.java

 1 import javax.jms.Connection;
 2 import javax.jms.ConnectionFactory;
 3 import javax.jms.Destination;
 4 import javax.jms.MessageProducer;
 5 import javax.jms.Session;
 6 import javax.jms.TextMessage;
 7 
 8 import org.apache.activemq.ActiveMQConnectionFactory;
 9 
10 public class JmsSend {
11     public static void main(String[] args) throws Exception {
12         ConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://192.168.1.81:61616");
13         Connection connection = connectionFactory.createConnection();
14         connection.start();
15         
16      Session session = connection.createSession(Boolean.TRUE, Session.AUTO_ACKNOWLEDGE);
17      Destination destination = session.createQueue("my-queue1,my-queue2");
18         
19         MessageProducer producer = session.createProducer(destination);
20         for(int i = 0;i < 5;i++){
21             TextMessage message = session.createTextMessage("messageAAA --->" + i);
22             //通过生产者发出消息
23             producer.send(message);
24         }
25         session.commit();
26         session.close();
27         connection.close();
28     }
29 }

运行消息发送程序后,去访问http://192.168.1.81:8161/admin/queues.jsp,可以看到如下:

技术分享

第一个消费者程序:JmsReceiver.java,  从"my-queue1"这个destination中消费消息

 1 import javax.jms.Connection;
 2 import javax.jms.ConnectionFactory;
 3 import javax.jms.Destination;
 4 import javax.jms.MessageConsumer;
 5 import javax.jms.Session;
 6 import javax.jms.TextMessage;
 7 
 8 import org.apache.activemq.ActiveMQConnectionFactory;
 9 
10 public class JmsReceiver {
11     public static void main(String[] args) throws Exception {
12         ConnectionFactory cf = new ActiveMQConnectionFactory("tcp://192.168.1.81:61616");
13         Connection connection =  cf.createConnection();
14         connection.start();
15         
16     Session session = connection.createSession(Boolean.TRUE, Session.AUTO_ACKNOWLEDGE);
17     Destination destination = session.createQueue("my-queue1");
18         MessageConsumer consumer = session.createConsumer(destination);
19         int i = 0;
20         while(i < 5){
21             Thread.sleep(1000);
22             i++;
23             TextMessage message = (TextMessage)consumer.receive();
24             session.commit();
25             System.out.println("111接收到的消息是:"+message.getText());
26         }
27         session.close();
28         connection.close();
29     }
30 }

运行结果:

技术分享

再去访问http://192.168.1.81:8161/admin/queues.jsp,可以看到如下:

技术分享

第二个消费者程序:JmsReceiver2.java, 从"my-queue2"这个destination中消费消息

 1 import javax.jms.Connection;
 2 import javax.jms.ConnectionFactory;
 3 import javax.jms.Destination;
 4 import javax.jms.MessageConsumer;
 5 import javax.jms.Session;
 6 import javax.jms.TextMessage;
 7 
 8 import org.apache.activemq.ActiveMQConnectionFactory;
 9 
10 public class JmsReceiver2 {
11     public static void main(String[] args) throws Exception {
12         ConnectionFactory cf = new ActiveMQConnectionFactory("tcp://192.168.1.81:61616");
13         Connection connection =  cf.createConnection();
14         connection.start();
15         
16       Session session = connection.createSession(Boolean.TRUE, Session.AUTO_ACKNOWLEDGE);
17       Destination destination = session.createQueue("my-queue2");
18         MessageConsumer consumer = session.createConsumer(destination);
19         int i = 0;
20         while(i < 5){
21             Thread.sleep(1000);
22             i++;
23             TextMessage message = (TextMessage)consumer.receive();
24             session.commit();
25             System.out.println("222接收到的消息是:"+message.getText());
26         }
27         session.close();
28         connection.close();
29     }
30 }

 运行结果:

技术分享

 再去访问http://192.168.1.81:8161/admin/queues.jsp,可以看到如下:

 技术分享技术分享

  第二种:在activemq.xml配置文件中实现

技术分享

消息发送程序,JmsSend.java

 1 import javax.jms.Connection;
 2 import javax.jms.ConnectionFactory;
 3 import javax.jms.Destination;
 4 import javax.jms.MessageProducer;
 5 import javax.jms.Session;
 6 import javax.jms.TextMessage;
 7 
 8 import org.apache.activemq.ActiveMQConnectionFactory;
 9 
10 public class JmsSend {
11     public static void main(String[] args) throws Exception {
12         ConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://192.168.1.81:61616");
13         Connection connection = connectionFactory.createConnection();
14         connection.start();
15         
16       Session session = connection.createSession(Boolean.TRUE, Session.AUTO_ACKNOWLEDGE);
17       Destination destination = session.createQueue("MY.QUEUE");
18         
19         MessageProducer producer = session.createProducer(destination);
20         for(int i = 0;i < 5;i++){
21             TextMessage message = session.createTextMessage("messageAAA --->" + i);
22             //通过生产者发出消息
23             producer.send(message);
24         }
25         session.commit();
26         session.close();
27         connection.close();
28     }
29 }

运行后,去访问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

 

 

 

 

 

 

 

 

 

 

 

 

ActiveMQ中的Destination高级特性(一)