消息队列是一种在不同组件或应用之间进行数据传递的技术，通常用于处理异步通信。它允许消息的发送者（生产者）和接收者（消费者）之间进行解耦。
概念
消息队列是一种先进先出（FIFO）的数据结构，它存储待处理的消息直到它们被消费。消息是生产者发送给队列的数据单元，消费者则从队列中读取这些消息进行处理。
原理
1. 生产者：
- 生产者是创建消息的实体，它负责将消息发送到队列。生产者不需要关心消息的具体处理过程，只需确保消息正确发送到队列。
2. 消息队列：
- 消息队列充当缓冲区，暂时存储从生产者那里发送过来的消息。队列管理消息的顺序，并确保按照发送的顺序逐一传递给消费者。
3. 消费者：
- 消费者从消息队列中读取消息，并进行相应的处理。消费者可以是同一应用的其他部分，或者是完全独立的应用。
4. 消息处理：
- 一旦消息被消费者读取，它可以被确认和删除，或者在处理失败时重新放回队列等待再次处理。
使用场景
异步处理：当应用执行耗时任务时，可以将任务封装成消息发送到队列，由消费者异步处理。
流量控制：在高流量事件如大促销或黑色星期五时，消息队列可以帮助缓冲入站流量，防止系统过载。
解耦服务：在微服务架构中，消息队列可以帮助减少服务之间的直接依赖，通过消息传递来通信，从而提高系统的可维护性和扩展性。
Java消息队列技术
在Java中，消息队列是一种数据结构或服务，用于在不同的应用组件或系统之间异步传递消息。它支持松耦合的架构，允许发送者和接收者独立地进行开发和扩展。消息队列可以帮助缓解高负载、增强系统的可伸缩性，并提供容错机制。下面是一些常见的Java消息队列技术：
1. Apache Kafka：
Kafka是一个分布式流处理平台，它不仅能够处理消息队列的功能，还能处理复杂的事件流。它特别适合需要高吞吐量和可靠性的大规模数据处理场景。
2. RabbitMQ：
RabbitMQ是一个开源消息代理，支持多种消息协议。它提供灵活的路由功能，能够保证消息的可靠传输。适合于复杂的消息传递需求和多种不同的通信模式。
3. ActiveMQ：
Apache ActiveMQ是一个强大的开源消息代理，支持多种JMS（Java Message Service）协议和客户端语言。适用于那些需要JMS标准支持的企业应用。
4. Amazon SQS (Simple Queue Service)：
SQS是一个托管的消息队列服务，提供简单的Web服务API来完全管理队列的消息传输。它能够无限扩展，并且不需要预先安装消息队列基础设施。
5. Google Cloud Pub/Sub：
Google的Pub/Sub提供了一种全球分布式的消息传递平台，适合处理大量数据的实时交换。
这个流程图展示了使用 ActiveMQ 实现消息队列的基本步骤，包括消息的发送和接收。以下是每个步骤的详细讲解。
1. 创建 ConnectionFactory
`ConnectionFactory` 是一个接口，用于创建连接到消息中间件（ActiveMQ）的工厂。它是创建连接的起点。
ConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://localhost:61616");
2. 使用 ConnectionFactory 创建 Connection
通过 `ConnectionFactory` 创建一个连接对象 `Connection`。
Connection connection = connectionFactory.createConnection();3. 启动 Connection
在使用连接之前，必须启动它。
connection.start();4. 使用 Connection 创建一个或多个 JMS Session
通过 `Connection` 创建会话 `Session`。会话是生产和消费消息的上下文。
Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
5. 使用 Session 创建 Queue 或 Topic
通过会话创建队列（Queue）或主题（Topic）。队列用于点对点消息传递，主题用于发布/订阅消息传递。
Queue queue = session.createQueue("testQueue");
// 或者
Topic topic = session.createTopic("testTopic");6. 使用 Session 创建 MessageProducer 或 MessageConsumer
根据需要创建消息生产者 `MessageProducer` 或消息消费者 `MessageConsumer`。
创建 MessageProducer
MessageProducer producer = session.createProducer(queue);创建 MessageConsumer
MessageConsumer consumer = session.createConsumer(queue);
7. 发送消息
使用 `MessageProducer` 发送消息。
TextMessage message = session.createTextMessage("Hello, ActiveMQ!");
producer.send(message);8. 接收消息
异步接收
设置消息监听器，当有消息到达时自动触发。
consumer.setMessageListener(new MessageListener() {
@Override
public void onMessage(Message message) {
if (message instanceof TextMessage) {
try {
String text = ((TextMessage) message).getText();
System.out.println("Received: " + text);
} catch (JMSException e) {
e.printStackTrace();
}
}
}
});同步接收
使用 `MessageConsumer.receive()` 方法同步接收消息。
Message message = consumer.receive();
if (message instanceof TextMessage) {
String text = ((TextMessage) message).getText();
System.out.println("Received: " + text);
}完整代码示例
生产者代码
import org.apache.activemq.ActiveMQConnectionFactory;
import javax.jms.*;
public class Producer {
public static void main(String[] args) throws JMSException {
// 创建连接工厂
ConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://localhost:61616");
// 创建连接
Connection connection = connectionFactory.createConnection();
connection.start();
// 创建会话
Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
// 创建队列
Queue queue = session.createQueue("testQueue");
// 创建生产者
MessageProducer producer = session.createProducer(queue);
// 创建消息
TextMessage message = session.createTextMessage("Hello, ActiveMQ!");
// 发送消息
producer.send(message);
// 关闭连接
connection.close();
}
}消费者代码
import org.apache.activemq.ActiveMQConnectionFactory;
import javax.jms.*;
public class Consumer {
public static void main(String[] args) throws JMSException {
// 创建连接工厂
ConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://localhost:61616");
// 创建连接
Connection connection = connectionFactory.createConnection();
connection.start();
// 创建会话
Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
// 创建队列
Queue queue = session.createQueue("testQueue");
// 创建消费者
MessageConsumer consumer = session.createConsumer(queue);
// 同步接收消息
Message message = consumer.receive();
if (message instanceof TextMessage) {
String text = ((TextMessage) message).getText();
System.out.println("Received: " + text);
}
// 异步接收消息
consumer.setMessageListener(new MessageListener() {
@Override
public void onMessage(Message message) {
if (message instanceof TextMessage) {
try {
String text = ((TextMessage) message).getText();
System.out.println("Received: " + text);
} catch (JMSException e) {
e.printStackTrace();
}
}
}
});
// 为了测试异步接收，保持程序运行一段时间
try {
Thread.sleep(10000);
} catch (InterruptedException e) {
e.printStackTrace();
}
// 关闭连接
connection.close();
}
}具体应用
调用百度云api
使用消息队列实现一个调用百度智能云 API 的校园卡程序有助于提高系统的可扩展性和可靠性。消息队列可以解耦生产者和消费者，并实现异步处理。
实现步骤
1. 配置消息队列：
- 安装并配置 RabbitMQ 或 ActiveMQ。
- 配置 Spring Boot 项目以连接到消息队列。
2. 创建生产者（Producer）：
- 接收用户上传的图片。
- 将图片编码为 Base64 格式，并发送到消息队列。
3. 创建消费者（Consumer）：
- 监听消息队列中的消息。
- 调用百度智能云 API 进行图片识别。
- 将识别结果存储以便后续查询。
4. 实现控制器（Controller）：
- 提供上传图片的接口。
- 提供获取识别结果的接口。
系统架构图
+-------------------+ +--------------------+ +-------------------+
| | | | | |
| User Uploads | | Message Queue | | API Consumer |
| (Controller) | -----> | (RabbitMQ/ActiveMQ)| -----> | (Baidu API Call) |
| | | | | |
+-------------------+ +--------------------+ +-------------------+具体实现
- 用户上传图片：用户通过前端页面上传图片，图片通过
RecognitionController接收并保存到消息队列。 - 消息队列：图片以消息的形式存储在消息队列中，保证消息的可靠传递。
- 消息处理：
RecognitionListener监听消息队列，当有新消息到达时，调用百度智能云 API 进行图片识别，并将结果保存到RecognitionService。 - 获取结果：用户可以通过访问
/api/resultPage来获取最新的识别结果。
代码部分：
1. 配置消息队列：
配置 `application.properties` 以连接到消息队列（以 ActiveMQ 为例）：
spring.activemq.broker-url=tcp://localhost:61616
spring.activemq.user=admin
spring.activemq.password=admin
spring.jms.pool.enabled=true2. 创建生产者（Producer）：
package com.example.mq;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.jms.core.JmsTemplate;
import org.springframework.stereotype.Service;
@Service
public class RecognitionService {
@Autowired
private JmsTemplate jmsTemplate;
public void sendImageForRecognition(byte[] imageBytes) {
jmsTemplate.convertAndSend("animal.recognition.queue", imageBytes);
}
// 其他方法...
}3. 创建消费者（Consumer）：
package com.example.mq;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.jms.annotation.JmsListener;
import org.springframework.stereotype.Component;
@Component
public class RecognitionListener {
@Autowired
private RecognitionService recognitionService;
@JmsListener(destination = "animal.recognition.queue")
public void processImage(byte[] imageBytes) {
String result = recognitionService.recognizeImage(imageBytes);
recognitionService.saveResult(result);
}
}4. 实现控制器（Controller）：
package com.example.mq;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Controller;
import org.springframework.ui.Model;
import org.springframework.web.bind.annotation.*;
import org.springframework.web.multipart.MultipartFile;
import org.springframework.web.servlet.view.RedirectView;
import java.io.IOException;
@Controller
@RequestMapping("/api")
public class RecognitionController {
@Autowired
private RecognitionService recognitionService;
@PostMapping("/recognize")
public RedirectView recognizeAnimal(@RequestParam("file") MultipartFile file, Model model) throws IOException {
if (file.isEmpty()) {
model.addAttribute("message", "File is empty");
return new RedirectView("/errorPage.html");
}
byte[] bytes = file.getBytes();
recognitionService.saveResult("等待识别结果...");
recognitionService.sendImageForRecognition(bytes);
return new RedirectView("/resultPage.html");
}
@GetMapping("/resultPage")
@ResponseBody
public String getResult() {
return recognitionService.getResult();
}
}
消息队列的意义
异步处理：允许用户在上传图片后立即获得响应，而不是等待图片识别结果。
解耦：上传图片的部分和图片识别的部分可以独立开发和扩展。
负载均衡：可以轻松增加更多的消费者以处理高并发请求。
可靠性：消息队列持久化消息，确保即使在系统故障时也不会丢失消息。
