0032ActiveMQ之java编码实现生产者和消费者操作队列queue
作者：互联网
今天学习了入门级的用java编写生产者producer和消费者consumer操作activemq的queue队列，为了之后复习回顾，现做整理如下：
maven工程的搭建此处不再讲解，在maven工程已搭建好的前提下进行如下步骤：
1、 pom.xml中引入activemq所需的两个jar包
<dependency>
<groupId>org.apache.activemq</groupId>
<artifactId>activemq-all</artifactId>
<version>5.15.9</version>
</dependency>
<dependency>
<groupId>org.apache.xbean</groupId>
<artifactId>xbean-spring</artifactId>
<version>3.16</version>
</dependency>
2、 JMS编码总体架构
3、生产者编码如下：
import org.apache.activemq.ActiveMQConnectionFactory;
import javax.jms.*;
public class JmsProduce {
public static
final String ACTIVEMQ_URL = "tcp://192.168.225.132:61616";
public static void main(String[] args) throws JMSException {
//1 按照给定的url地址，采用默认的用户名密码创建连接工厂（有三个参数的构造方法，其中两个参数是用户名密码）
ActiveMQConnectionFactory
activeMQConnectionFactory = new ActiveMQConnectionFactory(ACTIVEMQ_URL);
//2 通过连接工厂获得连接
Connection connection =
activeMQConnectionFactory.createConnection();
//3 启动访问，真正建立连接，需要确保防火墙、服务端配置等都正确，能正常访问
connection.start();
//4 创建会话session，false代表不开启事物，Session.AUTO_ACKNOWLEDGE代表自动签收
Session session =
connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
//5 创建目的地（具体是队列queue还是主题topic）,参数是队列的名称
//返回类型可以用Queue接收，也可以用Destination接收，因为Queue继承Destination
Queue queue = session.createQueue("queue1");
//6 创建消息的生产者，负责向指定队列中生产消息
MessageProducer producer =
session.createProducer(queue);
//7 创建3条消息，并由生产者发送到目的地
for(int i=1;i<=3;i++){
//创建消息
TextMessage textMessage =
session.createTextMessage("message" + i);
//发送消息
producer.send(textMessage);
}
//8 关闭资源
producer.close();
session.close();
connection.close();
System.out.println("生产者向activemq生产消息完成！");
}
}
4、控制台查看队列中是否有消息存在，如下图：
各列的含义分别是：待消费的消息数、消费者数量、消息入队数量、消息出队数量。
-----------------------------------------以上是生产者生产消息的简单过程-----------------------------
------------------------------------消费者消费消息的过程------------------------------------------------------
1、编码如下（不带等待时间）：
import org.apache.activemq.ActiveMQConnectionFactory;
import javax.jms.*;
public class JmsConsumer {
public static
final String ACTIVEMQ_URL = "tcp://192.168.225.132:61616";
public static void main(String[] args) throws JMSException {
//1 按照给定的url地址，采用默认的用户名密码创建连接工厂（有三个参数的构造方法，其中两个参数是用户名密码）
ActiveMQConnectionFactory
activeMQConnectionFactory = new ActiveMQConnectionFactory(ACTIVEMQ_URL);
//2 通过连接工厂获得连接
Connection connection =
activeMQConnectionFactory.createConnection();
//3 启动访问，真正建立连接，需要确保防火墙、服务端配置等都正确，能正常访问
connection.start();
//4 创建会话session，false代表不开启事物，Session.AUTO_ACKNOWLEDGE代表自动签收
Session session =
connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
//5 创建目的地（具体是队列queue还是主题topic）,参数是队列的名称
//返回类型可以用Queue接收，也可以用Destination接收，因为Queue继承Destination
Queue queue = session.createQueue("queue1");
MessageConsumer consumer =
session.createConsumer(queue);
//
while(true){
//receive()方法可以带一个参数，就是如果没有待消费的消息，最多等多久
//不待参数的receive()方法会一直等待，直到接收到消息，才会进行后边处理
TextMessage message =
(TextMessage)consumer.receive();
if(null == message){
break;
}
String s = message.getText();
System.out.println(s);
}
//8 关闭资源
consumer.close();
session.close();
connection.close();
System.out.println("消费者从activemq消费消息完成！");
}
}
发现待消费的消息为0，有一个消费者，入队的消息3条，出队的消息3条
如果不让消费者在得不到消息的情况下还一直等待，即断开消费者与queue之间的连接，或者receive(n*1000L)等待n秒后不再等待，则控制台显示的信息如下（显示的消费者会由1变为0）：
2 、测试含有等待时间的receive()方法，代码更改如下：
//TextMessage message = (TextMessage)consumer.receive();
TextMessage
message = (TextMessage)consumer.receive(5000L);
生产者重新生产3条消息，控制台显示如下：
消费者换成含有等待时间的接收消息的方法，控制台显示如下：
将消费者处理完消息后，如果等待5秒还没有获得消息，则断开连接。
以上两种消费方式即：receive()方法不带参数，一直等待（生产者新生产了消息，消费者就会消费掉消息）；带参数，超过等待时间就断开连接
3、 消费者以监听器的形式监听队列
import org.apache.activemq.ActiveMQConnectionFactory;
import javax.jms.*;
import java.io.IOException;
public class JmsConsumer {
public static
final String ACTIVEMQ_URL = "tcp://192.168.225.132:61616";
public static void main(String[] args) throws JMSException, IOException {
//1 按照给定的url地址，采用默认的用户名密码创建连接工厂（有三个参数的构造方法，其中两个参数是用户名密码）
ActiveMQConnectionFactory
activeMQConnectionFactory = new ActiveMQConnectionFactory(ACTIVEMQ_URL);
//2 通过连接工厂获得连接
Connection connection = activeMQConnectionFactory.createConnection();
//3 启动访问，真正建立连接，需要确保防火墙、服务端配置等都正确，能正常访问
connection.start();
//4 创建会话session，false代表不开启事物，Session.AUTO_ACKNOWLEDGE代表自动签收
Session session =
connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
//5 创建目的地（具体是队列queue还是主题topic）,参数是队列的名称
//返回类型可以用Queue接收，也可以用Destination接收，因为Queue继承Destination
Queue queue = session.createQueue("queue1");
MessageConsumer consumer =
session.createConsumer(queue);
//6 消费者消费消息
/* while(true){
TextMessage message =
(TextMessage)consumer.receive();
//TextMessage message =
(TextMessage)consumer.receive(5000L);
if(null == message){
break;
}
String s = message.getText();
System.out.println(s);
}*/
//6 以监听器的形式监听并消费消息
consumer.setMessageListener(new MessageListener()
{
@Override
public void onMessage(Message
message) {
if(message != null &&
message instanceof TextMessage){
TextMessage
textMessage = (TextMessage)message;
try {
System.out.println(textMessage.getText());
} catch (JMSException
e) {
e.printStackTrace();
}
}
}
});
//7 如果以监听器的形式监听消息，则需要此行代码，代表不关闭消费者，否则有可能上边还没有进行消费，下边就把资源都关闭了
//会抛出IOException
System.in.read();
//8 关闭资源
consumer.close();
session.close();
connection.close();
System.out.println("消费者从activemq消费消息完成！");
}
}
结果如下：
监听器的形式也是资源不关闭，生产者生产消息，消费者就能够监听到并进行消费。监听器是异步非阻塞的方式，当消息到达后，系统自动调用监听器MessageListener的onMessage(Message message)方法。
如果启动两个消费者监听消息队列，然后生产者向queue生产消息，两个消费者会轮询消费消息，就像activemq自带负载均衡的感觉。
activemq队列小总结：无论先启动消费者，还是先启动生产者，消费者都能消费到生产者生产的消息，两个没有时间上的关系（异步），且mq中的消息一旦被消费者消费掉，就不再储存该条消息了，所以消息不会被重复消费。
若有理解不到之处，望指正！
标签：java,queue,connection,session,消息,0032ActiveMQ,TextMessage,message 来源： https://www.cnblogs.com/xiao1572662/p/12080812.html