Java語言快速實現簡單MQ消息隊列服務

目錄

• MQ基礎回顧
• 主要角色
• 自定義協議
• 流程順序
  • 項目構建流程
  • 具體使用流程
• 代碼演示
  • 消息處理中心 Broker
  • 消息處理中心服務 BrokerServer
  • 客戶端 MqClient
  • 測試MQ
• 小結

MQ基礎回顧

在上一篇消息通訊之關於消息隊列MQ必須了解的相關概念中 , 我們盡可能地詳細的了解了一些關於MQ (消息隊列) 的相關概念,並且我們上一篇中提到一個最基本的MQ通訊模型如下所示,所以本章節使用JAVA語言自己動手來寫一個MQ (類似ActiveMQ,RabbitMQ)

主要角色

首先我們必須需要搞明白MQ (消息隊列) 中的三個基本角色

• Producer : 消息生產者,負責生產消息並發送到Broker
• Broker : 消息處理中心,負責接受消息,存儲消息,轉發消息
• Consumer :消息消費者,負責消費消息

整體架構如下所示

自定義協議

首先從上一篇中介紹了協議的相關信息,具體廠商的MQ(消息隊列)需要遵循某種協議或者自定義協議 , 消息的生產者和消費者需要遵循其協議(約定)才能后成功地生產消息和生產消息,所以在這里我們自定義一個協議如下．

消息處理中心 : 如果接收到的信息包含"SEND"字符串,即視為生產者發送的消息,消息處理中心需要將此信息存儲等待消費者消費

消息處理中心 : 如果接受到的信息為CONSUME，既視為消費者發送消費請求，需要將存儲的消息隊列頭部的信息轉發給消費者，然后將此消息從隊列中移除

消息處理中心 : 如果消息處理中心存儲的消息滿3條仍然沒有消費者進行消費,則不再接受生產者的生產請求

消息生產者：需要遵循協議將生產的消息頭部增加　＂SEND:＂ 表示生產消息

消息消費者：需要遵循協議向消息處理中心發送　＂CONSUME＂字符串表示消費消息

流程順序

項目構建流程

下面將整個ＭＱ的構建流程過一遍

1. 新建一個Broker類，內部維護一個ArrayBlockingQueue隊列，提供生產消息和消費消息的方法，僅僅具備存儲服務功能
2. 新建一個BrokerServer類,將Broker發布為服務到本地9999端口，監聽本地9999端口的Socket鏈接，在接受的信息中進行我們的協議校驗, 這里僅僅具備接受消息,校驗協議,轉發消息功能;
3. 新建一個MqClient類,此類提供與本地端口9999的Socket鏈接 , 僅僅具備生產消息和消費消息的方法
4. 測試：新建兩個MyClient類對象，分別執行其生產方法和消費方法

具體使用流程

1. 生產消息：客戶端執行生產消息方法，傳入需要生產的信息，該信息需要遵循我們自定義的協議，消息處理中心服務在接受到消息會根據自定義的協議校驗該消息是否合法，如果合法如果合法就會將該消息存儲到Broker內部維護的ArrayBlockingQueue隊列中．如果ArrayBlockingQueue隊列沒有達到我們協議中的最大長度將將消息添加到隊列中，否則輸出生產消息失敗．
2. 消息消息：客戶端執行消費消息方法，Broker服務會校驗請求的信息的信息是否等於CONSUME，如果驗證成功則從Broker內部維護的ArrayBlockingQueue隊列的Poll出一個消息返回給客戶端

代碼演示

消息處理中心 Broker

/**
 * 消息處理中心
 */
public class Broker {
    // 隊列存儲消息的最大數量
    private final static int MAX_SIZE = 3;

    // 保存消息數據的容器
    private static ArrayBlockingQueue<String> messageQueue = new ArrayBlockingQueue<String>(MAX_SIZE);

    // 生產消息
    public static void produce(String msg) {
        if (messageQueue.offer(msg)) {
            System.out.println("成功向消息處理中心投遞消息：" + msg + "，當前暫存的消息數量是：" + messageQueue.size());
        } else {
            System.out.println("消息處理中心內暫存的消息達到最大負荷，不能繼續放入消息！");
        }
        System.out.println("=======================");
    }

    // 消費消息
    public static String consume() {
        String msg = messageQueue.poll();
        if (msg != null) {
            // 消費條件滿足情況，從消息容器中取出一條消息
            System.out.println("已經消費消息：" + msg + "，當前暫存的消息數量是：" + messageQueue.size());
        } else {
            System.out.println("消息處理中心內沒有消息可供消費！");
        }
        System.out.println("=======================");

        return msg;
    }

}

消息處理中心服務 BrokerServer

/**
 * 用於啟動消息處理中心
 */
public class BrokerServer implements Runnable {

    public static int SERVICE_PORT = 9999;

    private final Socket socket;

    public BrokerServer(Socket socket) {
        this.socket = socket;
    }

    @Override
    public void run() {
        try (
                BufferedReader in = new BufferedReader(new InputStreamReader(
                        socket.getInputStream()));
                PrintWriter out = new PrintWriter(socket.getOutputStream())
        )
        {
            while (true) {
                String str = in.readLine();
                if (str == null) {
                    continue;
                }
                System.out.println("接收到原始數據：" + str);

                if (str.equals("CONSUME")) { //CONSUME 表示要消費一條消息
                    //從消息隊列中消費一條消息
                    String message = Broker.consume();
                    out.println(message);
                    out.flush();
                } else if (str.contains("SEND:")){
                    //接受到的請求包含SEND:字符串 表示生產消息放到消息隊列中
                    Broker.produce(str);
                }else {
                    System.out.println("原始數據:"+str+"沒有遵循協議,不提供相關服務");
                }
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    public static void main(String[] args) throws Exception {
        ServerSocket server = new ServerSocket(SERVICE_PORT);
        while (true) {
            BrokerServer brokerServer = new BrokerServer(server.accept());
            new Thread(brokerServer).start();
        }
    }
}

客戶端 MqClient

/**
 * 訪問消息隊列的客戶端
 */
public class MqClient {

    //生產消息
    public static void produce(String message) throws Exception {
        //本地的的BrokerServer.SERVICE_PORT 創建SOCKET
        Socket socket = new Socket(InetAddress.getLocalHost(), BrokerServer.SERVICE_PORT);
        try (
                PrintWriter out = new PrintWriter(socket.getOutputStream())
        ) {
            out.println(message);
            out.flush();
        }
    }

    //消費消息
    public static String consume() throws Exception {
        Socket socket = new Socket(InetAddress.getLocalHost(), BrokerServer.SERVICE_PORT);
        try (
                BufferedReader in = new BufferedReader(new InputStreamReader(
                        socket.getInputStream()));
                PrintWriter out = new PrintWriter(socket.getOutputStream())
        ) {
            //先向消息隊列發送命令
            out.println("CONSUME");
            out.flush();

            //再從消息隊列獲取一條消息
            String message = in.readLine();

            return message;
        }
    }

}

測試MQ

public class ProduceClient {

    public static void main(String[] args) throws Exception {
        MqClient client = new MqClient();

        client.produce("SEND:Hello World");
    }

}

public class ConsumeClient {

    public static void main(String[] args) throws Exception {
        MqClient client = new MqClient();
        String message = client.consume();

        System.out.println("獲取的消息為：" + message);
    }
}

我們多執行幾次客戶端的生產方法和消費方法就可以看到一個完整的MQ的通訊過程,下面是我執行了幾次的一些日志

接收到原始數據：SEND:Hello World
成功向消息處理中心投遞消息：SEND:Hello World，當前暫存的消息數量是：1
=======================
接收到原始數據：SEND:Hello World
成功向消息處理中心投遞消息：SEND:Hello World，當前暫存的消息數量是：2
=======================
接收到原始數據：SEND:Hello World
成功向消息處理中心投遞消息：SEND:Hello World，當前暫存的消息數量是：3
=======================
接收到原始數據：SEND:Hello World
消息處理中心內暫存的消息達到最大負荷，不能繼續放入消息！
=======================
接收到原始數據：Hello World
原始數據:Hello World沒有遵循協議,不提供相關服務

接收到原始數據：CONSUME
已經消費消息：SEND:Hello World，當前暫存的消息數量是：2
=======================
接收到原始數據：CONSUME
已經消費消息：SEND:Hello World，當前暫存的消息數量是：1
=======================
接收到原始數據：CONSUME
已經消費消息：SEND:Hello World，當前暫存的消息數量是：0
=======================
接收到原始數據：CONSUME
消息處理中心內沒有消息可供消費！
=======================

小結

本章示例代碼主要源自分布式消息中間件實踐一書 , 這里我們自己使用Java語言寫了一個MQ消息隊列 , 通過這個消息隊列我們對MQ中的幾個角色 "生產者,消費者,消費處理中心,協議"有了更深的理解 ; 那么下一章節我們就來一塊學習具體廠商的MQ RabbitMQ