目录
前言
1. 创建BrokerServer类
1.1 启动服务器
1.2 停止服务器
1.3 处理一个客户端的连接
1.3.1 解析请求得到Request对象
1.3.2 根据请求计算响应
1.3.3 将响应写回给客户端
1.3.4 遍历Session的哈希表,把断开的Socket对象的键值对进行删除
2. 处理订阅消息请求详解(补充)
3. 序列化/反序列化实现(补充)
结语
前言
上一章节,我们定义了本项目的应用层传输协议.并且创建了各种参数类.本章节的目标是对BrokerServer(实现一个TCP服务器)进行实现,对连接进行处理,根据请求计算响应返回给客户端.
1. 创建BrokerServer类
public class BrokerServer {
// 当前考虑一个一个服务器中只有一个虚拟主机
private VirtualHost virtualHost = new VirtualHost("default");
// 使用哈希表表示当前会话,也就是有哪些客户端在和服务器进行通信
// key: channelId value:对应对的Socket对象
private ConcurrentHashMap<String, Socket> sessions = new ConcurrentHashMap<>();
private ServerSocket serverSocket = null;
// 引入线程池来处理多个客户端
private ExecutorService executorService = null;
// 控制服务器是否继续运行
private volatile boolean runnable = true;
}1.1 启动服务器
1. 首先将线程池进行创建,用来处理多个连接.
2. 设置循环用来监听连接
3. 将处理连接交给线程池.
/**
* 1. 启动服务器
*/
public void start() throws IOException {
System.out.println("[BrokerServer] 启动!");
// newCachedThreadPool自动申请新的线程
executorService = Executors.newCachedThreadPool();
try {
while (runnable){
Socket clientSocket = serverSocket.accept();
// 把处理连接的逻辑发送给线程池
executorService.submit(()->{
processConnection(clientSocket);
});
}
}catch (SocketException e){
System.out.println("[BrokerServer] 服务器停止运行!");
// e.printStackTrace();
}
}1.2 停止服务器
1. 将标志位runnable设置为false
2. 停止线程池的服务
3. 关闭服务器套接字
/**
* 2. 停止服务器
*/
public void stop() throws IOException {
runnable = false;
// 停止线程池
executorService.shutdownNow();
serverSocket.close();
}
1.3 处理一个客户端的连接
1. 我们是从请求中获取的信息是二进制文件,我们不能直接使用InputStream和OutputStream,我们借助DataInputStream和DataOutputStream进行操作字节流.
2. 使用DataInputStream进行读取请求的时候,读到末尾的时候会抛出一个异常,我们将这个异常视作为处理正确的业务逻辑.我么catch掉这个异常就可以.
3. 解析得到请求对象
4. 更具请求计算响应
5. 当处理完响应之后,要进行关闭连接,并且将一个连接中其他Channel进行关闭.
/**
* 3. 处理一个客户端的连接
* 在一个连接中会出现多个请求和多个响应.在一个连接中要循环的处理
*/
private void processConnection(Socket clientSocket) {
try(InputStream inputStream = clientSocket.getInputStream();
OutputStream outputStream = clientSocket.getOutputStream()){
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){
// 处理正确的业务逻辑
// 上述进行读取数据的时候,如果数据读到末尾(EOF) ,就会抛出一个异常
// 借助这个异常结束上述循环
System.out.println("[BrokerServer] 连接关闭! 客户端地址:" +
clientSocket.getInetAddress().toString()
+ ",端口号: "+clientSocket.getPort());
}
catch (ClassNotFoundException | MqException e) {
e.printStackTrace();
} catch (IOException e) {
// 处理真正的异常
System.out.println("[BrokerServer] connection 出现异常");
e.printStackTrace();
}finally {
try {
// 当连接处理完成之后,进行关闭连接
clientSocket.close();
// 一个连接中可能会包含多个channel,需要把当前这个Socket对应的所有channel进行关闭
clearClosedSession(clientSocket);
} catch (IOException e) {
e.printStackTrace();
}
}
}1.3.1 解析请求得到Request对象
1. 根据我们自定义的格式,先读前4个字节是请求的类型,在读4个字节是payload的长度,在读就是payload.
2. 读取payload的时候,我们先根据长度创建字符数组,然后按照字符数组进行获取payload,比较读取完的长度是否与原来请求的长度一致,不一致说明有消息的丢失.进行抛出异常.
3. 最后得到完整的请求对象,交给下面的方法进行处理.
/**
* 3.1 解析请求得到Request对象
*/
private Request readRequest(DataInputStream dataInputStream) throws IOException {
Request request = new Request();
// 1. 首先读取四个字节,为请求的type
request.setType(dataInputStream.readInt());
// 2. 在读四个字节就是payload的长度
request.setLength(dataInputStream.readInt());
// 3. 创建字符数组,并进行读取到数组中
byte[] payload = new byte[request.getLength()];
int n = dataInputStream.read(payload);
if (n != request.getLength()){
throw new IOException("[BrokerServer] 请求格式出错");
}
// 4. 将读取的数组内容写入到实体的Request对象中
request.setPayload(payload);
return request;
}1.3.2 根据请求计算响应
1. 我们根据请求对象的payload进行解析,此处需要注意的是,我们读取到的payload是字节数组,我们需要进行反序列化成字符数组.
2. 根据请求对象的Type值进行区分,到底客户端要调用服务器那些功能.
3. 处理完请求之后就要进行构造响应了.
4. 返回响应对象
/**
* 3.2 根据请求计算响应
* @param request
* @param clientSocket
* @return
*/
private Response process(Request request, Socket clientSocket) throws IOException, ClassNotFoundException, MqException {
// 1. 根据request中的payload进行解析
// payload 是根据 request 中 type 进行变化的
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){
// 1. 创建一个channel
sessions.put(basicArguments.getChannelId(),clientSocket);
System.out.println("[BrokerServer] 创建channel完成 getChannelId="+ basicArguments.getChannelId());
}else if (request.getType() == 0x2){
// 2. 销毁一个channel
sessions.remove(basicArguments.getChannelId());
System.out.println("[BrokerServer] 销毁channel完成 getChannelId="+ basicArguments.getChannelId());
}else if (request.getType() == 0x3){
// 3. 创建交换机
ExchangeDeclareArguments arguments = (ExchangeDeclareArguments) basicArguments;
// 调用虚拟主机的功能方法
ok = virtualHost.exchangeDeclare(arguments.getExchangeName(),
arguments.getExchangeType(),
arguments.isDurable(),
arguments.isAutoDelete(),
arguments.getArguments());
System.out.println("[BrokerServer] 创建交换机完成 ExchangeName="+ arguments.getExchangeName());
}else if (request.getType() == 0x4){
// 4. 销毁交换机
ExchangeDeleteArguments arguments = (ExchangeDeleteArguments) basicArguments;
ok = virtualHost.exchangeDelete(arguments.getExchangeName());
System.out.println("[BrokerServer] 删除交换机完成 ExchangeName="+ arguments.getExchangeName());
}else if (request.getType() == 0x5) {
// 5. 创建队列
QueueDeclareArguments arguments = (QueueDeclareArguments) basicArguments;
ok = virtualHost.queueDeclare(arguments.getQueueName(), arguments.isDurable(),
arguments.isExclusive(), arguments.isAutoDelete(), arguments.getArguments());
System.out.println("[BrokerServer] 创建队列完成 QueueName="+ arguments.getQueueName());
} else if (request.getType() == 0x6) {
// 6. 删除队列
QueueDeleteArguments arguments = (QueueDeleteArguments) basicArguments;
ok = virtualHost.queueDelete((arguments.getQueueName()));
System.out.println("[BrokerServer] 删除队列完成 QueueName="+ arguments.getQueueName());
} else if (request.getType() == 0x7) {
// 7. 创建绑定
QueueBindArguments arguments = (QueueBindArguments) basicArguments;
ok = virtualHost.queueBind(arguments.getQueueName(),
arguments.getExchangeName(), arguments.getBindingKey());
System.out.println("[BrokerServer] 创建绑定完成 QueueName="+ arguments.getQueueName()
+ ",ExchangeName=" + arguments.getExchangeName());
} else if (request.getType() == 0x8) {
// 8. 删除绑定
QueueUnbindArguments arguments = (QueueUnbindArguments) basicArguments;
ok = virtualHost.queueUnbind(arguments.getQueueName(), arguments.getExchangeName());
System.out.println("[BrokerServer] 删除绑定完成 QueueName="+ arguments.getQueueName()
+ ",ExchangeName=" + arguments.getExchangeName());
} else if (request.getType() == 0x9) {
// 9. 发布消息
BasicPublishArguments arguments = (BasicPublishArguments) basicArguments;
ok = virtualHost.basicPublish(arguments.getExchangeName(), arguments.getRoutingKey(),
arguments.getBasicProperties(), arguments.getBody());
System.out.println("[BrokerServer] 发布消息完成 ExchangeName=" + arguments.getExchangeName());
}else if (request.getType() == 0xa) {
// 10. 订阅消息
BasicConsumeArguments arguments = (BasicConsumeArguments) basicArguments;
ok = virtualHost.basicConsume(arguments.getConsumeTag(), arguments.getQueueName(), arguments.isAutoAck(), new Consumer() {
@Override
public void handleDelivery(String consumerTag, BasicProperties basicProperties, byte[] body) throws IOException, MqException {
// 将服务器收到的消息进行推送给客户端
// 先知道当前这个收到的消息, 要发给哪个客户端.
// 此处 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) {
// 10. 调用 basicAck 确认消息.
BasicAckArguments arguments = (BasicAckArguments) basicArguments;
ok = virtualHost.basicAck(arguments.getQueueName(), arguments.getMessageId());
System.out.println("[BrokerServer] 消费者确认消息完成 QueueName=" + arguments.getQueueName()
+ ", MessageId=" + 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;
}1.3.3 将响应写回给客户端
1. 注意写入类型和长度是写入固定的4个字节,那么我们就使用dataOutputStream.writeInt()
2. 写完响应之后,记得要刷新缓冲区 dataOutputStream.flush();
/**
* 3.3 将响应写回给客户端
* @param dataOutputStream
* @param response
*/
private void writeResponse(DataOutputStream dataOutputStream, Response response) throws IOException {
// 将响应的属性从计算好的响应中进行设置
dataOutputStream.writeInt(response.getType());
dataOutputStream.writeInt(response.getLength());
dataOutputStream.write(response.getPayload());
// 刷新缓冲区
dataOutputStream.flush();
}1.3.4 遍历Session的哈希表,把断开的Socket对象的键值对进行删除
由于Socket都已经断开连接了,那么存储在内存中的Session也就没有存在的必要了.这个集合中存放的是一个连接中的change对应的Session,当连接断开之后,Channel也就不会再进行工作了,新的连接会创建新的Channel.
注意:我们在使用Map.entrySet进行遍历Map的时候,不要一遍遍历一遍进行删除,这样是不稳定的,我们遍历Map将需要进行移除的Session进行添加到待删除的链表中,最后遍历待删除的数据结构进行删除.
上述就是整个封装好的BrokerServer服务器.
下面呢,我对有关根据请求计算响应中订阅消息这一功能,再进行详细的阐述,这块比较难以理解,因为涉及到回调函数,大家可能不知道这个回调函数掉用的时机是哪里.
2. 处理订阅消息请求详解(补充)
第二个红框部分是回调函数.
只有消费者订阅的队列中有消息了,并且轮询的方式选中了这个消费者,才会获得消息的本体,此时线程池才会执行到这个回调方法,此时才拿到消息的本体,可以将消息的属性和本体写入到SubscribeReturn中,进而推送给消费者进行消费消息.如果没有消息给这个消费者,那么也不会进行断开连接,只要服务器不断开连接客户端一直在等待分配的消息进行消费.这一点希望,读者能够进一步的理解.等总结完客户端,那么我就会带着大家,再来理一遍这个订阅消息的这个思路.
3. 序列化/反序列化实现(补充)
要想能进行序列化和反序列化就必须对目标对象进行实现serializable接口.
1. 我们使用ByteArrayOutputStream和ObjectOutputStream进行将一个对象序列化为字节数组(输出的是字节用output)
2. 我们ByteArrayInputStream和ObjectInputStream将一个字节数组反序列化成一个对象(输入的是字节用Input)
结语
至此,我们就彻底的完成了mqserver 的搭建,只剩下mqclient的搭建,我们在下一系列完成客户端的搭建,请持续关注,谢谢!!!
完整的项目代码已上传Gitee,欢迎大家访问.👇👇👇
模拟实现消息队列
https://gitee.com/yao-fa/advanced-java-ee/tree/master/My-mq
