ActiveMQ

什么是activeMQ

activeMQ是一种开源的,实现了JMS1.1规范的,面向消息(MOM)的中间件,为应用程序提供高效的、可扩展的、稳定的和安全的企业级消息通信。

ActiveMQ的作用以及原理

Activemq 的作用就是系统之间进行通信。 当然可以使用其他方式进行系统间通信, 如果使用 Activemq 的话可以对系统之间的调用进行解耦, 实现系统间的异步通信。 原理就是生产者生产消息, 把消息发送给activemq。 Activemq 接收到消息, 然后查看有多少个消费者, 然后把消息转发给消费者, 此过程中生产者无需参与。 消费者接收到消息后做相应的处理和生产者没有任何关系

activemq的几种通信方式

(1)publish(发布)-subscribe(订阅)(发布-订阅方式)

发布/订阅方式用于多接收客户端的方式.作为发布订阅的方式,可能存在多个接收客户端,并且接收端客户端与发送客户端存在时间上的依赖。一个接收端只能接收他创建以后发送客户端发送的信息。作为subscriber ,在接收消息时有两种方法,destination的receive方法,和实现message listener 接口的onMessage 方法。

(2)p2p(point-to-point)(点对点)

p2p的过程则理解起来比较简单。它好比是两个人打电话,这两个人是独享这一条通信链路的。一方发送消息,另外一方接收,就这么简单。

ActiveMQ服务器宕机怎么办?

这得从ActiveMQ的储存机制说起。在通常的情况下,非持久化消息是存储在内存中的,持久化消息是存储在文件中的,它们的最大限制在配置文件的<systemUsage>节点中配置。但是,在非持久化消息堆积到一定程度,内存告急的时候,ActiveMQ会将内存中的非持久化消息写入临时文件中,以腾出内存。虽然都保存到了文件里,但它和持久化消息的区别是,重启后持久化消息会从文件中恢复,非持久化的临时文件会直接删除。

解决方案:尽量不要用非持久化消息,非要用的话,将临时文件限制尽可能的调大。

丢消息怎么办?

这得从java的java.net.SocketException异常说起。简单点说就是当网络发送方发送一堆数据,然后调用close关闭连接之后。这些发送的数据都在接收者的缓存里,接收者如果调用read方法仍旧能从缓存中读取这些数据,尽管对方已经关闭了连接。但是当接收者尝试发送数据时,由于此时连接已关闭,所以会发生异常,这个很好理解。不过需要注意的是,当发生SocketException后,原本缓存区中数据也作废了,此时接收者再次调用read方法去读取缓存中的数据,就会报Software caused connection abort: recv failed错误。

通过抓包得知,ActiveMQ会每隔10秒发送一个心跳包,这个心跳包是服务器发送给客户端的,用来判断客户端死没死。如果你看过上面第一条,就会知道非持久化消息堆积到一定程度会写到文件里,这个写的过程会阻塞所有动作,而且会持续20到30秒,并且随着内存的增大而增大。当客户端发完消息调用connection.close()时,会期待服务器对于关闭连接的回答,如果超过15秒没回答就直接调用socket层的close关闭tcp连接了。这时客户端发出的消息其实还在服务器的缓存里等待处理,不过由于服务器心跳包的设置,导致发生了java.net.SocketException异常,把缓存里的数据作废了,没处理的消息全部丢失。

解决方案:用持久化消息,或者非持久化消息及时处理不要堆积,或者启动事务,启动事务后,commit()方法会负责任的等待服务器的返回,也就不会关闭连接导致消息丢失了。

持久化消息非常慢

默认的情况下,非持久化的消息是异步发送的,持久化的消息是同步发送的,遇到慢一点的硬盘,发送消息的速度是无法忍受的。但是在开启事务的情况下,消息都是异步发送的,效率会有2个数量级的提升。所以在发送持久化消息时,请务必开启事务模式。其实发送非持久化消息时也建议开启事务,因为根本不会影响性能。

消息的不均匀消费

有时在发送一些消息之后,开启2个消费者去处理消息。会发现一个消费者处理了所有的消息,另一个消费者根本没收到消息。原因在于ActiveMQ的prefetch机制。当消费者去获取消息时,不会一条一条去获取,而是一次性获取一批,默认是1000条。这些预获取的消息,在还没确认消费之前,在管理控制台还是可以看见这些消息的,但是不会再分配给其他消费者,此时这些消息的状态应该算作“已分配未消费”,如果消息最后被消费,则会在服务器端被删除,如果消费者崩溃,则这些消息会被重新分配给新的消费者。但是如果消费者既不消费确认,又不崩溃,那这些消息就永远躺在消费者的缓存区里无法处理。更通常的情况是,消费这些消息非常耗时,你开了10个消费者去处理,结果发现只有一台机器吭哧吭哧处理,另外9台啥事不干。

解决方案:将prefetch设为1,每次处理1条消息,处理完再去取,这样也慢不了多少。

死信队列

如果你想在消息处理失败后,不被服务器删除,还能被其他消费者处理或重试,可以关闭AUTO_ACKNOWLEDGE,将ack交由程序自己处理。那如果使用了AUTO_ACKNOWLEDGE,消息是什么时候被确认的,还有没有阻止消息确认的方法?有!

消费消息有2种方法,一种是调用consumer.receive()方法,该方法将阻塞直到获得并返回一条消息。这种情况下,消息返回给方法调用者之后就自动被确认了。另一种方法是采用listener回调函数,在有消息到达时,会调用listener接口的onMessage方法。在这种情况下,在onMessage方法执行完毕后,消息才会被确认,此时只要在方法中抛出异常,该消息就不会被确认。那么问题来了,如果一条消息不能被处理,会被退回服务器重新分配,如果只有一个消费者,该消息又会重新被获取,重新抛异常。就算有多个消费者,往往在一个服务器上不能处理的消息,在另外的服务器上依然不能被处理。难道就这么退回--获取--报错死循环了吗?

在重试6次后,ActiveMQ认为这条消息是“有毒”的,将会把消息丢到死信队列里。如果你的消息不见了,去ActiveMQ.DLQ里找找,说不定就躺在那里。

java消息服务

不同系统之间的信息交换,是我们开发中比较常见的场景,比如系统A要把数据发送给系统B,这个问题我们应该如何去处理? 1999年,原来的SUN公司领衔提出了一种面向消息的中间件服务--JMS规范(标准);常用的几种信息交互技术(httpClient、hessian、dubbo、jms、webservice 五种).

JMS概述

JMS即Java消息服务(Java Message Service的简称),是Java EE 的标准/规范之一。这种规范(标准)指出:消息的发送应该是异步的、非阻塞的。也就是说消息的发送者发送完消息后就直接返回了,不需要等待接收者返回后才能返回,发送者和接收者可以说是互不影响。所以这种规范(标准)能够减轻或消除系统瓶颈,实现系统之间去除耦合,提高系统的整体可伸缩性和灵活性。JMS只是Java EE中定义的一组标准API,它自身并不是一个消息服务系统,它是消息传送服务的一个抽象,也就是说它定义了消息传送的接口而并没有具体实现。

ActiveMQ概述:

我们知道JMS只是消息服务的一组规范和接口,并没有具体的实现,而ActiveMQ就是JMS规范的具体实现;它是Apache下的一个项目,采用Java语言开发;是一款非常流行的开源消息服务器.

ActiveMQ与JMS关系

我们知道,JMS只是定义了一组有关消息传送的规范和标准,并没有真正实现,也就说JMS只是定义了一组接口而已;就像JDBC抽象了关系数据库访问、JPA抽象了对象与关系数据库映射、JNDI抽象了命名目录服务访问一样,JMS具体的实现由不同的消息中间件厂商提供,比如Apache ActiveMQ就是JMS规范的具体实现,Apache ActiveMQ才是一个消息服务系统,而JMS不是。

基本要素

1、生产者producer ; 2、消费者consumer ; 3、消息服务broker

JMS两种消息传送模式

点对点( Point-to-Point):专门用于使用队列Queue传送消息;基于队列Queue的点对点消息只能被一个消费者消费,如多个消费者都注册到同一个消息队列上,当生产者发送一条消息后,而只有其中一个消费者会接收到该消息,而不是所有消费者都能接收到该消息。

发布/订阅(Publish/Subscribe):专门用于使用主题Topic传送消息。基于主题的发布与订阅消息能被多个消费者消费,生产者发送的消息,所有订阅了该topic的消费者都能接收到。

JMS API 概览

JMS API可以分为3个主要部分:

1、公共API:可用于向一个队列或主题发送消息或从其中接收消息;

2、点对点API:专门用于使用队列Queue传送消息;

3、发布/订阅API:专门用于使用主题Topic传送消息。

JMS公共API

在JMS公共API内部,和发送与接收消息有关的JMS API接口主要是:ConnectionFactory / Connection / Session / Message / Destination / MessageProducer / MessageConsumer . 它们的关系是:一旦有了ConnectionFactory,就可以创建Connection,一旦有了Connection,就可以创建Session,而一旦有了Session,就可以创建 Message 、MessageProducer 和 MessageConsumer 。

JMS点对点API

点对点(p2p)消息传送模型API是指JMS API之内基于队列(Queue)的接口:QueueConnectionFactory / QueueConnection / QueueSession / Message / Queue / QueueSender / QueueReceiver .

从接口的命名可以看出,大多数接口名称仅仅是在公共API接口

名称之前添加Queue一词。一般来说,使用点对点消息传送模型的应用程序将使用基于队列的API,而不使用公共API 。

JMS发布/订阅API发布/订阅消息传送模型API是指JMS

API之内基于主题(Topic)的接口:TopicConnectionFactory / TopicConnection /

TopicSession / Message / Topic / TopicPublisher / TopicSubscriber . 由于基于主题(Topic)的JMS API类似于基于队列(Queue)

的API,因此在大多数情况下,Queue这个词会由Topic取代。

ActiveMQ点对点发送与接收消息示例

简单示例

发送

package com.kinglong.activemq.queue; 

import org.apache.activemq.ActiveMQConnectionFactory; 

import javax.jms.*; 

/** 
* 消息发送者 
* 
*/ 
public class Sender { 

/**消息服务器的连接地址**/ 
public static final String BROKER_URL = "tcp://192.168.174.129:61616"; 

public static void main(String[] args) { 
Sender sender = new Sender(); 
sender.sendMessage("Hello ActiveMQ."); 
} 

/** 
* 发送消息 
* 
* @param msg 
*/ 
public void sendMessage (String msg) { 

Connection connection = null; 
Session session = null; 
MessageProducer messageProducer = null; 

try { 
//1.创建一个连接工厂 
ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(BROKER_URL); 

//2.创建一个连接 
connection = connectionFactory.createConnection(); 

//3.创建一个Session 
session = connection.createSession(Boolean.FALSE, Session.AUTO_ACKNOWLEDGE); 

//4.创建消息,此处创建了一个文本消息 
Message message = session.createTextMessage(msg); 

//5.创建一个目的地 
Destination destination = session.createQueue("myQueue"); 

//6.创建一个消息的生产者(发送者) 
messageProducer = session.createProducer(destination); 

//7.发送消息 
messageProducer.send(message); 

} catch (JMSException e) { 
e.printStackTrace(); 
} finally { 
try { 
//关闭连接释放资源 
if (null != messageProducer) { 
messageProducer.close(); 
} 
if (null != session) { 
session.close(); 
} 
if (null != connection) { 
connection.close(); 
} 
} catch (JMSException e) { 
e.printStackTrace(); 
} 
} 
} 
}

接收

package com.kinglong.activemq.queue; 

import org.apache.activemq.ActiveMQConnectionFactory; 

import javax.jms.*; 

public class Receiver { 

/**消息服务器的连接地址**/ 
public static final String BROKER_URL = "tcp://192.168.174.129:61616"; 

public static void main(String[] args) { 
Receiver receiver = new Receiver(); 
receiver.receiveMessage(); 
} 

/** 
* 接收消息 
* 
*/ 
public void receiveMessage () { 

Connection connection = null; 
Session session = null; 
MessageConsumer messageConsumer = null; 

try { 
//1.创建一个连接工厂 
ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(BROKER_URL); 

//2.创建一个连接 
connection = connectionFactory.createConnection(); 

//3.创建一个Session 
session = connection.createSession(Boolean.FALSE, Session.AUTO_ACKNOWLEDGE); 

//4.创建一个目的地 
Destination destination = session.createQueue("myQueue"); 

//5.创建一个消息的消费者(接收者) 
messageConsumer = session.createConsumer(destination); 

//接收消息之前,需要把连接启动一下 
connection.start(); 

//6.接收消息 
Message message = messageConsumer.receive(); 

//判断消息的类型 
if (message instanceof TextMessage) { //判断是否是文本消息 
String text = ((TextMessage) message).getText(); 
System.out.println("接收到的消息内容是:" + text); 
} 
} catch (JMSException e) { 
e.printStackTrace(); 
} finally { 
try { 
//关闭连接释放资源 
if (null != messageConsumer) { 
messageConsumer.close(); 
} 
if (null != session) { 
session.close(); 
} 
if (null != connection) { 
connection.close(); 
} 
} catch (JMSException e) { 
e.printStackTrace(); 
} 
} 
} 
}

总结

1. session = connection.createSession(Boolean.FALSE, Session.AUTO_ACKNOWLEDGE); 
// 其中:Boolean.FALSE表示本次会话不开启事务管理,假如需要开启事务管理,将其改为Boolean.TRUE即可 
//同时需要在发送消息后添加session.commit(),否则,消息是不会被提交的. 
//Session.AUTO_ACKNOWLEDGE表示消息确认机制 
AUTO_ACKNOWLEDGE:自动确认 
CLIENT_ACKNOWLEDGE:客户端确认 
SESSION_TRANSACTED:事务确认,如果使用事务推荐使用该确认机制 
AUTO_ACKNOWLEDGE:懒散式确认,消息偶尔不会被确认,也就是消息可能会被重复发送.但发生的概率很小 
2. connection.start(); 
//在消息接收端,接受消息前需要加入这段代码,开启连接,否则一样无法获取消息. 
3. Destination destination = session.createQueue("myQueue"); 
//创建目的地时,如果做测试收不到信息,可以将目的地名称修改一下,我用的是IDEA,不清楚为何, 
//有时候收不到信息,修改一下就好了,猜测可能是缓存的原因吧

发布与订阅的topic方式实际与点对点的queue方式,代码通用很多,只是在创建目的地Destination时候创建为

Destination destination = session.createTopic("myTopic

Queue与Topic比较

比较项 队列 主题
概要说明 point-to-ponit点对点小欻 publish/subscribe发布订阅消息
消息完整性保障 队列保证每条数据都能被receiver接收 不保证publisher发布的每条数据,subscriber都能接收到
消息是否会丢失 sender发送消息到目标队列,receiver可以同步或者异步接收到这个队列上的消息,队列上的消息如果暂时没有被receiver接收到,也不会丢失 publisher发布消息到目标主题时,只有正在监听该主题的subscriber能够接收到消息,如果目标topic没有subscriber在监听,该主题上的消息就会丢失。
消息传送策略 一对一的消息发布接受策略,一个sender发送的消息,只能有一个receiver接收,receiver接收到后,通知mq服务器已接收,mq服务器对队列里的消息采取删除或其他操作。 一对多的消息发布接收策略,监听同一个主题地址的多个subscriber都能接收到publisher发送的消息,subscriber接收完通知mq服务器。

拉模式与推模式

a.点对点消息,如果没有消费者在监听队列,消息将保留在队列中,直至消息消费者连接到队列为止。这种消息传递模型是

传统意义上的懒模型或轮询模型。在此模型中,消息不是自动推动给消息消费者的,而是要由消息消费者从队列中请求获得(拉模式)。

b.pub/sub消息传递模型基本上是一个推模型。在该模型中,消息会自动广播,消息消费者无须通过主动请求或轮询主题的方法来获得新的消息。

ActiveMQ消息类型

1、TextMessage 文本消息:携带一个java.lang.String作为有效数据(负载)的消息,可用于字符串类型的信息交换;

2、ObjectMessage 对象消息:携带一个可以序列化的Java对象作为有效负载的消息,可用于Java对象类型的信息交换;

3、MapMessage 映射消息:携带一组键值对的数据作为有效负载的消息,有效数据值必须是Java原始数据类型(或者它们的包装类)及String。即:byte , short , int , long , float , double , char , boolean , String

4、BytesMessage 字节消息 :携带一组原始数据类型的字节流作为有效负载的消息;

5、StreamMessage 流消息:携带一个原始数据类型流作为有效负载的消息,它保持了写入流时的数据类型,写入什么类型,则读取也需要是相同的类型;

ActiveMQ事务消息和非事务消息

消息分为事务消息和非事务消息

1、事务消息:创建会话Session使用transacted=true

connection.createSession(Boolean.TRUE, Session.AUTO_ACKNOWLEDGE);

2、非事务消息:创建会话Session使用transacted=false

connection.createSession(Boolean.FALSE, Session.AUTO_ACKNOWLEDGE);

事务消息必须在发送和接收完消息后显式地调用session.commit();

事务性消息,不管设置何种消息确认模式,都会自动被确认;与设置的确认机制无关,但官方推荐事务性消息使用事务确认机制.

ActiveMQ消息确认机制

消息只有在被确认之后,才认为已经被成功消费,然后消息才会从队列或主题中删除。消息的成功消费通常包含三个阶段:


(1)、客户接收消息;(2)、客户处理消息;(3)、消息被确认;

确认机制前面提过一下,共有四种:

(1)、Session.AUTO_ACKNOWLEDGE;客户(消费者)成功从receive方法返回时,或者从MessageListener.onMessage方法成功返回时,会话自动确认消息,然后自动删除消息.

(2)、Session.CLIENT_ACKNOWLEDGE;客户通过显式调用消息的acknowledge方法确认消息,。 即在接收端调用message.acknowledge();方法,否则,消息是不会被删除的.

(3)、Session. DUPS_OK_ACKNOWLEDGE ;不是必须确认,是一种“懒散的”消息确认,消息可能会重复发送,在第二次重新传送消息时,消息头的JMSRedelivered会被置为true标识当前消息已经传送过一次,客户端需要进行消息的重复处理控制。

(4)、 Session.SESSION_TRANSACTED;事务提交并确认。

ActiveMQ持久化消息与非持久化消息

messageProducer.setDeliveryMode(DeliveryMode. NON_PERSISTENT);//不持久化
messageProducer.setDeliveryMode(DeliveryMode.
PERSISTENT);//持久化的,当然activemq发送消息默认都是持久化的

说明:

设置完后,如果为持久化,那么消息在没有被消费前都会被写入本地磁盘kahadb文件中保存起来,即使服务器宕机,也不会影响

消息.如果是非持久化的,那么,服务一旦宕机之类的情况发生,消息即会被删除.

ActiveMQ默认是持久化的.

ActiveMQ消息过滤

ActiveMQ提供了一种机制,可根据消息选择器中的标准来执行消息过滤,只接收符合过滤标准的消息;

生产者可在消息中放入特有的标志,而消费者使用基于这些特定的标志来接收消息;

1、发送消息放入特殊标志:message . setString Property ( name , value ) ;

2、接收消息使用基于特殊标志的消息选择器:

MessageConsumer createConsumer(Destination destination, String messageSelector);

注:消息选择器是一个字符串,语法与数据库的SQL相似,相当于SQL语句where条件后面的内容;

具体代码如下:

发送端代码:

package com.bjpowernode.activemq.selector; 

import org.apache.activemq.Active

MQConnectionFactory; 

import javax.jms.*; 

/** 
* 消息发送者 
* 
*/ 
public class Sender { 

/**消息服务器的连接地址**/ 
public static final String BROKER_URL = "tcp://192.168.174.129:61616"; 

public static void main(String[] args) { 
Sender sender = new Sender(); 
sender.sendMessage("Hello ActiveMQ."); 
} 

/** 
* 发送消息 
* 
* @param msg 
*/ 
public void sendMessage (String msg) { 

Connection connection = null; 
Session session = null; 
MessageProducer messageProducer = null; 

try { 
//1.创建一个连接工厂 
ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(BROKER_URL); 

//2.创建一个连接 
connection = connectionFactory.

createConnection(); 

//3.创建一个Session 
session = connection.createSession(Boolean.FALSE, Session.AUTO_ACKNOWLEDGE); 



//5.创建一个目的地 
Destination destination = session.create

Queue("myQueue"); 

//6.创建一个消息的生产者(发送者) 
messageProducer = session.createProducer

(destination); 

//设置发送的消息是否需要持久化 
messageProducer.setDeliveryMode(Delivery

Mode.NON_PERSISTENT);//这里使用不持久化 

//创建一个循环,测试消息标识的使用 
for (int i=0; i<20; i++) { 
//4.创建消息,此处创建了一个文本消息 
Message message = session.createText

Message(msg+i); 

//将消息设置一个特有的标识 
message.setIntProperty("id", i); 

//7.发送消息 
messageProducer.send(message); 
} 
} catch (JMSException e) { 
e.printStackTrace(); 
} finally { 
try { 
//关闭连接释放资源 
if (null != messageProducer) { 
messageProducer.close(); 
} 
if (null != session) { 
session.close(); 
} 
if (null != connection) { 
connection.close(); 
} 
} catch (JMSException e) { 
e.printStackTrace(); 
} 
} 
} 
}

接收端代码:

package com.bjpowernode.activemq.selector; 

import org.apache.activemq.ActiveMQ

ConnectionFactory; 

import javax.jms.*; 

public class Receiver { 

/**消息服务器的连接地址**/ 
public static final String BROKER_URL = "tcp://192.168.174.129:61616"; 

public static void main(String[] args) { 
Receiver receiver = new Receiver(); 
receiver.receiveMessage(); 
} 

/** 
* 接收消息 
* 
*/ 
public void receiveMessage () { 

Connection connection = null; 
Session session = null; 
MessageConsumer messageConsumer = null; 

try { 
//1.创建一个连接工厂 
ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(BROKER_URL); 

//2.创建一个连接 
connection = connectionFactory.

createConnection(); 

//3.创建一个Session 
session = connection.createSession

(Boolean.FALSE, Session.AUTO_ACKNOWLEDGE); 

//4.创建一个目的地 
Destination destination = session.createQueue("myQueue"); 

//5.创建一个消息的消费者(接收者),

selector即为消息选择器,通过选择需要的标识,过滤消息接受id为10-15之 //间的消息 
String selector = "id >=10 and id<=15"; 
messageConsumer = session.createConsumer

(destination, selector); 

//接收消息之前,需要把连接启动一下 
connection.start(); 

while (true) { 
//6.接收消息 同步接收 
Message message = messageConsumer.

receive(); 

//判断消息的类型 
if (message instanceof TextMessage)

{ //判断是否是文本消息 
String text = ((TextMessage) message).getText(); 
System.out.println("接收到的消息内容是:" + text); 
} 
} 
} catch (JMSException e) { 
e.printStackTrace(); 
} finally { 
try { 
//关闭连接释放资源 
if (null != messageConsumer) { 
messageConsumer.close(); 
} 
if (null != session) { 
session.close(); 
} 
if (null != connection) { 
connection.close(); 
} 
} catch (JMSException e) { 
e.printStackTrace(); 
} 
} 
} 
}

ActiveMQ消息接收方式

同步接收:receive()方法接收消息叫同步接收,就是之前的Demo代码使用的接收方式.在不使用循环方法时接收端代码执行

一次即结束.

异步接收:使用监听器接收消息,这种接收方式叫异步接收,接收端会一直处于监听状态,只要有消息产生,即会接收消息.

下面是异步接收代码:

package com.bjpowernode.activemq.listener; 

import org.apache.activemq.ActiveMQConnectionFactory; 

import javax.jms.*; 

public class Receiver { 

/**消息服务器的连接地址**/ 
public static final String BROKER_URL = "tcp://192.168.174.129:61616"; 

public static void main(String[] args) { 
Receiver receiver = new Receiver(); 
receiver.receiveMessage(); 
} 

/** 
* 接收消息 
* 
*/ 
public void receiveMessage () { 

Connection connection = null; 
Session session = null; 
MessageConsumer messageConsumer = null; 

try { 
//1.创建一个连接工厂 
ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(BROKER_URL); 

//2.创建一个连接 
connection = connectionFactory.createConnection(); 

//3.创建一个Session 
session = connection.createSession(Boolean.FALSE, Session.AUTO_ACKNOWLEDGE); 

//4.创建一个目的地 
Destination destination = session.createQueue("myQueue"); 

//5.创建一个消息的消费者(接收者) 
messageConsumer = session.createConsumer(destination); 

//接收消息之前,需要把连接启动一下 
connection.start(); 

//6.接收消息 同步接收 
//Message message = messageConsumer.receive(); 

//异步接收,使用监听器接收消息 
messageConsumer.setMessageListener(new MessageListener(){ 
public void onMessage(Message message) { 
//判断消息的类型 
if (message instanceof TextMessage) { //判断是否是文本消息 
String text = null; 
try { 
text = ((TextMessage) message).getText(); 
} catch (JMSException e) { 
e.printStackTrace(); 
} 
System.out.println("接收到的消息内容是:" + text); 
} 
} 
}); 
} catch (JMSException e) { 
e.printStackTrace(); 
} finally { 
/*try { 
//关闭连接释放资源 
if (null != messageConsumer) { 
messageConsumer.close(); 
} 
if (null != session) { 
session.close(); 
} 
if (null != connection) { 
connection.close(); 
} 
} catch (JMSException e) { 
e.printStackTrace(); 
}*/ 
} 
} 
}
最后编辑于
©著作权归作者所有,转载或内容合作请联系作者
  • 序言:七十年代末,一起剥皮案震惊了整个滨河市,随后出现的几起案子,更是在滨河造成了极大的恐慌,老刑警刘岩,带你破解...
    沈念sama阅读 206,723评论 6 481
  • 序言:滨河连续发生了三起死亡事件,死亡现场离奇诡异,居然都是意外死亡,警方通过查阅死者的电脑和手机,发现死者居然都...
    沈念sama阅读 88,485评论 2 382
  • 文/潘晓璐 我一进店门,熙熙楼的掌柜王于贵愁眉苦脸地迎上来,“玉大人,你说我怎么就摊上这事。” “怎么了?”我有些...
    开封第一讲书人阅读 152,998评论 0 344
  • 文/不坏的土叔 我叫张陵,是天一观的道长。 经常有香客问我,道长,这世上最难降的妖魔是什么? 我笑而不...
    开封第一讲书人阅读 55,323评论 1 279
  • 正文 为了忘掉前任,我火速办了婚礼,结果婚礼上,老公的妹妹穿的比我还像新娘。我一直安慰自己,他们只是感情好,可当我...
    茶点故事阅读 64,355评论 5 374
  • 文/花漫 我一把揭开白布。 她就那样静静地躺着,像睡着了一般。 火红的嫁衣衬着肌肤如雪。 梳的纹丝不乱的头发上,一...
    开封第一讲书人阅读 49,079评论 1 285
  • 那天,我揣着相机与录音,去河边找鬼。 笑死,一个胖子当着我的面吹牛,可吹牛的内容都是我干的。 我是一名探鬼主播,决...
    沈念sama阅读 38,389评论 3 400
  • 文/苍兰香墨 我猛地睁开眼,长吁一口气:“原来是场噩梦啊……” “哼!你这毒妇竟也来了?” 一声冷哼从身侧响起,我...
    开封第一讲书人阅读 37,019评论 0 259
  • 序言:老挝万荣一对情侣失踪,失踪者是张志新(化名)和其女友刘颖,没想到半个月后,有当地人在树林里发现了一具尸体,经...
    沈念sama阅读 43,519评论 1 300
  • 正文 独居荒郊野岭守林人离奇死亡,尸身上长有42处带血的脓包…… 初始之章·张勋 以下内容为张勋视角 年9月15日...
    茶点故事阅读 35,971评论 2 325
  • 正文 我和宋清朗相恋三年,在试婚纱的时候发现自己被绿了。 大学时的朋友给我发了我未婚夫和他白月光在一起吃饭的照片。...
    茶点故事阅读 38,100评论 1 333
  • 序言:一个原本活蹦乱跳的男人离奇死亡,死状恐怖,灵堂内的尸体忽然破棺而出,到底是诈尸还是另有隐情,我是刑警宁泽,带...
    沈念sama阅读 33,738评论 4 324
  • 正文 年R本政府宣布,位于F岛的核电站,受9级特大地震影响,放射性物质发生泄漏。R本人自食恶果不足惜,却给世界环境...
    茶点故事阅读 39,293评论 3 307
  • 文/蒙蒙 一、第九天 我趴在偏房一处隐蔽的房顶上张望。 院中可真热闹,春花似锦、人声如沸。这庄子的主人今日做“春日...
    开封第一讲书人阅读 30,289评论 0 19
  • 文/苍兰香墨 我抬头看了看天上的太阳。三九已至,却和暖如春,着一层夹袄步出监牢的瞬间,已是汗流浃背。 一阵脚步声响...
    开封第一讲书人阅读 31,517评论 1 262
  • 我被黑心中介骗来泰国打工, 没想到刚下飞机就差点儿被人妖公主榨干…… 1. 我叫王不留,地道东北人。 一个月前我还...
    沈念sama阅读 45,547评论 2 354
  • 正文 我出身青楼,却偏偏与公主长得像,于是被迫代替她去往敌国和亲。 传闻我的和亲对象是个残疾皇子,可洞房花烛夜当晚...
    茶点故事阅读 42,834评论 2 345

推荐阅读更多精彩内容