1.什么是ActiveMQ?

  ActiveMQ是apache提供的开源的,实现消息传递的一个中间插件,可以和spring整合,是目前最流行的开源消息总线,ActiveMQ是一个完全支持JMS1.1和J2EE 1.4规范的 JMS Provider实现。较相似的还有rabbitMQ和kafka等,都是最为消息传递的插件

2.ActiveMQ传递消息的两种方式

前提:需要引入activemq的jar包

点对点方式(PTP):一个消费者对应一个生产者

发布/订阅模式(Publish/Sub):一个生产者产生消息发送后,可以被多个消费者进行接收。

JMS定义了五种消息正文格式,以及消息的调用类型,允许发送和接收一些不同类型的数据,提供现有消息格式的一些级别的兼容性。

StreamMessage:--JAVA原始的数据流

TextMessage:一个字符串对象

ObjectMessage:一个系列化的java对象

BytesMessage:一个字节对象

MapMessage:key/value方式的键值对

(1)点对点的方式(PTP)

  即:一个消息的生产者对应一个消费者

生产者(Producer)实现步骤:

第一步:创建一个ConnectionFactory对象,将服务端activemq的 ip 和 port 作为构造参数传递

第二步:通过第一步创建的工厂对象获得连接对象Connection

第三步:开启连接,直接调用connection对象的start方法即可

第四步:创建一个Session对象,通过connection对象创建

第五步:通过Session对象创建一个Destination对象(该对象有两种方式:topic和quene),这里使用quene

第六步:通过Session对象创建一个生产者Producer对象

第七步:创建Message对象,这里使用TextMessage对象,设置消息内容

第八步:使用创建的生产者对象Producer发送消息

第九步:关闭资源(Producer对象,Connection对象,Session对象)

@Test
public void testQueueProducer() throws Exception {
// 第一步:创建ConnectionFactory对象,需要指定服务端ip及端口号。
//brokerURL服务器的ip及端口号
ConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://ip地址:61616");
// 第二步:使用ConnectionFactory对象创建一个Connection对象。
Connection connection = connectionFactory.createConnection();
// 第三步:开启连接,调用Connection对象的start方法。
connection.start();
// 第四步:使用Connection对象创建一个Session对象。
//第一个参数:是否开启事务。true:开启事务,第二个参数忽略。
//第二个参数:当第一个参数为false时,才有意义。消息的应答模式。1、自动应答2、手动应答。一般是自动应答。
Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
// 第五步:使用Session对象创建一个Destination对象(topic、queue),此处创建一个Queue对象。
//参数:队列的名称。
Queue queue = session.createQueue("test-queue");
// 第六步:使用Session对象创建一个Producer对象。
MessageProducer producer = session.createProducer(queue);
// 第七步:创建一个Message对象,创建一个TextMessage对象。
/*TextMessage message = new ActiveMQTextMessage();
message.setText("hello activeMq,this is my first test.");*/
TextMessage textMessage = session.createTextMessage("hello activeMq,this is my first test.");
// 第八步:使用Producer对象发送消息。
producer.send(textMessage);
// 第九步:关闭资源。
producer.close();
session.close();
connection.close();
}

消费者实现:

第一步:创建一个ConnectionFactory对象,将服务端activemq的 ip 和 port 作为构造参数传递

第二步:通过第一步创建的工厂对象获得连接对象Connection

第三步:开启连接,直接调用connection对象的start方法即可

第四步:创建一个Session对象,通过connection对象创建

第五步:创建一个Destination对象,使用quene,需要和生产者的quene一致

第六步:创建一个消费者对象

第七步:接收消息

第八步:打印接收的消息

第九步:关闭资源

消费者的代码:

@Test
public void testQueueConsumer() throws Exception {
// 第一步:创建一个ConnectionFactory对象。
ConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://192.168.25.168:61616");
// 第二步:从ConnectionFactory对象中获得一个Connection对象。
Connection connection = connectionFactory.createConnection();
// 第三步:开启连接。调用Connection对象的start方法。
connection.start();
// 第四步:使用Connection对象创建一个Session对象。
Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
// 第五步:使用Session对象创建一个Destination对象。和发送端保持一致queue,并且队列的名称一致。
Queue queue = session.createQueue("test-queue");
// 第六步:使用Session对象创建一个Consumer对象。
MessageConsumer consumer = session.createConsumer(queue);
// 第七步:接收消息。
consumer.setMessageListener(new MessageListener() { @Override
public void onMessage(Message message) {
try {
TextMessage textMessage = (TextMessage) message;
String text = null;
//取消息的内容
text = textMessage.getText();
// 第八步:打印消息。
System.out.println(text);
} catch (JMSException e) {
e.printStackTrace();
}
}
});
//等待键盘输入
System.in.read();
// 第九步:关闭资源
consumer.close();
session.close();
connection.close();
}

(2)订阅发布方式传递消息:Topic

补充:由于topic传递消息的特点是,一个生产者可以有多个消费者,生产者生产的消息在没有被消费者消费之前,并不会将消息持久化到activemq的服务端,发送的消息会自动消失。所以 测试的时候需要先创建消费者对象,然后在发送消息,防止消息丢失。

生产者实现步骤:

步骤和PTP的方式完全一样,不同的是在创建Destination对象的时候,需要创建topic对象

直接上代码:

@Test
public void testTopicProducer() throws Exception {
// 第一步:创建ConnectionFactory对象,需要指定服务端ip及端口号。
// brokerURL服务器的ip及端口号
ConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://IP地址:61616");
// 第二步:使用ConnectionFactory对象创建一个Connection对象。
Connection connection = connectionFactory.createConnection();
// 第三步:开启连接,调用Connection对象的start方法。
connection.start();
// 第四步:使用Connection对象创建一个Session对象。
// 第一个参数:是否开启事务。true:开启事务,第二个参数忽略。
// 第二个参数:当第一个参数为false时,才有意义。消息的应答模式。1、自动应答2、手动应答。一般是自动应答。
Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
// 第五步:使用Session对象创建一个Destination对象(topic、queue),此处创建一个topic对象。
// 参数:话题的名称。
Topic topic = session.createTopic("test-topic");
// 第六步:使用Session对象创建一个Producer对象。
MessageProducer producer = session.createProducer(topic);
// 第七步:创建一个Message对象,创建一个TextMessage对象。
/*
* TextMessage message = new ActiveMQTextMessage(); message.setText(
* "hello activeMq,this is my first test.");
*/
TextMessage textMessage = session.createTextMessage("hello activeMq,this is my topic test");
// 第八步:使用Producer对象发送消息。
producer.send(textMessage);
// 第九步:关闭资源。
producer.close();
session.close();
connection.close();
}

消费者实现的步骤:

步骤和PTP消费者实现的步骤一样,唯一不同的是在创建Destination对象的时候,创建topic对象,同时要和发布订阅的生产者的topic一致

消费者代码:

@Test
public void testTopicConsumer() throws Exception {
// 第一步:创建一个ConnectionFactory对象。
ConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://IP地址:61616");
// 第二步:从ConnectionFactory对象中获得一个Connection对象。
Connection connection = connectionFactory.createConnection();
// 第三步:开启连接。调用Connection对象的start方法。
connection.start();
// 第四步:使用Connection对象创建一个Session对象。
Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
// 第五步:使用Session对象创建一个Destination对象。和发送端保持一致topic,并且话题的名称一致。
Topic topic = session.createTopic("test-topic");
// 第六步:使用Session对象创建一个Consumer对象。
MessageConsumer consumer = session.createConsumer(topic);
// 第七步:接收消息。
consumer.setMessageListener(new MessageListener() { @Override
public void onMessage(Message message) {
try {
TextMessage textMessage = (TextMessage) message;
String text = null;
// 取消息的内容
text = textMessage.getText();
// 第八步:打印消息。
System.out.println(text);
} catch (JMSException e) {
e.printStackTrace();
}
}
});
System.out.println("topic的消费端03。。。。。");
// 等待键盘输入
System.in.read();
// 第九步:关闭资源
consumer.close();
session.close();
connection.close();
}

总结:两种传递消息的方式的异同

相同点:实现步骤基本一样,大同小异

不同点:PTP传递消息的方法,消息的生产者发送以后,消息会持久化在activemq的服务端,如果该消息给消费者消费,在服务端持久化的消息也就同时被删除。

发布订阅传递消息的方法:消息的生产者发送消息以后,如果没有消费者消费,消息不会持久化在activemq的客户端,会立即消失。如果创建的消息被消费,会的activemq的服务端显示消息相关内容。这一点和PTP刚好相反。

注意:发布订阅传递消息的方式:也是可以实现消息持久化在服务端的,需要消费者首先在activemq的服务端订阅消息(注册),将消费者客户端的ID(作为唯一标识,因为可以有多个消费者)和消息的ID传递给服务端即可。

ActiveMQ消息传递的两种方式的更多相关文章

  1. 《连载 | 物联网框架ServerSuperIO教程》- 10.持续传输大块数据流的两种方式(如:文件)

    1.C#跨平台物联网通讯框架ServerSuperIO(SSIO)介绍 <连载 | 物联网框架ServerSuperIO教程>1.4种通讯模式机制. <连载 | 物联网框架Serve ...

  2. 分布式理论基础(一)一致性及解决一致性的两种方式:2PC和3PC (转载 不错)

    分布式理论基础(一)一致性及解决一致性的两种方式:2PC和3PC 1 一致性 1.1 简述 一致性,是指对每个节点一个数据的更新,整个集群都知道更新,并且是一致的 假设一个具有N个节点的分布式系统,当 ...

  3. Struts2实现ajax的两种方式

    基于Struts2框架下实现Ajax有两种方式,第一种是原声的方式,另外一种是struts2自带的一个插件. js部分调用方式是一样的: JS代码: function testAjax() { var ...

  4. CSharpGL(18)分别处理glDrawArrays()和glDrawElements()两种方式下的拾取(ColorCodedPicking)

    CSharpGL(18)分别处理glDrawArrays()和glDrawElements()两种方式下的拾取(ColorCodedPicking) 我在(Modern OpenGL用Shader拾取 ...

  5. 两种方式实现java生成Excel

    Web应用中难免会遇到需要将数据导出并生成excel文件的需求.同样,对于本博客中的总结,也是建立在为了完成这样的一个需求,才开始去了解其实现形式,并且顺利完成需求的开发,先将实现过程总结于此.本博文 ...

  6. Android ScrollView监听滑动到顶部和底部的两种方式(你可能不知道的细节)

    Android ScrollView监听滑动到顶部和底部,虽然网上很多资料都有说,但是不全,而且有些细节没说清楚 使用场景: 1. 做一些复杂动画的时候,需要动态判断当前的ScrollView是否滚动 ...

  7. 在基于MVC的Web项目中使用Web API和直接连接两种方式混合式接入

    在我之前介绍的混合式开发框架中,其界面是基于Winform的实现方式,后台使用Web API.WCF服务以及直接连接数据库的几种方式混合式接入,在Web项目中我们也可以采用这种方式实现混合式的接入方式 ...

  8. (转)SqlServer 数据库同步的两种方式 (发布、订阅),主从数据库之间的同步

    最近在琢磨主从数据库之间的同步,公司正好也需要,在园子里找了一下,看到这篇博文比较详细,比较简单,本人亲自按步骤来过,现在分享给大家. 在这里要提醒大家的是(为了更好的理解,以下是本人自己理解,如有错 ...

  9. Xamarin Android Activity全屏的两种方式

    方式一 直接在Activity的Attribute中定义 如下 在 MainActivity 中 [Activity(Label = "app", MainLauncher = t ...

随机推荐

  1. MVC View中获取Url参数的值

    如果url是 /home/index?id=3 直接Request就ok. Razor方法 @Html.ViewContext.RouteData.Values["id"] @Re ...

  2. java_文件类

    例一:创建文件 package java文件类; import java.io.File; import java.io.IOException; public class aasd { public ...

  3. python--day4--迭代器、生成器

    列表生成式: 需求:列表[1,2,3,4,5,6,7,8,9]每个值加1,实现的方法: a = [0,1,2,3,4,5,6,7,8,9] b = [] for i in a:b.append(i+1 ...

  4. CodeForces 701C They Are Everywhere

    双指针. 先统计一下字符种类数$sum$,然后进行尺取. 如果目前的区间$[L,R]$中不同字符个数$k<sum$,那么区间右端往右移动一位,更新种类数. 如果目前的区间$[L,R]$中不同字符 ...

  5. Yii AR中处理多表关联的relations配置

    关系型 Active Record官方文档中指出: 两张表之间的关联是根据外键来的,但是这种外键关联虽然在数据容错方面有益处,但是在性能上是个损伤,所以,一般是不定义外键的. 这种情况下,他们之间的关 ...

  6. 洛谷P1072 Hankson 的趣味题(题解)

    https://www.luogu.org/problemnew/show/P1072(题目传送) 数学的推理在编程的体现越来越明显了.(本人嘀咕) 首先,我们知道这两个等式: (a0,x)=a1,[ ...

  7. python经典程序

    输入输出 #判断输入整数是否在[0,100]之间 num = eval(input("请输入一个整数:")) if num > 100 or num < 0: #判断[ ...

  8. 文本编辑器 EditPlus 的激活与设置

    说明 1.EditPlus是Windows的文本编辑器,具有内置的FTP,FTPS和sftp功能.虽然它可以作为一个很好的记事本替代品,但它也为网页作者和程序员提供了许多强大的功能. 2.语法高亮显示 ...

  9. day_5.25py

    作用域

  10. 使用windowAnimations定义Activity及Dialog的进入退出效果

    看了android的源代码和资源文件,终于明白如何去修改设置Dialog和Activity的进入和退出效果了.设置Dialog首先通过getWindow()方法获取它的窗口,然后通过getAttrib ...