当前位置：首页 > ai >正文

【从零开始：自制一个Java消息队列（MQ）】

ai 2025/7/6 23:16:43

🚀 从零开始：自制一个Java消息队列（MQ）

在现代分布式系统中，消息队列（Message Queue，MQ）已经成为一个至关重要的组件。它帮助系统在异步处理、负载均衡、解耦等方面提供了强大的支持。而今天，我们将从零开始，手把手教你如何用Java自制一个简单的消息队列。

📌 什么是消息队列？

消息队列是一种基于异步通信的中间件系统，它允许不同的系统或应用程序通过发送、接收消息的方式进行通信。消息队列通常使用以下两个基本概念：

Producer（生产者）：发送消息的应用程序或组件。
Consumer（消费者）：接收并处理消息的应用程序或组件。

消息队列通过消息的异步发送和接收，实现了系统之间的解耦，使得系统更具扩展性和弹性。

📊 消息队列的常见使用场景：

异步处理：解耦前端和后端，减少系统间的同步调用。
削峰填谷：在高峰时段，将高并发请求暂时缓存到消息队列中，逐步消费。
系统解耦：减少不同系统间的直接依赖，降低耦合性。

🔑 设计思路：自制消息队列

要自己实现一个消息队列系统，我们需要考虑以下几个基本功能：

消息的发送（生产者）。
消息的接收（消费者）。
消息的持久化：消息队列应该能够持久化数据，以防系统崩溃丢失数据。
消息的顺序性：消息的消费顺序应该与生产顺序一致。
并发与线程安全：考虑到高并发的场景，消息队列要能够高效且线程安全地工作。

下面是我们设计的简单消息队列的架构图：

在这里插入图片描述

🛠️ 实现步骤：如何用Java实现消息队列

1. 消息队列的基础类

首先，我们要设计一个Message类，表示队列中的单个消息。每个消息包含一个ID和内容。

public class Message {private String id;private String content;public Message(String id, String content) {this.id = id;this.content = content;}public String getId() {return id;}public String getContent() {return content;}
}

2. 消息队列类

接下来，我们设计一个MessageQueue类来存储消息。这个类应该具有以下功能：

enqueue：将消息添加到队列中。
dequeue：从队列中取出消息。
peek：查看队列中的下一个消息。

import java.util.LinkedList;
import java.util.Queue;public class MessageQueue {private Queue<Message> queue = new LinkedList<>();private final Object lock = new Object();// 添加消息到队列public void enqueue(Message message) {synchronized (lock) {queue.add(message);lock.notifyAll();  // 通知等待的消费者线程}}// 从队列中取出消息public Message dequeue() throws InterruptedException {synchronized (lock) {while (queue.isEmpty()) {lock.wait();  // 如果队列为空，消费者线程等待}return queue.poll();}}// 查看下一个消息public Message peek() {synchronized (lock) {return queue.peek();}}
}

3. 生产者（Producer）和消费者（Consumer）

接下来我们分别创建生产者和消费者类：

生产者（Producer）

生产者将消息放入队列中。

public class Producer implements Runnable {private MessageQueue messageQueue;public Producer(MessageQueue messageQueue) {this.messageQueue = messageQueue;}@Overridepublic void run() {for (int i = 1; i <= 10; i++) {Message message = new Message(String.valueOf(i), "Message " + i);messageQueue.enqueue(message);System.out.println("Produced: " + message.getContent());try {Thread.sleep(1000);  // 模拟消息产生的时间} catch (InterruptedException e) {e.printStackTrace();}}}
}

消费者（Consumer）

消费者从队列中获取消息并进行处理。

public class Consumer implements Runnable {private MessageQueue messageQueue;public Consumer(MessageQueue messageQueue) {this.messageQueue = messageQueue;}@Overridepublic void run() {try {while (true) {Message message = messageQueue.dequeue();System.out.println("Consumed: " + message.getContent());Thread.sleep(2000);  // 模拟消息处理时间}} catch (InterruptedException e) {e.printStackTrace();}}
}

4. 启动生产者和消费者

我们可以通过主函数来启动生产者和消费者的线程。

public class MQDemo {public static void main(String[] args) {MessageQueue messageQueue = new MessageQueue();Producer producer = new Producer(messageQueue);Consumer consumer = new Consumer(messageQueue);// 启动线程new Thread(producer).start();new Thread(consumer).start();}
}

📊 架构与工作原理

在上面的实现中，消息队列由LinkedList实现，使用synchronized来保证线程安全。生产者向队列中添加消息，而消费者从队列中取出消息。消费者在队列为空时会进入等待状态，直到生产者向队列中添加消息。

下图展示了消息队列的架构图：

+---------------------+
|    MessageQueue     |        <-- 存储消息的队列
+---------------------+^|+------------+       +------------+|  Producer  | ----> | Consumer   |     <-- 消费者从队列中消费消息+------------+       +------------+