一、客户端代码实现

1.1、需求分析

RabbitMQ 的客户端设定：一个客户端可以有多个模块，每个模块都可以和 broker server 之间建立 “逻辑上的连接” （channel），这几个模块的channel 彼此之间是互相不影响的，同时这几个 channel 又复用的同一个 TCP 连接，省去了频繁建立/销毁 TCP 连接的开销（三次握手、四次挥手......）.

这里，我们也按照这样的逻辑实现消息队列的客户端，主要涉及到以下三个核心类：

ConnectionFactory：连接工厂，这个类持有服务器的地址，主要功能就是创建 Connection 对象.
Connection：表示一个 TCP连接，持有 Socket 对象，用来写入请求/读取响应，管理多个Channel 对象.
Channel：表示一个逻辑上的连接，需要提供一系列的方法，去和服务器提供的核心 API 对应（客户端提供的这些方法的内部，就是写入了一个特定的请求，然后等待服务器响应）.

1.2、具体实现

1）实现 ConnectionFactory

主要用来创建 Connection 对象.

public class ConnectionFactory {

    //broker server 的 ip 地址
    private String host;
    //broker server 的端口号
    private int port;

//    //访问 broker server 的哪个虚拟主机
//    //这里暂时先不涉及
//    private String virtualHostName;
//    private String username;
//    private String password;

    public Connection newConnection() throws IOException {
        Connection connection = new Connection(host, port);
        return connection;
    }

    public String getHost() {
        return host;
    }

    public void setHost(String host) {
        this.host = host;
    }

    public int getPort() {
        return port;
    }

    public void setPort(int port) {
        this.port = port;
    }
}

2）实现 Connection

属性如下

    private Socket socket;
    //一个 socket 连接需要管理多个 channel
    private ConcurrentHashMap<String, Channel> channelMap = new ConcurrentHashMap<>();
    private InputStream inputStream;
    private OutputStream outputStream;
    // DataXXX 主要用来 读取/写入 特定格式数据(例如 readInt())
    private DataInputStream dataInputStream;
    private DataOutputStream dataOutputStream;
    //用来处理 0xc 的回调，这里开销可能会很大，不希望把 Connection 阻塞住，因此使用 线程池 来处理
    private ExecutorService callbackPool;

构造如下

这里不光需要初始化属性，还需要创建一个扫描线程，由这个线程负责不停的从 socket 中读取响应数据，把这个响应数据再交给对应的 channel 负责处理

    public Connection(String host, int port) throws IOException {
        socket = new Socket(host, port);
        inputStream = socket.getInputStream();
        outputStream = socket.getOutputStream();
        dataInputStream = new DataInputStream(inputStream);
        dataOutputStream = new DataOutputStream(outputStream);

        callbackPool = Executors.newFixedThreadPool(4);

        //创建一个扫描线程，由这个线程负责不停的从 socket 中读取响应数据，把这个响应数据再交给对应的 channel 负责处理
        Thread t = new Thread(() -> {
            try {
                while(!socket.isClosed()) {
                    Response response = readResponse();
                    dispatchResponse(response);
                }
            } catch (SocketException e) {
                //连接正常断开的，此时这个异常可以忽略
                System.out.println("[Connection] 连接正常断开！");
            } catch(IOException | ClassNotFoundException | MqException e) {
                System.out.println("[Connection] 连接异常断开！");
                e.printStackTrace();
            }
        });
        t.start();
    }

释放 Connection 相关资源

    public void close() {
        try {
            callbackPool.shutdown();
            channelMap.clear();
            inputStream.close();
            outputStream.close();
            socket.close();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

使用这个方法来区别，当前的响应是一个针对控制请求的响应，还是服务器推送过来的消息.

如果是服务器推送过来的消息，就响应表明是 0xc，也就是一个回调，通过线程池来进行处理；

如果只是一个普通的响应，就把这个结果放到 channel 的哈希表中（随后 channel 会唤醒所有阻塞等待响应的线程，去 map 中拿数据）.

    public void dispatchResponse(Response response) throws IOException, ClassNotFoundException, MqException {
        if(response.getType() == 0xc) {
            //服务器推送过来的消息数据
            SubScribeReturns subScribeReturns = (SubScribeReturns) BinaryTool.fromBytes(response.getPayload());
            //根据 channelId 找到对应的 channel 对象
            Channel channel = channelMap.get(subScribeReturns.getChannelId());
            if(channel == null) {
                throw new MqException("[Connection] 该消息对应的 channel 再客户端中不存在！channelId=" + channel.getChannelId());
            }
            //执行该 channel 对象内部的回调(这里的开销未知，有可能很大，同时不希望把这里阻塞住，所以使用线程池来执行)
            callbackPool.submit(() -> {
                try {
                    channel.getConsumer().handlerDelivery(subScribeReturns.getConsumerTag(), subScribeReturns.getBasicProperties(),
                            subScribeReturns.getBody());
                } catch(MqException | IOException e) {
                    e.printStackTrace();
                }
            });
        } else {
            //当前响应是针对刚才的控制请求的响应
            BasicReturns basicReturns = (BasicReturns) BinaryTool.fromBytes(response.getPayload());
            //把这个结果放到 channel 的 哈希表中
            Channel channel = channelMap.get(basicReturns.getChannelId());
            if(channel == null) {
                throw new MqException("[Connection] 该消息对应的 channel 在客户端中不存在！channelId=" + channel.getChannelId());
            }
            channel.putReturns(basicReturns);
        }
    }

发送请求和读取响应

    /**
     * 发送请求
     * @param request
     * @throws IOException
     */
    public void writeRequest(Request request) throws IOException {
        dataOutputStream.writeInt(request.getType());
        dataOutputStream.writeInt(request.getLength());
        dataOutputStream.write(request.getPayload());
        dataOutputStream.flush();
        System.out.println("[Connection] 发送请求！type=" + request.getType() + ", length=" + request.getLength());
    }

    /**
     * 读取响应
     */
    public Response readResponse() throws IOException {
        Response response = new Response();
        response.setType(dataInputStream.readInt());
        response.setLength(dataInputStream.readInt());
        byte[] payload = new byte[response.getLength()];
        int n = dataInputStream.read(payload);
        if(n != response.getLength()) {
            throw new IOException("读取的响应格式不完整! n=" + n + ", responseLen=" + response.getLength());
        }
        response.setPayload(payload);
        System.out.println("[Connection] 收到响应！type=" + response.getType() + ", length=" + response.getLength());
        return response;
    }

在 Connection 中提供创建 Channel 的方法

    public Channel createChannel() throws IOException {
        String channelId = "C-" + UUID.randomUUID().toString();
        Channel channel = new Channel(channelId, this);
        //放到 Connection 管理的 channel 的 Map 集合中
        channelMap.put(channelId, channel);
        //同时也需要把 “创建channel” 这个消息告诉服务器
        boolean ok = channel.createChannel();
        if(!ok) {
            //如果创建失败，就说明这次创建 channel 操作不顺利
            //把刚才加入 hash 表的键值对再删了
            channelMap.remove(channelId);
            return null;
        }
        return channel;
    }

Ps：代码中使用了很多次 UUID ，这里我们和之前一样，使用加前缀的方式来进行区分.

3）实现 Channel

属性和构造如下

    private String channelId;
    // 当前这个 channel 是属于哪一个连接
    private Connection connection;
    //用来存储后续客户端收到的服务器响应，已经辨别是哪个响应(要对的上号) key 是 rid
    private ConcurrentHashMap<String, BasicReturns> basicReturnsMap = new ConcurrentHashMap<>();
    //如果当前 Channel 订阅了某个队列，就需要记录对应的回调是什么，当该队列消息返回回来的时候，调用回调
    //此处约定一个 Channel 只能有一个回调
    private Consumer consumer;

    public Channel(String channelId, Connection connection) {
        this.channelId = channelId;
        this.connection = connection;
    }

    public String getChannelId() {
        return channelId;
    }

    public void setChannelId(String channelId) {
        this.channelId = channelId;
    }

    public Connection getConnection() {
        return connection;
    }

    public void setConnection(Connection connection) {
        this.connection = connection;
    }

    public ConcurrentHashMap<String, BasicReturns> getBasicReturnsMap() {
        return basicReturnsMap;
    }

    public void setBasicReturnsMap(ConcurrentHashMap<String, BasicReturns> basicReturnsMap) {
        this.basicReturnsMap = basicReturnsMap;
    }

    public Consumer getConsumer() {
        return consumer;
    }

    public void setConsumer(Consumer consumer) {
        this.consumer = consumer;

实现 0x1 创建 channel

主要就是构造构造出 request，然后发送请求到 BrokerServer 服务器，阻塞等待服务器响应.

    /**
     * 0x1
     * 和服务器进行交互，告诉服务器，此处客户端已经创建了新的 channel 了
     * @return
     */
    public boolean createChannel() throws IOException {
        //构造 payload
        BasicArguments arguments = new BasicArguments();
        arguments.setChannelId(channelId);
        arguments.setRid(generateRid());
        byte[] payload = BinaryTool.toBytes(arguments);
        //发送请求
        Request request = new Request();
        request.setType(0x1);
        request.setLength(payload.length);
        request.setPayload(payload);
        connection.writeRequest(request);

        //等待服务器响应
        BasicReturns basicReturns = waitResult(arguments.getRid());
        return basicReturns.isOk();
    }

    /**
     * 生成 rid
     * @return
     */
    public String generateRid() {
        return "R-" + UUID.randomUUID().toString();
    }


    /**
     * 阻塞等待服务器响应
     * @param rid
     * @return
     */
    private BasicReturns waitResult(String rid) {
        BasicReturns basicReturns = null;
        while((basicReturns = basicReturnsMap.get(rid)) == null) {
            //查询结果为空，就说明咱们去菜鸟驿站要取的包裹还没到
            //此时就需要阻塞等待
            synchronized (this) {
                try {
                    wait();
                } catch (InterruptedException e) {
                    throw new RuntimeException(e);
                }
            }
        }
        basicReturnsMap.remove(rid);
        return basicReturns;
    }


    /**
     * 由 Connection 中的方法调用，区分为普通响应之后触发
     * 将响应放回到 channel 管理的 map 中，并唤醒所有线程
     * @param basicReturns
     */
    public void putReturns(BasicReturns basicReturns) {
        basicReturnsMap.put(basicReturns.getRid(), basicReturns);
        synchronized (this) {
            //当前也不知道有多少线程再等待上述的这个响应
            //因此就把所有等待的线程唤醒
            notifyAll();
        }
    }

Ps：其他的请求操作也和 0x1 的方式几乎一样，这里不一一展示了，主要说一下 0xa

0xa 消费者订阅队列消息，这里要先设置好回调到属性中，方便 Connection 通过这个属性来处理回调

值得注意的一点，我们约定 channelId 就是 consumerTag

    public boolean basicConsume(String queueName, boolean autoAck, Consumer consumer) throws IOException, MqException {
        //先设置回调
        if(this.consumer != null) {
            throw new MqException("该 channel 已经设置过消费消息回调了，不能重复！");
        }
        this.consumer = consumer;
        BasicConsumeArguments basicConsumeArguments = new BasicConsumeArguments();
        basicConsumeArguments.setRid(generateRid());
        basicConsumeArguments.setChannelId(channelId);
        basicConsumeArguments.setConsumerTag(channelId); // 注意：此处的 consumerTag 使用 channelId 来表示
        basicConsumeArguments.setQueueName(queueName);
        basicConsumeArguments.setAutoAck(autoAck);
        byte[] payload = BinaryTool.toBytes(basicConsumeArguments);

        Request request = new Request();
        request.setType(0xa);
        request.setLength(payload.length);
        request.setPayload(payload);

        connection.writeRequest(request);
        BasicReturns basicReturns = waitResult(basicConsumeArguments.getRid());
        return basicReturns.isOk();
    }

二、编写 Demo

2.1、实例

到了这里基本就实现完成了一个跨主机/服务器之间的生产者消费者模型了（功能上可以满足日常开发对消息队列的使用），但是还具有很强的扩展性，可以继续参考 RabbitMQ，如果有想法的，或者是遇到不会的问题，可以私信我~

以下我来我来编写一个 demo，模拟跨主机/服务器之间的生产者消费者模型（这里为了方便，就在本机演示）.

首先再 spring boot 项目的启动类中创建 BrokerServer ，绑定端口号，然后启动

@SpringBootApplication
public class RabbitmqProjectApplication {
    public static ConfigurableApplicationContext context;
    public static void main(String[] args) throws IOException {
        context = SpringApplication.run(RabbitmqProjectApplication.class, args);
        BrokerServer brokerServer = new BrokerServer(9090);
        brokerServer.start();
    }

}

编写消费者

public class DemoConsumer {

    public static void main(String[] args) throws IOException, MqException, InterruptedException {
        //建立连接
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("127.0.0.1");
        factory.setPort(9090);
        Connection connection = factory.newConnection();
        Channel channel = connection.createChannel();

        //创建交换机和队列(这里和生产者创建交换机和队列不冲突，谁先启动，就按照谁的创建，即使已经存在交换机和队列，再创建也不会有什么副作用)
        channel.exchangeDeclare("demoExchange", ExchangeType.DIRECT, true, false, null);
        channel.queueDeclare("demoQueue", true, false, false, null);

        //消费者消费消息
        channel.basicConsume("demoQueue", true, new Consumer() {
            @Override
            public void handlerDelivery(String consumerTag, BasicProperties basicProperties, byte[] body) throws MqException, IOException {
                System.out.println("开销消费");
                System.out.println("consumerTag=" + consumerTag);
                System.out.println("body=" + new String(body));
                System.out.println("消费完毕");
            }
        });

        //由于消费者不知道生产者要生产多少，就在这里通过循环模拟一直等待
        while(true) {
            Thread.sleep(500);
        }
    }

}

编写生产者

public class DemoProducer {

    public static void main(String[] args) throws IOException, InterruptedException {
        //建立连接
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("127.0.0.1");
        factory.setPort(9090);
        Connection connection = factory.newConnection();
        Channel channel = connection.createChannel();

        //创建交换机和队列(这里和消费者创建交换机和队列不冲突，谁先启动，就按照谁的创建，即使已经存在交换机和队列，再创建也不会有什么副作用)
        channel.exchangeDeclare("demoExchange", ExchangeType.DIRECT, true, false, null);
        channel.queueDeclare("demoQueue", true, false, false, null);

        //生产消息
        byte[] body1 = "Im cyk1 !".getBytes();
        channel.basicPublish("demoExchange", "demoQueue", null, body1);

        Thread.sleep(500);

        //关闭连接
        channel.close();
        connection.close();
    }

}