目录

前言

1. 创建ConnectionFactory

2. Connection 和 Channel 的定义

2.1 Connection

2.2 Channel

3. 封装请求响应读写操作

3.1 写入请求

3.2 读取响应

3.3 Connection中创建Channel

4. 封装发送请求的操作

4.1 创建Channel

4.2 阻塞等待服务器响应

4.3 添加响应信息到 basicReturnsMap

4.4 构造其他功能的请求 

5. 处理响应

5.1 在Connection中创建扫描线程

5.2 分发响应

6. 关闭Connection

---

前言

        上一章节,我们将BrokerServer服务器进行了实现,对请求的处理以及返回响应的格式进行了详细的阐述,本章节将实现消息队列的客户端,用来给生产者和消费者提供服务.主要实现提供用来创建连接的工厂类,以及Channel类的实现(一个客户端可以进行创建多个连接,一个连接对用一个Socket,一个TCP连接,一个连接中又包括多个Channel.)最后封装请响应读写操作.本项目全部代码已上传Gitee,链接放在文章末尾,欢迎大家访问!

---

1. 创建ConnectionFactory

用来进行实现创建连接的一个工厂类

1. 主机IP地址

2. 端口号

3. newConnection() 

此处可以实现链接指定的虚拟主机,并且链接的时候进行用户名和密码的验证(此处没有实现,后续进行扩展)

package com.example.demo.mqclient;

import lombok.Getter;
import lombok.Setter;

import java.io.IOException;

/**
 * Created with IntelliJ IDEA.
 * Description:连接的工厂
 * User: YAO
 * Date: 2023-08-03
 * Time: 9:10
 */
@Getter
@Setter
public class ConnectionFactory {
    // 1. 服务器的端口号
    private int port;

    // 2. 服务器的端口号
    private String host;

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

    // 可以进行扩展,实现多个虚拟主机,用户和密码的验证
    // private String virtualHostName;
    // private String username;
    // private String password;

}

2. Connection 和 Channel 的定义

        我们为了实现连接的复用性,为了区分出请求分类,我们引出了Channel的概念,一个连接中有多个Channel,但是站在服务器的角度来说,还是在一个连接中接收多个请求,但是在服务器端,根据ChannelID对连接进行了存储.这样就可以确定哪些客户端现在正在连接服务器.当连接断开的时候,我们对这个存储的数据结构,按照clientSocket进行清除没有价值的连接.保证服务器能正确的了解当时哪个客户端正在连接.

2.1 Connection

1. Socket 对象,用于连接服务器

2. InputStream  OutputStream  DataInputStream  DataOutputStream 均为socket通信的接口.

3. channelMap 是管理当前连接中的Channel

4. callbackPool 是用来在客户端进行处理用户回调的线程池,主要就是进行消费消息.

@Getter
@Setter
public class Connection {
    // 1. Socket 对象
    private Socket socket = null;

    // 2. 需要将这个连接中的Channel进行使用哈希表组织
    // 一个连接中对应了多个Channel,对次连接的多个Channel进行管理
    // key: ChannelID value: channel对象
    private ConcurrentHashMap<String, Channel> channelMap = new ConcurrentHashMap<>();

    // 3. 字节流对象
    private InputStream inputStream;
    private OutputStream outputStream;
    private DataInputStream dataInputStream;
    private DataOutputStream dataOutputStream;

    // 4. 线程池 执行回调,当客户端收到服务器推送的消息的时候,在这个回调函数中进行消费消息.
    private ExecutorService callbackPool = null;
}

2.2 Channel

1. ChannelID: Channel的唯一标识,使用UUID进行创建

2. Connection: 当前Channel在哪个Connection中

3. basicReturnsMap: 存储客户端收到服务器的响应

4. Consumer: 回调函数,当前Channel中进行了订阅消息,记录当前的回调函数是什么,当消息返回到客户端的时候调用回调函数进行消费消息.

@Getter
@Setter
public class Channel {
    // channel的标识
    private String channelId;

    // 当前这个Channel属于哪个连接
    private Connection connection;

    // 存储后续客户端收到服务器的响应, 方便后续客户端发送请求之后找到对应的响应
    // key:Rid value:响应信息
    private ConcurrentHashMap<String, BasicReturns> basicReturnsMap = new ConcurrentHashMap<>();

    // 回调方法  如果当前的Channel订阅了某个队列,就要记录当前的回调函数是什么,当该队列的消息返回到客户端的时候,进行调用回调
    private Consumer consumer = null;
}

3. 封装请求响应读写操作

3.1 写入请求

按照我们之前约定好的应用层协议的格式进行发送请求.最后记得刷新缓冲区.

/**
     * 1. 发送请求
     */
    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());
    }

3.2 读取响应

/**
     * 2. 读取响应
     */
    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("读取的响应数据不完整!");
        }
        response.setPayload(payload);
        System.out.println("[Connection] 收到响应! type=" + response.getType() + ", length=" + response.getLength());
        return response;
    }

3.3 Connection中创建Channel

在Connection中进行完成创建本连接的Channel

1. 先生成一个ChannelID,使用UUID进行生成

2. 创建Channel对象,传入ChannelID以及当前连接的对象

3. 将这个Channel添加到Connection的Map中,进行存储

4. 发送请求告诉服务器进行创建Channel,使得服务器那边进行存储当前连接的客户端信息.(在Channel类中创建发送请求的方法)

5. 如果服务器创建失败,记得将当前Connection的Map中刚创建的Channel对象按照ChannelID进行删除

/**
     * 创建连接中的Channel
     * @return
     * @throws IOException
     */
    public Channel createChannel() throws IOException {
        // 1. 先定义Channel的ID
        String channelId = "C-" + UUID.randomUUID().toString();
        // 2. 创建Channel对象,传入ChannelID以及所属的连接对象(当前连接)
        Channel channel = new Channel(channelId, this);
        // 3. 把这个 channel 对象放到 Connection 管理 channel 的 哈希表 中.
        channelMap.put(channelId, channel);
        // 4. 同时也需要把 "创建 channel" 的这个消息也告诉服务器.
        boolean ok = channel.createChannel();
        if (!ok) {
            // 服务器这里创建失败了!! 整个这次创建 channel 操作不顺利!!
            // 把刚才已经加入 hash 表的键值对, 再删了.
            channelMap.remove(channelId);
            return null;
        }
        return channel;
    }

4. 封装发送请求的操作

在Channel提供发送请求的操作

主要的步骤如下:

1. 按照要代用服务器功能的参数进行传参

2. 组织传入的参数到对应参数类的成员属性中

3. 将参数类进行序列化,添加到payload中,进一步按照请求服务器功能设置请求类型(这都是我们提前进行约定好的)

4. 阻塞等待服务器的响应

5. 得到响应

4.1 创建Channel

    /**
     * 1. 创建Channel 用于和服务器进行交互
     */
    public boolean createChannel() throws IOException {
        // 对于创建 Channel 操作来说, payload 就是一个 basicArguments 对象
        BasicArguments basicArguments = new BasicArguments();
        basicArguments.setChannelId(channelId);
        basicArguments.setRid(generateRid());
        byte[] payload = BinaryTool.toBytes(basicArguments);

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

        // 构造出完整请求之后, 就可以发送这个请求了.
        connection.writeRequest(request);
        // 等待服务器的响应
        BasicReturns basicReturns = waitResult(basicArguments.getRid());
        return basicReturns.isOk();
    }

4.2 阻塞等待服务器响应

        我们的服务器的响应式是异步的,我们在发送请求之后要进行阻塞等待服务器的响应.

        我们之前在Channel中设置了一个basicReturnsMap用来记录请求对应的响应,我们设置一个循环不断的从这个Map中使用连接的ID进行获取,响应的信息,针对当前的Channel对象进行加锁,直到获取到响应信息(在后面,我们会在Connection中设置一个扫描线程,用来获取响应,收到响应就会响应插入到Channel的哈希表上,然后进行唤醒线程进行获取响应).读取成功之后在哈希表进行删除这个响应信息.

/**
     * 阻塞等待服务器响应(根据当前请求的ID进行获取)
     * @param rid
     * @return
     */
    private BasicReturns waitResult(String rid) {
        BasicReturns basicReturns = null;
        while ((basicReturns = basicReturnsMap.get(rid)) == null) {
            // 如果查询结果为 null, 说明包裹还没回来.没得到响应
            // 此时就需要阻塞等待.
            synchronized (this) {
                try {
                    wait();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }
        // 读取成功之后, 还需要把这个消息从哈希表中删除掉.
        basicReturnsMap.remove(rid);
        return basicReturns;
    }

4.3 添加响应信息到 basicReturnsMap

        此方法,用来添加响应信息到basicReturnsMap,进而唤醒线程进行获取响应,此处是获取所有线程,拿到锁的获取,没拿到的进行阻塞等待.

/**
     * 往哈希表添加请求ID对应的响应信息,并且通知需要获取对应请求ID的请求,进行获取响应
     * @param basicReturns
     */
    public void putReturns(BasicReturns basicReturns) {
        basicReturnsMap.put(basicReturns.getRid(), basicReturns);
        synchronized (this) {
            // 当前也不知道有多少个线程在等待上述的这个响应.
            // 把所有的等待的线程都唤醒.
            notifyAll();
        }
    }

4.4 构造其他功能的请求 

下面给出下图中与之对相应的构造请求的代码

    /**
     * 2. 关闭channel
     * @return
     * @throws IOException
     */
    public boolean close() throws IOException {
        BasicArguments basicArguments = new BasicArguments();
        basicArguments.setRid(generateRid());
        basicArguments.setChannelId(channelId);

        byte[] payload = BinaryTool.toBytes(basicArguments);

        Request request = new Request();
        request.setType(0x2);
        request.setLength(payload.length);
        request.setPayload(payload);
        // 2. 发送请求
        connection.writeRequest(request);

        // 3. 等待服务器的处理
        //    阻塞等待
        BasicReturns basicReturns = waitResult(basicArguments.getRid());
        return basicReturns.isOk();
    }

    /**
     * 3. 实现创建交换机
     * @param exchangeName
     * @param exchangeType
     * @param durable
     * @param autoDelete
     * @param arguments
     * @return
     * @throws IOException
     */
    public boolean exchangeDeclare(String exchangeName, ExchangeType exchangeType, boolean durable, boolean autoDelete,
                                   Map<String, Object> arguments) throws IOException {
        ExchangeDeclareArguments exchangeDeclareArguments = new ExchangeDeclareArguments();
        exchangeDeclareArguments.setRid(generateRid());
        exchangeDeclareArguments.setChannelId(channelId);
        exchangeDeclareArguments.setExchangeName(exchangeName);
        exchangeDeclareArguments.setExchangeType(exchangeType);
        exchangeDeclareArguments.setAutoDelete(autoDelete);
        exchangeDeclareArguments.setDurable(durable);
        exchangeDeclareArguments.setArguments(arguments);

        byte[] payload = BinaryTool.toBytes(exchangeDeclareArguments);
        Request request = new Request();
        request.setType(0x3);
        request.setLength(payload.length);
        request.setPayload(payload);
        // 2. 发送请求
        connection.writeRequest(request);

        // 3. 等待服务器的处理
        //    阻塞等待
        BasicReturns basicReturns = waitResult(exchangeDeclareArguments.getRid());
        return basicReturns.isOk();
    }

    /**
     * 4. 删除交换机
     * @param exchangeName
     * @return
     * @throws IOException
     */
    public boolean exchangeDelete(String exchangeName) throws IOException {
        ExchangeDeleteArguments arguments = new ExchangeDeleteArguments();
        arguments.setRid(generateRid());
        arguments.setChannelId(channelId);
        arguments.setExchangeName(exchangeName);
        byte[] payload = BinaryTool.toBytes(arguments);

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

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

    /**
     * 5. 创建队列
     * @param queueName
     * @param durable
     * @param exclusive
     * @param autoDelete
     * @param arguments
     * @return
     * @throws IOException
     */
    public boolean queueDeclare(String queueName, boolean durable, boolean exclusive, boolean autoDelete,
                                Map<String, Object> arguments) throws IOException {
        QueueDeclareArguments queueDeclareArguments = new QueueDeclareArguments();
        queueDeclareArguments.setRid(generateRid());
        queueDeclareArguments.setChannelId(channelId);
        queueDeclareArguments.setQueueName(queueName);
        queueDeclareArguments.setDurable(durable);
        queueDeclareArguments.setExclusive(exclusive);
        queueDeclareArguments.setAutoDelete(autoDelete);
        queueDeclareArguments.setArguments(arguments);
        byte[] payload = BinaryTool.toBytes(queueDeclareArguments);

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

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

    /**
     * 6. 删除队列
     * @param queueName
     * @return
     * @throws IOException
     */
    public boolean queueDelete(String queueName) throws IOException {
        QueueDeleteArguments arguments = new QueueDeleteArguments();
        arguments.setRid(generateRid());
        arguments.setChannelId(channelId);
        arguments.setQueueName(queueName);
        byte[] payload = BinaryTool.toBytes(arguments);

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

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

    /**
     * 7. 创建绑定
     * @param queueName
     * @param exchangeName
     * @param bindingKey
     * @return
     * @throws IOException
     */
    public boolean queueBind(String queueName, String exchangeName, String bindingKey) throws IOException {
        QueueBindArguments arguments = new QueueBindArguments();
        arguments.setRid(generateRid());
        arguments.setChannelId(channelId);
        arguments.setQueueName(queueName);
        arguments.setExchangeName(exchangeName);
        arguments.setBindingKey(bindingKey);
        byte[] payload = BinaryTool.toBytes(arguments);

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

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

    /**
     * 8. 解除绑定
     * @param queueName
     * @param exchangeName
     * @return
     * @throws IOException
     */
    public boolean queueUnbind(String queueName, String exchangeName) throws IOException {
        QueueUnbindArguments arguments = new QueueUnbindArguments();
        arguments.setRid(generateRid());
        arguments.setChannelId(channelId);
        arguments.setQueueName(queueName);
        arguments.setExchangeName(exchangeName);
        byte[] payload = BinaryTool.toBytes(arguments);

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

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

    /**
     *  9. 发送消息
     * @param exchangeName
     * @param routingKey
     * @param basicProperties
     * @param body
     * @return
     * @throws IOException
     */

    public boolean basicPublish(String exchangeName, String routingKey, BasicProperties basicProperties, byte[] body) throws IOException {
        BasicPublishArguments arguments = new BasicPublishArguments();
        arguments.setRid(generateRid());
        arguments.setChannelId(channelId);
        arguments.setExchangeName(exchangeName);
        arguments.setRoutingKey(routingKey);
        arguments.setBasicProperties(basicProperties);
        arguments.setBody(body);
        byte[] payload = BinaryTool.toBytes(arguments);

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

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

    /**
     * 10. 订阅消息
     * @param queueName
     * @param autoAck
     * @param consumer
     * @return
     * @throws MqException
     * @throws IOException
     */
    public boolean basicConsume(String queueName, boolean autoAck, Consumer consumer) throws MqException, IOException {
        // 先设置回调.
        if (this.consumer != null) {
            throw new MqException("该 channel 已经设置过消费消息的回调了, 不能重复设置!");
        }
        this.consumer = consumer;

        BasicConsumeArguments arguments = new BasicConsumeArguments();
        arguments.setRid(generateRid());
        arguments.setChannelId(channelId);
        arguments.setConsumeTag(channelId);  // 此处 consumerTag 也使用 channelId 来表示了.
        arguments.setQueueName(queueName);
        arguments.setAutoAck(autoAck);
        byte[] payload = BinaryTool.toBytes(arguments);

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

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

    /**
     * 11. 确认消息
     * @param queueName
     * @param messageId
     * @return
     * @throws IOException
     */
    public boolean basicAck(String queueName, String messageId) throws IOException {
        BasicAckArguments arguments = new BasicAckArguments();
        arguments.setRid(generateRid());
        arguments.setChannelId(channelId);
        arguments.setQueueName(queueName);
        arguments.setMessageId(messageId);
        byte[] payload = BinaryTool.toBytes(arguments);

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

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

5. 处理响应

5.1 在Connection中创建扫描线程

创建扫描线程,用来不停地读取Socket中的响应数据.

1. ⼀个Connection中可能包含多个channel,需要把响应分别放到对应的channel中.

2. 读取响应:(要实现响应的分发)

        1. 正常调用服务器功能,返回的响应. (写回到change中存放数据的哈希表中)

        2. 客户端订阅的队列推送给客户端的消息(提交给线程池,进行处理回调)

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);

        // 创建一个扫描线程,来获取响应信息
        Thread thread = new Thread(()->{
            try {
                while (!socket.isClosed()){
                    Response response = readResponse();
                    // 响应的类型主要有两种
                    // 1. 一种是正常进行调用服务器的功能,得到的响应
                    // 2. 一种是服务器给订阅了消息的消费者进行推送消息
                    dispatchResponse(response);
                }
            } catch (SocketException e){
                System.out.println("[Connection] 连接正常断开");
            }
            catch (IOException | ClassNotFoundException | MqException e) {
                System.out.println("[Connection] 连接异常断开");
                e.printStackTrace();
            }
        });
        thread.start();
    }

5.2 分发响应

1. 客户端正常请求,服务器进行响应

        1. 将响应中的payload进行反序列化得到BasicReturns

        2. 根据响应中的ChannelID在Connection中获取Channel对象,并对这个对象进行验证

        3. 将响应写入到对应的Channel中

2. 客户端订阅了消息,服务器给客户端不停的推送消息

        1. 将响应中的payload进行反序列化得到BasicReturns

        2. 根据响应中的ChannelID在Connection中获取Channel对象,并对这个对象进行验证

        3. 交给线程池执行回调函数,消费消息.

/**
     * 处理返回的两种不同的响应
     * @param response
     */
    private void dispatchResponse(Response response) throws IOException, ClassNotFoundException, MqException {
        // 分为两种响应
        // 1. 客户端正常请求,服务器进行响应
        // 2. 客户端订阅了消息,服务器给客户端不停的推送消息
        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=" +subScribeReturns.getChannelId());
            }
            callbackPool.submit(()->{
                try {
                    channel.getConsumer().handleDelivery(subScribeReturns.getConsumerTag(),
                            subScribeReturns.getBasicProperties(),subScribeReturns.getBody());
                } catch (IOException | MqException e) {
                    e.printStackTrace();
                }
            });
        }else {
            // 客户端正常请求,服务器进行响应
            BasicReturns basicReturns = (BasicReturns)BinaryTool.fromBytes(response.getPayload());
            Channel channel = channelMap.get(basicReturns.getChannelId());
            if (channel == null){
                throw new MqException("[Connection] 该消息对应的channel不存在, ChannelId=" +basicReturns.getChannelId());
            }
            // 将此时的返回响应写入到channel对象用来存储请求与响应信息的Map里面
            channel.putReturns(basicReturns);
        }
    }

6. 关闭Connection

1. 关闭线程池服务

2. 清空Connection中的Channelmap

3. 关闭输入输出流

4. 关闭套接字

---

结语

        至此!!!,我们呢的项目终于完结了,本以为这个项目使用4个系列就能完成,但是为了能够深刻理解消息队列的思想,我还是觉得总结的更加详细点,这样才有意义.8个章节完成了整个消息队列的构建,内容还是很多的.但是还是有很多功能没有实现,不仅限于虚拟机的管理,死信队列等功能,但我相信大家跟着做完这个项目一定会有很大的提升.接下来还有最后一节,就是对我们实现的这个消息对列项目写一个Demo,使用生产者消费者模型对我们的消息队列进行测试.请大家继续关注,谢谢!!!❤️

        完整的项目代码已上传Gitee,欢迎大家访问.👇👇👇

模拟实现消息队列https://gitee.com/yao-fa/advanced-java-ee/tree/master/My-mq