分布式-信息方式-ActiveMQ的Destination高级特性3
虚拟destination用来创建逻辑destination,客户端可以通过它来生产和消费消息,它会把消息映射到物理destination.
ActiveMQ支持2种方式:
1:虚拟主题(Virtual Topics)
2:组合Destinations(Composite Destinations)
为什么使用虚拟主题?
ActiveMQ只有在持久订阅才是持久化的。持久订阅时,每一个持久订阅者,都相当于一个queue的客户端,它会收取所有消息。这种情况下存在两个问题:
第一:同一应用内消费者端护在均衡的问题。也就是说一个应用程序内的持久化消息,不能使用对个消费者共同承担消息处理能力。因为每个消费者都会获取所有消息。因为每一个消费者都会获取所有信息。
Queue到时可以解决这个问题,但broker端又不能将消息发送到多个应用端,所以纪要发布订阅,又要让消费者分组,这个功能JMS本身是没有的
第二:同一应用内消费者端failover问题,由于只能使用单个的持久订阅者,如果这个订阅者出错,则应用就无法处理,系统的健壮性不高。
如何使用虚拟topic?
第一:对于消息发布者来说,就是一个正常的topic,名称以VirtualTopic.开始,比如VirtualTopic.Orders,代码示例如下:
Topicdestination = session.createTopic("VirtualTopic.Orders");
第二:对于消息接收端来说,是个队列,不同应用里使用不同的前缀作为队列名称,即可表明自己的身份即可实现消费端应用分组。
例如Consumer.A.VirtualTopic.Orders说明它是名称为A的消费端,同理Consumer.B VirtualTopic.Orders说明是一名称为B的消费端。可以在同一个应用中使用多个消费者消费这个队列
又因为不同应用使用的topic名称不一样,前缀不同,所以不同应用中都可以接受到全部消息。每一个客户端相当于一个持久订阅者,而且这个客户端可以使用多个消费者共同来承担消费任务。
代码示例:
Destination dest = session.createQueue("Consumer.A.VirtualTopic.Orders");
代码如下:
package test.mq.visualDestinations; import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.DeliveryMode;
import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.MapMessage;
import javax.jms.MessageProducer;
import javax.jms.Session;
import javax.jms.TextMessage; import org.apache.activemq.ActiveMQConnectionFactory; public class Sender {
public static void main(String[] args) throws JMSException { ConnectionFactory ConnectionFactory=new ActiveMQConnectionFactory(
"tcp://localhost:61616"
); Connection connection=ConnectionFactory.createConnection();
connection.start(); Session session=connection.createSession(Boolean.TRUE, Session.AUTO_ACKNOWLEDGE); Destination destination=session.createTopic("VirtualTopic.mytopic1"); MessageProducer messageProducer=session.createProducer(destination); for(int i=1;i<=5;i++){
TextMessage textMessage=session.createTextMessage();
textMessage.setText("我是TOM ID为"+i);
messageProducer.send(textMessage);
System.out.println("生产者:"+textMessage.getText()); }
session.commit();
session.close();
connection.close();
}
}
package test.mq.visualDestinations; 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 Receiver1 {
public static void main(String[] args) throws Exception {
ConnectionFactory cf = new ActiveMQConnectionFactory("tcp://localhost:61616");
Connection connection = cf.createConnection();
connection.start(); Session session = connection.createSession(Boolean.TRUE, Session.AUTO_ACKNOWLEDGE);
Destination destination = session.createQueue("Consumer.A.VirtualTopic.mytopic1");
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("1接收到的消息是:"+message.getText());
}
session.close();
connection.close();
}
}
package test.mq.visualDestinations; import java.util.Enumeration;
import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageConsumer;
import javax.jms.MessageListener;
import javax.jms.Session;
import javax.jms.TextMessage;
import org.apache.activemq.ActiveMQConnectionFactory;
public class Receiver2{ public static void main(String[] args) throws JMSException {
ConnectionFactory connectionFactory=new ActiveMQConnectionFactory(
"tcp://localhost:61616"
);
for(int i=0;i<5;i++){
Thread t=new MyThread2(connectionFactory);
t.start();
try {
Thread.sleep(1000l);
} catch (InterruptedException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
} }
}
class MyThread2 extends Thread{
private ConnectionFactory connectionFactory=null;
public MyThread2(ConnectionFactory connectionFactory){
this.connectionFactory = connectionFactory;
}
public void run(){
try {
final Connection connection = connectionFactory.createConnection();
connection.start();
Enumeration names=connection.getMetaData().getJMSXPropertyNames(); final Session session=connection.createSession(Boolean.TRUE, Session.AUTO_ACKNOWLEDGE);
Destination destination=session.createQueue("Consumer.B.VirtualTopic.mytopic1");
MessageConsumer Consumer=session.createConsumer(destination);
Consumer.setMessageListener(new MessageListener() {
@Override
public void onMessage(Message msg) {
TextMessage txtmsg=(TextMessage) msg;
try {
System.out.println("接收信息2--->"+txtmsg.getText());
} catch (JMSException e1) {
e1.printStackTrace();
}
try {
session.commit();
} catch (JMSException e) {
e.printStackTrace();
}
try {
session.close();
} catch (JMSException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
try {
connection.close();
} catch (JMSException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
}
}); } catch (JMSException e) {
// TODO Auto-generated catch block
e.printStackTrace();
} }
}
其实把消费者队列化了。
修改虚拟主题的前缀:
默认前缀是VirtualTopic.>
自定义消费虚拟地址默认格式:Consumer.*.VirtualTopic.>
修改配置:
<broker xmlns="http://activemq.apache.org/schema/core">
<destinationInterceptors>
<virtualDestinationInterceptor>
<virtualDestinations>
<virtualTopic name=">" prefix="VirtualTopicConsumers.*." selectorAware="false" />
</virtualDestinations>
</virtualDestinationInterceptor>
</destinationInterceptors>
</broker>
最新文章
- 搞定.NET MVC IOC控制反转,依赖注入
- winserver 08 64位安装sql05 64位提示asp版本注册
- Errore Http 404.2
- JS延时提示框
- git merge和个git rebase的区别
- 更改linux系统提示信息
- mongo数据库时间存储的问题
- ASP.NET Core源码学习(一)Hosting
- bzoj 1076 奖励关 状压+期望dp
- Tomcat配置域名/IP访问及其中遇到的问题注意事项
- ant安装
- Python 3.X 要使用urllib.request 来抓取网络资源。转
- 解决:Tomcat 局域网IP地址 访问不了
- go 删除数组元素
- jquery chosen api
- PD 15.1 安装 破解 , 简单使用 (一对多,多对多关系生成sql脚本) , CDM 和 PDM 的区别;PD15.1 生成sql2008 无FK外键约束的解决方法
- 关于ueditor与arcgis js api同用会报错的问题
- docker安装脚本
- django中的setting最佳配置小结
- 前端技术Jquery与Ajax使用总结
热门文章
- python:set() 函数
- 数据绑定-@ CookieValue
- 关于redis的几件小事(八)缓存与数据库双写时的数据一致性
- 反selenium关键字
- VIM简单操作
- 使用sproxy.exe访问基于soap的webservice
- win 10.0.17134.915 版本无法更新处理方法
- 如何在 Ubuntu 上安装 pip
- Adaptive Synchronization of Dynamics on Evolving Complex Networks
- POJ2449 K短路模板