- 大纲
- ActiveMQ是什么
- Apache推出的，开源的，完全支持 JMS1.1 和 J2EE1.4 规范的JMS Provider实现的消息中间件。
- 能干什么
- 主要功能：实现 JMS Provider，用来帮助实现高可用、高性能、可伸缩、易用和安全的企业级面向消息服务的系统。
- 直白来说：一个消息 中间件
- 异步调用
- 特点
- 很容易集成Tomcat Spring
- 支持多种语言
- 支持多种协议
- 可插拔的体系结构，可以灵活定制、如 消息存储方式、安全管理等
- 设计上保证了高性能
- 消息中间件
- MOM基本功能：将信息以消息的形式，从一个应用程序传送到另一个或多个应用程序。
- 主要特点
- 消息异步接受，类似手机短信的欣慰，消息发送者不需要等待消息接收者的响应。
- 消息可靠接收，确保消息在中间件可靠保存，只有接收方收到消息后才删除消息。
- 应用场景
- 在多个系统间进行整合和通讯的时候，通常会要求：
- 可靠传输：数据不能丢失，有的时候，也会要求不能重复传输。
- 异步传输：否则各个系统同步发送接受数据，互相等待，造成系统瓶颈。
- 基本使用
- 控制台展示
- MQ存数取数
- 放入至MQ中
- Map<String, Object> stringObjectMap = activeMQUtil.mQUtil(bms);
- 返回: success
- 在MQ中取出
- 收到消息：message--Bms{chargingStationId='CAGDBF016108', chargingPileId='0107011106032414', timeStamp='2021-03-19 16:30:04', SOC=60, leftTime='2021-03-16 10:10:10', Voltage=234.0, Current=112.0, voltageMin=2.0, voltageMax=4.0, degreeMin=25.0, degreeMax=38.0, carNo='蒙A50219', equipmentStatusTime='2021-03-16 10:10:10'}
- 项目中添加使用的工具类
package com.example.demo.util;
import com.example.demo.pojo.Bms;
import com.example.demo.pojo.Staff;
import org.apache.activemq.ActiveMQConnectionFactory;
import org.springframework.stereotype.Repository;
import javax.jms.*;
import java.util.HashMap;
import java.util.Map;
/**
* @ClassName activeMQUtil
* @Description ActiveMQ
* @Author CodeSheep
* @Date 2020/10/27 15:46
* @Version V1.0
* @Package com.example.demo.util
* 消费者连接 {@link consumer}
* 生产者连接 {@link producer}
*/
@Repository
public class ActiveMQUtil {
public Map<String, Object> mQUtil(Bms bms) throws Exception {
Map<String, Object> map = new HashMap<>();
ConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://127.0.0.1:61616");
Connection connection = connectionFactory.createConnection();
connection.start();
Session session = connection.createSession(Boolean.TRUE, Session.AUTO_ACKNOWLEDGE);
Destination destination = session.createQueue("my-queue");
MessageProducer producer = session.createProducer(destination);
for (int i = 0; i < 1; i++) {
TextMessage message = session.createTextMessage("message--" + bms);
Thread.sleep(1000);
//通过消息生产者发出消息
producer.send(message);
}
session.commit();
session.close();
connection.close();
map.put("ActiveMQ", "success");
return map;
}
}
- 生产者测试
package com.example.demo.util;
import org.apache.activemq.ActiveMQConnectionFactory;
import javax.jms.*;
/**
* @ClassName producer
* @Description ActiveMQ
* @Author CodeSheep
* @Date 2020/10/27 15:16
* @Version V1.0
* @Package com.example.demo.util
*/
public class producer {
public static void main(String[] args) throws Exception {
ConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://127.0.0.1:61616");
Connection connection = connectionFactory.createConnection();
connection.start();
Session session = connection.createSession(Boolean.TRUE, Session.AUTO_ACKNOWLEDGE);
Destination destination = session.createQueue("my-queue");
MessageProducer producer = session.createProducer(destination);
for (int i = 0; i < 1; i++) {
TextMessage message = session.createTextMessage("于老板--" + i);
Thread.sleep(1000);
//通过消息生产者发出消息
producer.send(message);
}
session.commit();
session.close();
connection.close();
}
}
- 消费者测试
package com.example.demo.util;
import org.apache.activemq.ActiveMQConnectionFactory;
import javax.jms.*;
/**
* @ClassName consumer
* @Description consumer
* @Author CodeSheep
* @Date 2020/10/27 15:37
* @Version V1.0
* @Package com.example.demo.util
*/
public class consumer {
public static void main(String[] args) throws JMSException {
ConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://127.0.0.1:61616");
Connection connection = connectionFactory.createConnection();
connection.start();
final Session session = connection.createSession(Boolean.TRUE, Session.AUTO_ACKNOWLEDGE);
Destination destination = session.createQueue("my-queue");
MessageConsumer consumer = session.createConsumer(destination);
Integer i = 0;
while (i < 1) {
i++;
TextMessage message = (TextMessage) consumer.receive();
session.commit();
System.out.println("收到消息：" + message.getText());
}
session.close();
connection.close();
}
}
