admin管理员组文章数量:1633102
----------------------------------------------------------------------------------------
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 sessionmit();
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 sessionmit();
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 sessionmit();
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 sessionmit();
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
队列选项(Destination options)
队列选项是给consumer在JMS规范之外添加的功能特性,通过在队列名称后面使用类似URL的语法添加多个选项。
选项 | 默认值 | 描述 |
---|---|---|
consumer.prefetchSize | 不定 | consumer持有的未确认最大消息数量 参见:prefetch。 |
consumer.maximumPendingMessageLimit | 0 | 控制非持久主题在慢消费(slow consumer)情况下丢弃消息的最大数量。 |
consumer.noLocal | false | |
consumer.dispatchAsync | true | 是否异步分发,参见 dispatch messages asynchronously。 |
consumer.retroactive | false | 是否为回溯消费者 Retroactive Consumer. |
consumer.selector | null | JMS Selector. |
consumer.exclusive | false | 是否为独占消费者 Exclusive Consumer. |
consumer.priority | 0 | 配置消费者优先级 Consumer Priority. |
配置示例:
queue=new ActiveMQQueue("TEST.QUEUE?consumer.dispatchAsync=false&consumer.prefetchSize=10");
consumer=session.createConsumer(queue);
本文标签: 学习笔记特性高级ActiveMqdestination
版权声明:本文标题:ActiveMQ学习笔记(十六)Destination高级特性一 内容由热心网友自发贡献,该文观点仅代表作者本人, 转载请联系作者并注明出处:https://m.elefans.com/dongtai/1729143681a1187723.html, 本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如发现本站有涉嫌抄袭侵权/违法违规的内容,一经查实,本站将立刻删除。
发表评论