taikewu 2018-04-12
本文主要以一个简单的示例展示ActiveMQ收发消息:
一、ActiveMQ说明
二、代码示例
一、ActiveMQ说明
1、当前ActiveMQ的最新版本为: ActiveMQ 5.15.3 Release ====》点击:官网下载ActiveMQ地址
2、下载好之后解压,然后启动ActiveMQ,启动时要根据系统的位数来选择
二、代码示例
导入依赖jar包
<dependency>
<groupId>org.apache.activemq</groupId>
<artifactId>activemq-all</artifactId>
<version>
5.15
.
3
</version>
</dependency>
(1)生产者代码
package com.chinasoft.activemq; import javax.jms.Connection; import javax.jms.ConnectionFactory; import javax.jms.Destination; import javax.jms.JMSException; import javax.jms.MessageProducer; import javax.jms.Session; import javax.jms.TextMessage; import org.apache.activemq.ActiveMQConnectionFactory; /** * ActiveMQ消息的生产者 生产者发送消息到消息中间件,实现JMS规范接口的消息中间件称为JMS Provider * * @author Freedom * */ public class Sender { public static void main(String[] args) throws JMSException { sender(); } public static void sender() throws JMSException { // 1.建立ConnectionFactory ConnectionFactory f = new ActiveMQConnectionFactory(ActiveMQConnectionFactory.DEFAULT_USER, ActiveMQConnectionFactory.DEFAULT_PASSWORD, "tcp://localhost:61616"); // 2.通过ConnectionFactory工厂建立Connection的连接,并调用start开启连接 Connection c = f.createConnection(); c.start(); // 3.通过Connection对象创建session会话,用于接收消息,参数1:是否开启事务,参数2:设置签收方式 Session session = c.createSession(false, Session.AUTO_ACKNOWLEDGE);// 自动签收 // 4.通过session创建Destination对象,指的是一个客户端用来指定生产消息的目标和消费消息的来源对象 // 在PTP的模式中Destination被称为Queue队列;在pub/sub模式下Destination被称为Topic主题 Destination d = session.createQueue("firstMQ"); // 5.通过session创建消息的发送者和接收者(生产者和消费者) MessageProducer p = session.createProducer(null);// 在生产者发送消息的同时指定Destination // 6.MessageProducer设置持久化特性和非持久化特性 // 7.JMS规范的TextMessage形式数据通过Session来创建,并用MessageProducer发送消息,客户端receive方法进行数据的接受 // 使用完成一定要关闭Connection连接 TextMessage msg = null; for (int i = 0; i < 100; i++) { msg = session.createTextMessage(); msg.setText("ActiveMQ作为消息中间件:" + i); System.out.println("生产者发送的消息==" + msg.getText()); try { Thread.sleep(1000); } catch (InterruptedException e) { // TODO Auto-generated catch block e.printStackTrace(); } // 发送消息 p.send(d, msg); } // 关闭连接 c.close(); } }
注意:
①为了保证消息中间件中数据的可靠性,数据默认保存到kahadb中
②通过管控台可以看到生产者数据的变化
ActiveMQ内置jetty容器,访问管控台(http://localhost:8161/admin)
(2)消费者代码示例
package com.chinasoft.activemq; import javax.jms.Connection; import javax.jms.ConnectionFactory; import javax.jms.Destination; import javax.jms.JMSException; import javax.jms.MessageConsumer; import javax.jms.Session; import javax.jms.TextMessage; import org.apache.activemq.ActiveMQConnectionFactory; public class Consumer { public static void main(String[] args) throws JMSException { receiver(); } public static void receiver() throws JMSException { // 1.建立ConnectionFactory ConnectionFactory f = new ActiveMQConnectionFactory(ActiveMQConnectionFactory.DEFAULT_USER, ActiveMQConnectionFactory.DEFAULT_PASSWORD, "tcp://localhost:61616"); // 2.通过ConnectionFactory工厂建立Connection的连接,并调用start开启连接 Connection c = f.createConnection(); c.start(); // 3.通过Connection对象创建session会话,用于接收消息,参数1:是否开启事务,参数2:设置签收方式 Session session = c.createSession(false, Session.AUTO_ACKNOWLEDGE);// 自动签收 // 4.通过session创建Destination对象,指的是一个客户端用来指定生产消息的目标和消费消息的来源对象 // 在PTP的模式中Destination被称为Queue队列;在pub/sub模式下Destination被称为Topic主题 Destination d = session.createQueue("firstMQ"); // 5.通过session创建消息的发送者和接收者(生产者和消费者) MessageConsumer consumer = session.createConsumer(d); // 消费者用于接受MQ的数据 // 7.JMS规范的TextMessage形式数据通过Session来创建,并用MessageProducer发送消息,客户端receive方法进行数据的接受 TextMessage msg = null; while (true) { // 阻塞等待接受MQ的数据 msg = (TextMessage) consumer.receive(); System.out.println("消费者接受消息===" + msg.getText()); } } }
控制台显示结果:
管控台显示结果: