ActiveMQ是Apache所提供的一个开源的消息系统，完全采用Java来实现，因此，它能很好地支持J2EE提出的JMS（Java Message Service,即Java消息服务）规范。JMS是一组Java应用程序接口，它提供消息的创建、发送、读取等一系列服务。JMS提供了一组公共应用程序接口和响应的语法，类似于Java数据库的统一访问接口JDBC,它是一种与厂商无关的API，使得Java程序能够与不同厂商的消息组件很好地进行通信。
JMS支持两种消息发送和接收模型。
一种称为P2P(Ponit to Point)模型，即采用点对点的方式发送消息。P2P模型是基于队列的，消息生产者发送消息到队列，消息消费者从队列中接收消息，队列的存在使得消息的异步传输称为可能，P2P模型在点对点的情况下进行消息传递时采用。
另一种称为Pub/Sub(Publish/Subscribe，即发布-订阅)模型，发布-订阅模型定义了如何向一个内容节点发布和订阅消息，这个内容节点称为topic(主题)。主题可以认为是消息传递的中介，消息发布这将消息发布到某个主题，而消息订阅者则从主题订阅消息。主题使得消息的订阅者与消息的发布者互相保持独立，不需要进行接触即可保证消息的传递，发布-订阅模型在消息的一对多广播时采用。
apache-activemq-5.15.3\conf)
apache-activemq-5.15.3\conf)--默认为8161
\apache-activemq-5.15.3\bin\win64\目录下双击activemq.bat文件，在浏览器中输入http://localhost:8161/admin/, 用户名和密码输入admin即可
消息中间件有很多的用途和优点：
I. 导包--activemq-all-5.15.3.jar
II. Producer
/**
* 定义消息的生产者
* @author mazaiting
*/
public class Producer {
// 用户名
private static final String USERNAME = ActiveMQConnection.DEFAULT_USER;
// 密码
private static final String PASSWORD = ActiveMQConnection.DEFAULT_PASSWORD;
// 链接
private static final String BROKENURL = ActiveMQConnection.DEFAULT_BROKER_URL;
/**
* 定义消息并发送，等待消息的接收者(消费者)消费此消息
* @param args
* @throws JMSException
*/
public static void main(String[] args) throws JMSException {
// 消息中间件的链接工厂
ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(
USERNAME, PASSWORD, BROKENURL);
// 连接
Connection connection = null;
// 会话
Session session = null;
// 消息的目的地
Destination destination = null;
// 消息生产者
MessageProducer messageProducer = null;
try {
// 通过连接工厂获取链接
connection = connectionFactory.createConnection();
// 创建会话，进行消息的发送
// 参数一：是否启用事务
// 参数二：设置自动签收
session = connection.createSession(true, Session.AUTO_ACKNOWLEDGE);
// 创建消息队列
destination = session.createQueue("talkWithMo");
// 创建一个消息生产者
messageProducer = session.createProducer(destination);
// 设置持久化/非持久化， 如果非持久化，MQ重启后可能后导致消息丢失
messageProducer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);
// 模拟发送消息
for (int i = 0; i < 5; i++) {
TextMessage textMessage = session.createTextMessage("给妈妈发送的消息:"+i);
System.out.println("textMessage: " + textMessage);
messageProducer.send(textMessage);
}
// 如果设置了事务，会话就必须提交
session.commit();
} catch (JMSException e) {
e.printStackTrace();
} finally {
if (null != connection) {
connection.close();
}
}
}
}
III. Consumer
/**
* 定义消息的消费者
* @author mazaiting
*/
public class Consumer {
// 用户名
private static final String USERNAME = ActiveMQConnection.DEFAULT_USER;
// 密码
private static final String PASSWORD = ActiveMQConnection.DEFAULT_PASSWORD;
// 链接
private static final String BROKENURL = ActiveMQConnection.DEFAULT_BROKER_URL;
/**
* 接收消息
* @param args
* @throws JMSException
*/
public static void main(String[] args) throws JMSException {
// 消息中间件的链接工厂
ConnectionFactory connectionFactory = null;
// 链接
Connection connection = null;
// 会话
Session session = null;
// 消息的目的地
Destination destination = null;
// 消息的消费者
MessageConsumer messageConsumer = null;
// 实例化链接工厂，创建一个链接
connectionFactory = new ActiveMQConnectionFactory(USERNAME, PASSWORD, BROKENURL);
try {
// 通过工厂获取链接
connection = connectionFactory.createConnection();
// 启动链接
connection.start();
// 创建会话，进行消息的接收
session = connection.createSession(true, Session.AUTO_ACKNOWLEDGE);
// 创建消息队列
destination = session.createQueue("talkWithMo");
// 创建一个消息的消费者
messageConsumer = session.createConsumer(destination);
// 模拟接收消息
while (true) {
TextMessage textMessage = (TextMessage) messageConsumer.receive(10000);
if (null != textMessage) {
System.out.println("收到消息： " + textMessage);
} else {
break;
}
}
// 提交
session.commit();
} catch (JMSException e) {
e.printStackTrace();
} finally {
if (null != connection) {
connection.close();
}
}
}
}
IV. 测试
先运行生产者Producer
ActiveMQ控制台
再运行消费者Consumer
ActiveMQ控制台
V. 消息类型
I. 实现思路
Client的Producer发出一个JMS message形式的request，request上附加了一些额外的属性：
Worker的consumer收到requset，处理request并用producer发出reply，destination就从requset的JMSReplyTo属性中得到。
II. Server代码
public class Server implements MessageListener {
// 经纪人链接
private static final String BROKER_URL = "tcp://localhost:61616";
// 请求队列
private static final String REQUEST_QUEUE = "requestQueue";
// 经纪人服务
private BrokerService brokerService;
// 会话
private Session session;
// 生产者
private MessageProducer producer;
// 消费者
private MessageConsumer consumer;
private void start() throws Exception {
createBroker();
setUpConsumer();
}
/**
* 创建经纪人
* @throws Exception
*/
private void createBroker() throws Exception {
// 创建经纪人服务
brokerService = new BrokerService();
// 设置是否持久化
brokerService.setPersistent(false);
// 设置是否使用JMX
brokerService.setUseJmx(false);
// 添加链接
brokerService.addConnector(BROKER_URL);
// 启动
brokerService.start();
}
/**
* 设置消费者
* @throws JMSException
*/
private void setUpConsumer() throws JMSException {
// 创建连接工厂
ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory(BROKER_URL);
// 创建连接
Connection connection = connectionFactory.createConnection();
// 启动连接
connection.start();
// 创建Session
session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
// 创建队列
Destination adminQueue = session.createQueue(REQUEST_QUEUE);
// 创建生产者
producer = session.createProducer(null);
// 设置持久化模式
producer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);
// 创建消费者
consumer = session.createConsumer(adminQueue);
// 消费者设置消息监听
consumer.setMessageListener(this);
}
public void stop() throws Exception {
producer.close();
consumer.close();
session.close();
brokerService.stop();
}
@Override
public void onMessage(Message message) {
try {
// 创建新消息
TextMessage response = this.session.createTextMessage();
// 判断消息是否是文本消息
if (message instanceof TextMessage) {
// 强转为文本消息
TextMessage textMessage = (TextMessage) message;
// 获取消息内容
String text = textMessage.getText();
// 设置消息
response.setText(handleRequest(text));
}
response.setJMSCorrelationID(message.getJMSCorrelationID());
producer.send(message.getJMSReplyTo(), response);
} catch (JMSException e) {
e.printStackTrace();
}
}
/**
* 构建消息内容
* @param text 文本
* @return
*/
private String handleRequest(String text) {
return "Response to ‘" + text + "‘";
}
public static void main(String[] args) throws Exception {
Server server = new Server();
// 启动
server.start();
System.out.println();
System.out.println("Press any key to stop the server");
System.out.println();
System.in.read();
server.stop();
}
}
III. Client代码
public class Client implements MessageListener {
// 经纪人链接
private static final String BROKER_URL = "tcp://localhost:61616";
// 请求队列
private static final String REQUEST_QUEUE = "requestQueue";
// 连接
private Connection connection;
// 会话
private Session session;
// 生产者
private MessageProducer producer;
// 消费者
private MessageConsumer consumer;
// 请求队列
private Queue tempDest;
public void start() throws JMSException {
// 连接工厂
ActiveMQConnectionFactory activeMQConnectionFactory = new ActiveMQConnectionFactory(BROKER_URL);
// 创建连接
connection = activeMQConnectionFactory.createConnection();
// 开启连接
connection.start();
// 创建会话
session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
// 创建队列
Destination adminQueue = session.createQueue(REQUEST_QUEUE);
// 创建生产者
producer = session.createProducer(adminQueue);
// 设置持久化模式
producer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);
// 创建模板队列
tempDest = session.createTemporaryQueue();
// 创建消费者
consumer = session.createConsumer(tempDest);
// 设置消息监听
consumer.setMessageListener(this);
}
/**
* 停止
* @throws JMSException
*/
public void stop() throws JMSException {
producer.close();
consumer.close();
session.close();
}
/**
* 请求
* @param request
* @throws JMSException
*/
public void request(String request) throws JMSException {
System.out.println("Request: " + request);
// 创建文本消息
TextMessage textMessage = session.createTextMessage();
// 设置文本内容
textMessage.setText(request);
// 设置回复
textMessage.setJMSReplyTo(tempDest);
// 获取UUID
String correlationId = UUID.randomUUID().toString();
// 设置JMS id
textMessage.setJMSCorrelationID(correlationId);
// 发送消息
this.producer.send(textMessage);
}
@Override
public void onMessage(Message message) {
try {
System.out.println("Received response for: " + ((TextMessage)message).getText());
} catch (JMSException e) {
e.printStackTrace();
}
}
public static void main(String[] args) throws JMSException, InterruptedException {
Client client = new Client();
// 启动
client.start();
int i = 0;
while(i++ < 10) {
client.request("REQUEST- " + i);
}
Thread.sleep(3000);
client.stop();
}
}
IV. 测试
启动Server
启动Client
如果您觉得我的
分布式--ActiveMQ 消息中间件(一) https://www.jianshu.com/p/8b9bfe865e38
原文：https://www.cnblogs.com/Jeely/p/11168287.html