进阶高阶IoT架构-教你如何简单实现一个消息队列

sanshengshui

发布于 2021-02-25 16:39:27

5560

发布于 2021-02-25 16:39:27

文章被收录于专栏：穆书伟穆书伟

前言

消息队列是软件系统领域用来实现系统间通信最广泛的中间件。基于消息队列的方式是指由应用中的某个系统负责发送消息，由关心这条消息的相关系统负责接收消息，并在收到消息后进行各自系统内的业务处理。消息可以非常简单，比如只包含文本字符串；也可以很复杂，比如包含字节流、字节数组，还可以包含嵌入对象，甚至是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();
    }
}

作为一个消息处理中心中，至少要有一个数据容器用来保存接受到的消息。

Java中的队列(Queue)是提供该功能的一种简单的数据结构，同时为简化队列操作的并发访问处理，我们选择了它的一个子类LinkedBlockingDeque。该类提供了对数据的插入、获取、查询等操作，其底层将数据以链表的形式保存。如果用 offer方法插入数据时队列没满，则数据插入成功，并立即返回:如果队列满了，则直接返回 false。如果用 poll方法删除数据时队列不为空，则返回队列头部的数据;如果队列为空，则立刻返回 null。

消息格式定义

队列消息接口定义(QueueMsg)

/**
 * @author james mu
 * @date 2020/7/27 22:00
 */
public interface QueueMsg {
		//消息键
    String getKey();
  	//消息头
    QueueMsgHeaders getHeaders();
		//消息负载byte数组
    byte[] getData();
}

队列消息头接口定义(QueueMsgHeaders)

import java.util.Map;

/**
 * @author james mu
 * @date 2020/7/27 21:55
 */
public interface QueueMsgHeaders {
		//消息头放入
    byte[] put(String key, byte[] value);
		//消息头通过key获取byte数组
    byte[] get(String key);
		//消息头数据全部读取方法
    Map<String, byte[]> getData();
}

队列消息格式(ProtoQueueMsg)

/**
 * @author jamesmsw
 * @date 2021/2/19 2:23 下午
 */
public class ProtoQueueMsg implements QueueMsg {
    private final String key;
    private final String value;
    private final QueueMsgHeaders headers;

    public ProtoQueueMsg(String key, String value) {
        this(key, value, new DefaultQueueMsgHeaders());
    }

    public ProtoQueueMsg(String key, String value, QueueMsgHeaders headers) {
        this.key = key;
        this.value = value;
        this.headers = headers;
    }

    @Override
    public String getKey() {
        return key;
    }

    @Override
    public QueueMsgHeaders getHeaders() {
        return headers;
    }

    @Override
    public byte[] getData() {
        return value.getBytes();
    }
}

默认队列消息头(DefaultQueueMsgHeaders)

import java.util.HashMap;
import java.util.Map;

/**
 * @author james mu
 * @date 2020/7/27 21:57
 */
public class DefaultQueueMsgHeaders implements QueueMsgHeaders {

    protected final Map<String, byte[]> data = new HashMap<>();

    @Override
    public byte[] put(String key, byte[] value) {
        return data.put(key, value);
    }

    @Override
    public byte[] get(String key) {
        return data.get(key);
    }

    @Override
    public Map<String, byte[]> getData() {
        return data;
    }
}

消息生产者

import iot.technology.mqtt.storage.msg.QueueMsg;
import iot.technology.mqtt.storage.queue.QueueCallback;

/**
 * @author james mu
 * @date 2020/8/31 11:05
 */
public class Producer<T extends QueueMsg> {

    private final InMemoryStorage storage = InMemoryStorage.getInstance();

    private final String defaultTopic;

    public Producer(String defaultTopic) {
        this.defaultTopic = defaultTopic;
    }

    public void send(String topicName, T msg) {
        boolean result = storage.put(topicName, msg);
    }
}

消息消费者

import lombok.extern.slf4j.Slf4j;

import java.util.Collections;
import java.util.List;
import java.util.Set;
import java.util.stream.Collectors;

/**
 * @author james mu
 * @date 2020/8/31 11:23
 */
@Slf4j
public class Consumer<T extends QueueMsg> {
    private final InMemoryStorage storage = InMemoryStorage.getInstance();
    private volatile Set<String> topics;
    private volatile boolean stopped;
    private volatile boolean subscribed;
    private final String topic;

  	//虚构函数
    public Consumer(String topic) {
        this.topic = topic;
        stopped = false;
    }

    public String getTopic() {
        return topic;
    }

    public void subscribe() {
        topics = Collections.singleton(topic);
        subscribed = true;
    }

  	//批量订阅主题
    public void subscribe(Set<String> topics) {
        this.topics = topics;
        subscribed = true;
    }

    public void unsubscribe() {
        stopped = true;
    }

  	//不断读取topic集合下阻塞队列中的数据集合
    public List<T> poll(long durationInMillis) {
        if (subscribed) {
            List<T> messages = topics
                    .stream()
                    .map(storage::get)
                    .flatMap(List::stream)
                    .map(msg -> (T) msg).collect(Collectors.toList());

            if (messages.size() > 0) {
                return messages;
            }
            try {
                Thread.sleep(durationInMillis);
            } catch (InterruptedException e) {
                if (!stopped) {
                    log.error("Failed to sleep.", e);
                }
            }
        }

        return Collections.emptyList();
    }
}

至此，一个简单的消息队列中就实现完毕了。