目录
前言
1. 明确需求
2. 设计应用层协议
2. 定义Request 和 Response
3. 定义参数的父类
4. 定义其他参数类
4.1 创建交换机
4.2 删除交换机
4.3 创建队列
4.4 删除队列
4.5 创建绑定
4.6 删除绑定
4.7 发布消息
4.8 订阅消息
4.9 推送消息
结语
前言
上一章节,我们完成了虚拟主机的封装,同时实现了API供服务器进行调用.那么从这一章节,我们就要将消息队列进行添加网络通信了,毕竟消息队列的目的就是实现消息的转发,网络通信是必须要实现的.在开始之前,我们思考一个问题,我们之前的博客系统,我们采用的是http的通信协议实现前后端的通信的,但是我们之前传输的是文本文件,这里我们的Message是二进制文件,所以我们要使用一个支持二进制传输的应用层协议,这里我们不打算使用市面的应用层协议.我们自己定义一个应用层协议.然后使用TCP协议作为通信底层的协议.本项目全部代码已上传Gitee,链接放在文章末尾,欢迎大家访问!
1. 明确需求
生产者和消费者都是客户端,都需要通过网络和BrokerServer进行通信.需要调用服务器的功能如下:
2. 设计应用层协议
2. 定义Request 和 Response
表示网络通信的请求和响应对象.
@Getter
@Setter
public class Request {
private int type;
private int length;
private byte[] payload;
}@Getter
@Setter
public class Response {
private int type;
private int length;
private byte[] payload;
}3. 定义参数的父类
对于请求的参数我们放在payload里面,但是每种请求的参数都是不一样的,但是会有公共的参数属性:
- 1. rid:此次请求的唯一ID,用来对应请求和响应 (使用UUID随机成成唯一的ID)
- 2. ChannelID : 表示此次通信的ChannelID
@Setter
@Getter
public class BasicArguments implements Serializable {
// 表⽰⼀次请求/响应的唯⼀ id. ⽤来把响应和请求对上.
protected String rid;
// 此次通信的channel的身份标识
protected String channelId;
}对于响应也是一样,定义一个父类,不同的响应返回不同的值,继承父类的公共属性
- 1. rid:此次请求的唯一ID,用来对应请求和响应 (使用UUID随机成成唯一的ID)
- 2. ChannelID : 表示此次通信的ChannelID
- 3. ok:表示响应是否成功
@Getter
@Setter
public class BasicReturns implements Serializable {
// 表⽰⼀次请求/响应的唯⼀ id. ⽤来把响应和请求对上.
protected String rid;
protected String channelId;
protected boolean ok;
}4. 定义其他参数类
4.1 创建交换机
下面以创建交换机为例:
我们针对以上参数设计属性
@Getter
@Setter
public class ExchangeDeclareArguments extends BasicArguments implements Serializable {
private String exchangeName;
private ExchangeType exchangeType;
private boolean durable;
private boolean autoDelete;
private Map<String, Object> arguments;
}最后构造出来的请求如下:
后续的请求也是这样只不过是payload的内容和Type的类型以及length发生改变,但是结构不会变.
下面给出对应不同请求的参数类.
4.2 删除交换机
@Setter
@Getter
public class ExchangeDeleteArguments extends BasicArguments implements Serializable {
private String exchangeName;
}4.3 创建队列
@Getter
@Setter
public class QueueDeclareArguments extends BasicArguments implements Serializable {
private String queueName;
private boolean durable;
private boolean exclusive;
private boolean autoDelete;
private Map<String, Object> arguments;
}4.4 删除队列
@Setter
@Getter
public class QueueDeleteArguments extends BasicArguments implements Serializable {
private String queueName;
}4.5 创建绑定
@Getter
@Setter
public class QueueBindArguments extends BasicArguments implements Serializable {
private String queueName;
private String exchangeName;
private String bindingKey;
}4.6 删除绑定
@Getter
@Setter
public class QueueUnbindArguments extends BasicArguments implements Serializable {
private String queueName;
private String exchangeName;
}4.7 发布消息
@Getter
@Setter
public class BasicPublishArguments extends BasicArguments implements Serializable {
private String exchangeName;
private String routingKey;
private BasicProperties basicProperties;
private byte[] body;
}4.8 订阅消息
这里需要特别说明一下:这个类对应的订阅消息的方法的参数还有一个回调函数,回调函数不能通过网络进行传输的,站在消息队列服务器这边,用户执行这个回调函数是统一的行为.不在乎行为的具体内容.
@Getter
@Setter
public class BasicConsumeArguments extends BasicArguments implements Serializable {
private String consumeTag;
private String queueName;
private boolean autoAck;
// 这个类对应的订阅消息的方法的参数还有一个回调函数
// 回调函数不能通过网络进行传输的
// 站在消息队列服务器这边,用户执行这个回调函数是统一的行为.不在乎行为的具体内容
}4.9 推送消息
这个不是参数,是返回值.是服务器给消费者推送的订阅消息.当消费者订阅了队列,消息队列就会从这个队列中取出消息推送给消费者.
结语
以上我们自己实现了应用层的协议,为BrokerServer与客户端通信奠定了基础.
下一章节正式进入BrokerServer的封装,请大家持续关注.
完整的项目代码已上传Gitee,欢迎大家访问.👇👇👇
模拟实现消息队列
https://gitee.com/yao-fa/advanced-java-ee/tree/master/My-mq
