1、引入依赖
<dependency>
<groupId>org.apache.activemq</groupId>
<artifactId>activemq-all</artifactId>
<version>5.9.0</version>
</dependency>
2、queue模式
2.1、创建producer
package com.szl.jms.activemq.queue;
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;
/**
* @author szl
* @data 2018年8月5日 下午9:50:06
*
* 消息生产者
*/
public class MsgProducer {
private static final String url ="tcp://127.0.0.1:61616";
private static final String queueName ="activemq-queue";
public static void main(String[] args) throws JMSException {
//1、创建ConnectionFactory
ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(url);
//2、创建Connection
Connection connection = connectionFactory.createConnection();
//3、启动连接
connection.start();
//4、创建会话 false:不使用事务处理,Session.AUTO_ACKNOWLEDGE自动应答
Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
//5、创建目标
Destination destination = session.createQueue(queueName);
//6、创建生产者
MessageProducer producer = session.createProducer(destination);
for (int i = 0; i < 100; i++) {
//7、创建消息
TextMessage textMessage = session.createTextMessage("queueTest " + i);
//8、发布消息
producer.send(textMessage);
System.out.println("发送消息:" + textMessage.getText());
}
//9、关闭连接
connection.close();
}
}
2.2、创建consumer
package com.szl.jms.activemq.queue;
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;
/**
* @author szl
* @data 2018年8月5日 下午10:11:41
*
* 消息消费者
*/
public class MsgConsumer {
private static final String url ="tcp://127.0.0.1:61616";
private static final String queueName ="activemq-queue";
public static void main(String[] args) throws JMSException {
//1、创建ConnectionFactory
ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(url);
//2、创建Connection
Connection connection = connectionFactory.createConnection();
//3、启动连接
connection.start();
//4、创建会话 false:不使用事务处理,Session.AUTO_ACKNOWLEDGE自动应答
Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
//5、创建目标
Destination destination = session.createQueue(queueName);
//6、创建消费者
MessageConsumer consumer = session.createConsumer(destination);
//7、创建监听器
consumer.setMessageListener(new MessageListener() {
@Override
public void onMessage(Message message) {
TextMessage textMessage = (TextMessage) message;
try {
System.out.println("接收消息:" + textMessage.getText());
} catch (JMSException e) {
e.printStackTrace();
}
}
});
}
}
2.3、运行
可以先启动Consumer或者先启动Producer,只要发送消息成功,就能接受到消息。运行多个Consumer,每个Consumer平均接收Producer发送的消息。可以在http://localhost:8161/admin/queues.jsp查看信息。
3、Topic模式
3.1、创建producer
package com.szl.jms.activemq.topic;
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;
/**
* @author szl
* @data 2018年8月5日 下午10:27:36
*
*/
public class MsgProducer {
private static final String url ="tcp://127.0.0.1:61616";
private static final String topicName ="activemq-topic";
public static void main(String[] args) throws JMSException {
//1、创建ConnectionFactory
ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(url);
//2、创建Connection
Connection connection = connectionFactory.createConnection();
//3、启动连接
connection.start();
//4、创建会话 false:不使用事务处理,Session.AUTO_ACKNOWLEDGE自动应答
Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
//5、创建目标
Destination destination = session.createTopic(topicName);
//6、创建生产者
MessageProducer producer = session.createProducer(destination);
for (int i = 0; i < 100; i++) {
//7、创建消息
TextMessage textMessage = session.createTextMessage("test " + i);
//8、发布消息
producer.send(textMessage);
System.out.println("发送消息:" + textMessage.getText());
}
//9、关闭连接
connection.close();
}
}
1.2、创建Consumer
package com.szl.jms.activemq.topic;
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;
/**
* @author szl
* @data 2018年8月5日 下午10:27:24
*
*/
public class MsgConsumer {
private static final String url ="tcp://127.0.0.1:61616";
private static final String topicName ="activemq-topic";
public static void main(String[] args) throws JMSException {
//1、创建ConnectionFactory
ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(url);
//2、创建Connection
Connection connection = connectionFactory.createConnection();
//3、启动连接
connection.start();
//4、创建会话 false:不使用事务处理,Session.AUTO_ACKNOWLEDGE自动应答
Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
//5、创建目标
Destination destination = session.createTopic(topicName);
//6、创建消费者
MessageConsumer consumer = session.createConsumer(destination);
//7、创建监听器
consumer.setMessageListener(new MessageListener() {
@Override
public void onMessage(Message message) {
TextMessage textMessage = (TextMessage) message;
try {
System.out.println("接收消息:" + textMessage.getText());
} catch (JMSException e) {
e.printStackTrace();
}
}
});
}
}
2.3、运行
要先运行Consumer,再运行Producer,否则Consumer接收不到Producer发送的消息。启动多个Consumer,每个Consumer接收到全部的消息。可以在http://localhost:8161/admin/topic.jsp查看信息。