消息队列是软件系统领域用来实现系统间通信最广泛的中间件。基于消息队列的方式是指由应用中的某个系统负责发送消息，由关心这条消息的相关系统负责接收消息，并在收到消息后进行各自系统内的业务处理。消息可以非常简单，比如只包含文本字符串；也可以很复杂，比如包含字节流、字节数组，还可以包含嵌入对象，甚至是Java对象(经过序列化的对象)。
消息在被发送后可以立即返回，由消息队列来负责消息的传递，消息发布者只管将消息发布到消息队列而不用管谁来取，消息使用者只管从消息队列中取消息而不管是谁发布的，这样发布者和使用者都不用知道对方的存在(见下图)。
从上面的描述可以看出，消息队列（MQ）是一种系统间相互协作的通信机制。那么什么时候需要使用消息队列呢？
举个例子。某天产品人员说“系统要增加一个锅炉设备报警功能，当锅炉设备温度大于260度后，用户能收到邮件”。在实际场景中这种需求很常见，开发人员觉得这个很简单，就是提供一个判断逻辑，当锅炉设备温度大于260度进行判断，然后发送邮件，最好返回报警信息以警示。
该功能上线运行了一段时间后，产品人员说“设备高温后收到邮件的响应有点慢，很多人都提出这个意见，能不能优化一下”。开发人员首先想到的优化方案是将锅炉设备温度判断逻辑与发送邮件分开执行，怎么分呢？可以单独开启线程来做发送邮件的事情。
没多久，产品人员又说“现在设备高温并收到邮件的响应是快了，但有用户反映没收到报警邮件，能不能在发送邮件的时候先保存所发送邮件的内容，如果邮件发送失败了则进行补发”。
看着开发人员愁眉苦脸的样子，产品人员说“在邮件发送这块平台部门已经做好方案了，你直接用他们提供的服务就行”。开发人员一听，赶紧和平台部门沟通，对方的答复是“我们提供一个类似于邮局信箱的东西，你直接往这个信箱里写上发送邮件的地址、邮件标题和内容，之后就不用你操心了，我们会直接从信箱里取消息，向你所填写的邮件地址发送响应邮箱”。
这个故事讲的就是使用消息队列的典型场景---异步处理。消息队列还可用于解决解耦、流量削峰、日志收集等问题。
简单实现一个消息队列回到消息队列这个术语本身，它包含了两个关键词: 消息和队列。消息是指在应用间传送的数据，消息的表现形式是多样的，可以简单到只包含文本字符串，也可以复杂到有一个结构化的对象定义格式。对于队列，从抽象意义上来理解，就是指消息的进和出。从时间顺序上说，进和出并不一定是同步进行的，所以需要一个容器来暂存和处理消息。因此，一个典型意义上的消息队列，至少需要包含消息的发送、接受和暂存功能。
Broker: 消息处理中心，负责消息的接受、存储、转发等。
Producer: 消息生产者，负责产生和发送消息和消息处理中心。
Consumer: 消息消费者，负责从消息处理中心获取消息，并进行相应的处理。
可以看到，消息队列服务的核心是消息处理中心，它至少要具备消息发送、消息接受和消息暂存功能。所以，我们就从消息处理中心开始逐步搭建一个消息队列。
消息处理中心先看一下消息处理中心类(InMemoryStorage)的实现
import java.util.ArrayList; import java.util.Collections; import java.util.List; import java.util.concurrent.BlockingDeque; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.LinkedBlockingDeque; /** * @author james mu * @date 2020/7/27 21:47 */ public final class InMemoryStorage { //保存消息数据的容器，<topic,消息阻塞队列> 键值对 private final ConcurrentHashMap<String, BlockingQueque<QueueMsg>> storage; private static InMemoryStorage instance; private InMemoryStorage() { storage = new ConcurrentHashMap<>(); } //利用双重检查加锁(double-checked locking),首先检查是否示例已经创建了，如果尚未创建，"才"进行同步。这样以来，只有第一次会同步，这正是我们想要的。 public static InMemoryStorage getInstance() { if (instance == null) { synchronized (InMemoryStorage.class) { if (instance == null) { instance = new InMemoryStorage(); } } } return instance; } //保存消息到主题中，若topic对应的value为空，会将第二个参数的返回值存入并返回 public boolean put(String topic, QueueMsg msg) { return storage.computeIfAbsent(topic, (t) -> new LinkedBlockingDeque<>()).add(msg); } //获得主题中的消息 public <T extends QueueMsg> List<T> get(String topic) { //判断map中是否包含此topic if (storage.containsKey(topic)) { List<T> entities; //从此主题对应的阻塞队列中出队一个元素 T first = (T) storage.get(topic).poll(); if (first != null) { entities = new ArrayList<>(); entities.add(first); List<QueueMsg> otherList = new ArrayList<>(); //移动阻塞队列中最大999个元素到arrayList中 storage.get(topic).drainTo(otherList, 999); for (QueueMsg other : otherList) { entities.add((T) other); } } else { entities = Collections.emptyList(); } } return Collections.emptyList(); } //删除此map中所有的键值对 public void cleanup() { storage.clear(); } }作为一个消息处理中心中，至少要有一个数据容器用来保存接受到的消息。
