【消息队列开发】 实现BrokerServer类——本体服务器

本文涉及的产品
云解析 DNS,旗舰版 1个月
全局流量管理 GTM,标准版 1个月
公共DNS(含HTTPDNS解析),每月1000万次HTTP解析
简介: 【消息队列开发】 实现BrokerServer类——本体服务器

🍃前言

本次开发任务

  • 实现 BrokerServer 类,也就是咱们消息队列的本体服务器。

其实本质上就是一个 TCP 的服务器

🎋创建 BrokerServer 类

创建 BrokerServer 类如下:

public class BrokerServer {
  // 当前程序只考虑⼀个虚拟主机的情况.
  private VirtualHost virtualHost = new VirtualHost("default-VirtualHost");
  // key 为 channelId, value 为 channel 对应的 socket 对象.
  private ConcurrentHashMap<String, Socket> sessions = new ConcurrentHashMap<>
  private ServerSocket serverSocket;
  private ExecutorService executorService;
  private volatile boolean runnable = true;
}
  • virtualHost表示服务器持有的虚拟主机.队列,交换机,绑定,消息都是通过虚拟主机管理.
  • sessions ⽤来管理所有的客⼾端的连接. 记录每个客户端的 socket.
  • serverSocket 是服务器自身的 socket
  • executorService 这个线程池用来处理响应
  • runnable 这个标志位用来控制服务器的运行停⽌

🎍启动与停止服务器

代码实现如下:

public BrokerServer(int port) throws IOException {
    serverSocket = new ServerSocket(port);
}
public void start() throws IOException {
    System.out.println("[BrokerServer] 启动!");
    executorService = Executors.newCachedThreadPool();
    try {
        while (runnable) {
            Socket clientSocket = serverSocket.accept();
            // 把处理连接的逻辑丢给这个线程池.
            executorService.submit(() -> {
                processConnection(clientSocket);
            });
        }
    } catch (SocketException e) {
        System.out.println("[BrokerServer] 服务器停止运行!");
        // e.printStackTrace();
    }
}
// 一般来说停止服务器, 就是直接 kill 掉对应进程就行了.
// 此处还是搞一个单独的停止方法. 主要是用于后续的单元测试.
public void stop() throws IOException {
    runnable = false;
    // 把线程池中的任务都放弃了. 让线程都销毁.
    executorService.shutdownNow();
    serverSocket.close();
}

🍀实现处理连接

通过这个方法, 来处理一个客户端的连接.

我们使用 InputStreamOutputStream,由于后面要按照特定格式来读取并解析.

此时就需要用到 DataInputStreamDataOutputStream

在这一个连接中, 可能会涉及到多个请求和响应,我们使用一个while(true)来进行实现

在此循环我们要做的事情有三件:

  1. 读取请求并解析
  2. 根据请求计算响应
  3. 把响应写回客户端

具体处理逻辑,我们后面再仔细实现,

那么我们怎么结束这个循环呢?

注意我们上面使用的是 DataInputStreamDataOutputStream,当没有数据进行读取的时候,就会进行抛出异常而结束循环

最后当连接处理完了, 就需要记得关闭 socket, 一个 TCP 连接中, 可能包含多个 channel. 需要把当前这个 socket 对应的所有 channel 也顺便清理掉.

代码实现如下:

// 通过这个方法, 来处理一个客户端的连接.
// 在这一个连接中, 可能会涉及到多个请求和响应.
private void processConnection(Socket clientSocket) {
    try (InputStream inputStream = clientSocket.getInputStream();
         OutputStream outputStream = clientSocket.getOutputStream()) {
        // 这里需要按照特定格式来读取并解析. 此时就需要用到 DataInputStream 和 DataOutputStream
        try (DataInputStream dataInputStream = new DataInputStream(inputStream);
             DataOutputStream dataOutputStream = new DataOutputStream(outputStream)) {
            while (true) {
                // 1. 读取请求并解析.
                Request request = readRequest(dataInputStream);
                // 2. 根据请求计算响应
                Response response = process(request, clientSocket);
                // 3. 把响应写回给客户端
                writeResponse(dataOutputStream, response);
            }
        }
    } catch (EOFException | SocketException e) {
        // 对于这个代码, DataInputStream 如果读到 EOF , 就会抛出一个 EOFException 异常.
        // 需要借助这个异常来结束循环
        System.out.println("[BrokerServer] connection 关闭! 客户端的地址: " + clientSocket.getInetAddress().toString()
                + ":" + clientSocket.getPort());
    } catch (IOException | ClassNotFoundException | MqException e) {
        System.out.println("[BrokerServer] connection 出现异常!");
        e.printStackTrace();
    } finally {
        try {
            // 当连接处理完了, 就需要记得关闭 socket
            clientSocket.close();
            // 一个 TCP 连接中, 可能包含多个 channel. 需要把当前这个 socket 对应的所有 channel 也顺便清理掉.
            clearClosedSession(clientSocket);
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
}

🎄实现 readRequest 与 writeResponse

image.png

关于读取请求,我们前面定义了一个类为 Request ,此时我们构造相应的对象,并对该对象相应属性进行填充即可。

代码实现如下:

private Request readRequest(DataInputStream dataInputStream) throws IOException {
    Request request = new Request();
    request.setType(dataInputStream.readInt());
    request.setLength(dataInputStream.readInt());
    byte[] payload = new byte[request.getLength()];
    int n = dataInputStream.read(payload);
    if (n != request.getLength()) {
        throw new IOException("读取请求格式出错!");
    }
    request.setPayload(payload);
    return request;
}

关于响应,实现相反,传入的 响应对象 相应的属性返回即可。

最后不要忘了刷新缓冲区

代码实现如下:

private void writeResponse(DataOutputStream dataOutputStream, Response response) throws IOException {
    dataOutputStream.writeInt(response.getType());
    dataOutputStream.writeInt(response.getLength());
    dataOutputStream.write(response.getPayload());
    // 这个刷新缓冲区也是重要的操作!!
    dataOutputStream.flush();
}

🌴实现处理请求

先把请求转换成 BaseArguments , 获取到其中的 channelId 和 rid

再根据不同的 type, 分别处理不同的逻辑. (主要是调用virtualHost中不同的方法).

针对消息订阅操作,则需要在存在消息的时候通过回调,把响应结果写回给对应的客⼾端.

最后构造成统⼀的响应.

代码实现如下:

private Response process(Request request, Socket clientSocket) throws IOException, ClassNotFoundException, MqException {
    // 1. 把 request 中的 payload 做一个初步的解析.
    BasicArguments basicArguments = (BasicArguments) BinaryTool.fromBytes(request.getPayload());
    System.out.println("[Request] rid=" + basicArguments.getRid() + ", channelId=" + basicArguments.getChannelId()
            + ", type=" + request.getType() + ", length=" + request.getLength());
    // 2. 根据 type 的值, 来进一步区分接下来这次请求要干啥.
    boolean ok = true;
    if (request.getType() == 0x1) {
        // 创建 channel
        sessions.put(basicArguments.getChannelId(), clientSocket);
        System.out.println("[BrokerServer] 创建 channel 完成! channelId=" + basicArguments.getChannelId());
    } else if (request.getType() == 0x2) {
        // 销毁 channel
        sessions.remove(basicArguments.getChannelId());
        System.out.println("[BrokerServer] 销毁 channel 完成! channelId=" + basicArguments.getChannelId());
    } else if (request.getType() == 0x3) {
        // 创建交换机. 此时 payload 就是 ExchangeDeclareArguments 对象了.
        ExchangeDeclareArguments arguments = (ExchangeDeclareArguments) basicArguments;
        ok = virtualHost.exchangeDeclare(arguments.getExchangeName(), arguments.getExchangeType(),
                arguments.isDurable(), arguments.isAutoDelete(), arguments.getArguments());
    } else if (request.getType() == 0x4) {
        ExchangeDeleteArguments arguments = (ExchangeDeleteArguments) basicArguments;
        ok = virtualHost.exchangeDelete(arguments.getExchangeName());
    } else if (request.getType() == 0x5) {
        QueueDeclareArguments arguments = (QueueDeclareArguments) basicArguments;
        ok = virtualHost.queueDeclare(arguments.getQueueName(), arguments.isDurable(),
                arguments.isExclusive(), arguments.isAutoDelete(), arguments.getArguments());
    } else if (request.getType() == 0x6) {
        QueueDeleteArguments arguments = (QueueDeleteArguments) basicArguments;
        ok = virtualHost.queueDelete((arguments.getQueueName()));
    } else if (request.getType() == 0x7) {
        QueueBindArguments arguments = (QueueBindArguments) basicArguments;
        ok = virtualHost.queueBind(arguments.getQueueName(), arguments.getExchangeName(), arguments.getBindingKey());
    } else if (request.getType() == 0x8) {
        QueueUnbindArguments arguments = (QueueUnbindArguments) basicArguments;
        ok = virtualHost.queueUnbind(arguments.getQueueName(), arguments.getExchangeName());
    } else if (request.getType() == 0x9) {
        BasicPublishArguments arguments = (BasicPublishArguments) basicArguments;
        ok = virtualHost.basicPublish(arguments.getExchangeName(), arguments.getRoutingKey(),
                arguments.getBasicProperties(), arguments.getBody());
    } else if (request.getType() == 0xa) {
        BasicConsumeArguments arguments = (BasicConsumeArguments) basicArguments;
        ok = virtualHost.basicConsume(arguments.getConsumerTag(), arguments.getQueueName(), arguments.isAutoAck(),
                new Consumer() {
                    // 这个回调函数要做的工作, 就是把服务器收到的消息可以直接推送回对应的消费者客户端
                    @Override
                    public void handleDelivery(String consumerTag, BasicProperties basicProperties, byte[] body) throws MqException, IOException {
                        // 先知道当前这个收到的消息, 要发给哪个客户端.
                        // 此处 consumerTag 其实是 channelId. 根据 channelId 去 sessions 中查询, 就可以得到对应的
                        // socket 对象了, 从而可以往里面发送数据了
                        // 1. 根据 channelId 找到 socket 对象
                        Socket clientSocket = sessions.get(consumerTag);
                        if (clientSocket == null || clientSocket.isClosed()) {
                            throw new MqException("[BrokerServer] 订阅消息的客户端已经关闭!");
                        }
                        // 2. 构造响应数据
                        SubScribeReturns subScribeReturns = new SubScribeReturns();
                        subScribeReturns.setChannelId(consumerTag);
                        subScribeReturns.setRid(""); // 由于这里只有响应, 没有请求, 不需要去对应. rid 暂时不需要.
                        subScribeReturns.setOk(true);
                        subScribeReturns.setConsumerTag(consumerTag);
                        subScribeReturns.setBasicProperties(basicProperties);
                        subScribeReturns.setBody(body);
                        byte[] payload = BinaryTool.toBytes(subScribeReturns);
                        Response response = new Response();
                        // 0xc 表示服务器给消费者客户端推送的消息数据.
                        response.setType(0xc);
                        // response 的 payload 就是一个 SubScribeReturns
                        response.setLength(payload.length);
                        response.setPayload(payload);
                        // 3. 把数据写回给客户端.
                        //    注意! 此处的 dataOutputStream 这个对象不能 close !!!
                        //    如果 把 dataOutputStream 关闭, 就会直接把 clientSocket 里的 outputStream 也关了.
                        //    此时就无法继续往 socket 中写入后续数据了.
                        DataOutputStream dataOutputStream = new DataOutputStream(clientSocket.getOutputStream());
                        writeResponse(dataOutputStream, response);
                    }
                });
    } else if (request.getType() == 0xb) {
        // 调用 basicAck 确认消息.
        BasicAckArguments arguments = (BasicAckArguments) basicArguments;
        ok = virtualHost.basicAck(arguments.getQueueName(), arguments.getMessageId());
    } else {
        // 当前的 type 是非法的.
        throw new MqException("[BrokerServer] 未知的 type! type=" + request.getType());
    }
    // 3. 构造响应
    BasicReturns basicReturns = new BasicReturns();
    basicReturns.setChannelId(basicArguments.getChannelId());
    basicReturns.setRid(basicArguments.getRid());
    basicReturns.setOk(ok);
    byte[] payload = BinaryTool.toBytes(basicReturns);
    Response response = new Response();
    response.setType(request.getType());
    response.setLength(payload.length);
    response.setPayload(payload);
    System.out.println("[Response] rid=" + basicReturns.getRid() + ", channelId=" + basicReturns.getChannelId()
            + ", type=" + response.getType() + ", length=" + response.getLength());
    return response;
}

🌲实现 clearClosedSession

这里要做的事情, 主要就是遍历上述 sessions hash 表, 把该被关闭的 socket 对应的键值对, 统统删掉

需要注意的是:

  • 我们在进行迭代的时候,不要直接删除,这样会影响集合类的结构

代码实现如下:

private void clearClosedSession(Socket clientSocket) {
    // 这里要做的事情, 主要就是遍历上述 sessions hash 表, 把该被关闭的 socket 对应的键值对, 统统删掉.
    List<String> toDeleteChannelId = new ArrayList<>();
    for (Map.Entry<String, Socket> entry : sessions.entrySet()) {
        if (entry.getValue() == clientSocket) {
            // 不能在这里直接删除!!!
            // 这属于使用集合类的一个大忌!!! 一边遍历, 一边删除!!!
            // sessions.remove(entry.getKey());
            toDeleteChannelId.add(entry.getKey());
        }
    }
    for (String channelId : toDeleteChannelId) {
        sessions.remove(channelId);
    }
    System.out.println("[BrokerServer] 清理 session 完成! 被清理的 channelId=" + toDeleteChannelId);
}

⭕总结

关于《【消息队列开发】 实现BrokerServer类——本体服务器》就讲解到这儿,感谢大家的支持,欢迎各位留言交流以及批评指正,如果文章对您有帮助或者觉得作者写的还不错可以点一下关注,点赞,收藏支持一下

相关文章
|
2天前
|
关系型数据库 API 数据库
后端开发的艺术:从零到一构建高效服务器
在数字化时代,后端开发是支撑现代互联网应用的基石。本文旨在探讨后端开发的核心概念、关键技术以及如何构建一个高效的服务器。我们将从基础的编程语言选择开始,逐步深入到数据库设计、API开发和性能优化等关键领域。通过实际案例分析,我们将揭示后端开发的复杂性和挑战性,同时提供实用的解决方案和最佳实践。无论你是初学者还是有经验的开发者,这篇文章都将为你提供宝贵的见解和启发。
|
15天前
|
网络安全 Docker 容器
VScode远程服务器之远程 远程容器 进行开发(五)
VScode远程服务器之远程 远程容器 进行开发(五)
17 1
|
25天前
|
Java PHP
PHP作为广受青睐的服务器端脚本语言,在Web开发中占据重要地位。理解其垃圾回收机制有助于开发高效稳定的PHP应用。
【10月更文挑战第1天】PHP作为广受青睐的服务器端脚本语言,在Web开发中占据重要地位。其垃圾回收机制包括引用计数与循环垃圾回收,对提升应用性能和稳定性至关重要。本文通过具体案例分析,详细探讨PHP垃圾回收机制的工作原理,特别是如何解决循环引用问题。在PHP 8中,垃圾回收机制得到进一步优化,提高了效率和准确性。理解这些机制有助于开发高效稳定的PHP应用。
39 3
|
6天前
|
NoSQL PHP Redis
布谷语音app源码服务器环境配置及技术开发语言
布谷语音app源码服务器环境配置及技术语言研发。。
|
15天前
|
Kubernetes 网络安全 容器
VScode远程服务器进行开发(三)
VScode远程服务器进行开发(三)
20 0
|
26天前
|
Linux 开发工具 Docker
各个类linux服务器安装docker教程
各个类linux服务器安装docker教程
49 0
|
2月前
|
开发框架 JavaScript 前端开发
|
3月前
|
前端开发 大数据 数据库
🔥大数据洪流下的决战:JSF 表格组件如何做到毫秒级响应?揭秘背后的性能魔法!💪
【8月更文挑战第31天】在 Web 应用中,表格组件常用于展示和操作数据,但在大数据量下性能会成瓶颈。本文介绍在 JavaServer Faces(JSF)中优化表格组件的方法,包括数据处理、分页及懒加载等技术。通过后端分页或懒加载按需加载数据,减少不必要的数据加载和优化数据库查询,并利用缓存机制减少数据库访问次数,从而提高表格组件的响应速度和整体性能。掌握这些最佳实践对开发高性能 JSF 应用至关重要。
63 0
|
3月前
|
前端开发 安全 开发者
JSF文件上传,让Web应用如虎添翼!一招实现文件上传,让用户爱不释手!
【8月更文挑战第31天】在现代Web应用开发中,文件上传是重要功能之一。JSF(JavaServer Faces)框架提供了强大的文件上传支持,简化了开发流程。本文将介绍JSF文件上传的基本步骤:创建前端表单、处理上传文件的后端Action类、将文件保存到服务器指定目录以及返回结果页面。通过示例代码,我们将展示如何利用JSF实现文件上传功能,包括使用`h:inputFile`控件和`ManagedBean`处理上传逻辑。此外,JSF文件上传还具备类型安全、解耦合和灵活性等优点,有助于提升程序的健壮性和可维护性。
37 0
|
3月前
|
缓存 前端开发 测试技术
Angular 与 Angular Universal 超厉害!实现服务器端渲染,开启高效开发与优质用户体验之旅!
【8月更文挑战第31天】在现代前端开发中,用户体验至关重要。Angular Universal 为 Angular 应用提供服务器端渲染(SSR)解决方案,显著提升性能和可访问性。本文介绍 SSR 的优势,包括更快的首屏加载时间、更好的 SEO 和更稳定的用户体验。接着详细讲解安装和配置 Angular Universal 的步骤,并提供优化 SSR 性能的方法,如延迟加载、缓存和预取技术。同时讨论了处理动态内容的策略,并介绍了测试和部署的最佳实践。通过遵循这些实践,可以确保应用高效稳定运行,提升用户体验。附带的示例代码展示了如何在服务器端和客户端处理数据。
33 0