java实现简单的消息队列

设计一个简单的消息队列

  • 简单的消息中间件
    • 代码部分
      • 消息中间件
      • 消息中间件服务器
      • 消息中间件客户端
      • 生产者客户端
      • 消费者客户端
  • 演示效果
    • 生产消息
    • 消息中间件
      • 生产者
    • 消费消息
      • 消息中间件
      • 消息者
    • 过度生产消息
      • 消息中间件
    • 过度消费消息
      • 消息中间件
      • 消费者
  • 总结

简单的消息中间件

代码部分

消息中间件

package com.test;

import java.util.concurrent.ArrayBlockingQueue;

public class Broker {
    //队列存储消息的最大数量
    private  final  static  int MAX_SIZE = 3;
    //保存消息数据的容器
    private static ArrayBlockingQueue<String> messageQueue = new ArrayBlockingQueue<>(MAX_SIZE);
    //生产消息
    public static  void  produce(String msg){
        System.out.println("[消息中间件]=============================================");
        if(messageQueue.offer(msg)){
            System.out.println("[消息中间件]收到消息:" + msg + ",暂存消息:" + messageQueue.size() + "条。");
        }else{
            System.out.println("[消息中间件]消息已满!");
        }
        System.out.println("[消息中间件]=============================================");
    }
    //消费消息
    public static  String  consume(){
        System.out.println("[消息中间件]=============================================");
        String msg = messageQueue.poll();
        if(msg != null){
            System.out.println("[消息中间件]消费消息:" + msg + ",暂存消息:" + messageQueue.size() + "条。");
        }else{
            System.out.println("[消息中间件]消息已空!");
        }
        System.out.println("[消息中间件]=============================================");
        return msg;
    }
}

消息中间件服务器

package com.test;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.io.PrintWriter;
import java.net.ServerSocket;
import java.net.Socket;

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 (socket.getInputStream().available() != 0){
                String str = in.readLine();
                if(str == null){
                    continue;
                }
                System.out.println("[服务器]收到消息:" + str);
                if("CONSUME".equals(str)){
                    String message = Broker.consume();
                    out.println(message);
                    out.flush();
                }else{
                   Broker.produce(str);
                }
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

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

消息中间件客户端

package com.test;

import java.io.BufferedReader;
import java.io.InputStreamReader;
import java.io.PrintWriter;
import java.net.InetAddress;
import java.net.Socket;


public class MqClient {

    public static void produce(String message) throws Exception {
        Socket socket = new Socket(InetAddress.getLocalHost(), BrokerServer.SERVICE_PORT);
        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);
        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;
    }
}

生产者客户端

package com.test;

public class ProduceClient {
    public static void main(String[] args) throws Exception {
        MqClient mqClient = new MqClient();
        mqClient.produce("Linux is very much!");
    }
}

消费者客户端

package com.test;

public class ConsumeClient {
    public static void main(String[] args) throws Exception {
        MqClient mqClient = new MqClient();
       String message = mqClient.consume();
        System.out.println("消费消息:" + message);
    }
}

演示效果

生产消息

消息中间件

java实现简单的消息队列_第1张图片

生产者

java实现简单的消息队列_第2张图片

消费消息

消息中间件

java实现简单的消息队列_第3张图片

消息者

java实现简单的消息队列_第4张图片

过度生产消息

消息中间件

java实现简单的消息队列_第5张图片

过度消费消息

消息中间件

java实现简单的消息队列_第6张图片

消费者

java实现简单的消息队列_第7张图片

总结

本文采取一个由数组结构组成的有界阻塞队列 ArrayBlockingQueue,套接字通讯,自定义简单的消息中间件。

你可能感兴趣的