【消息队列开发】 创建核心类

/*
 * 这个类表示一个交换机
 */
public class Exchange {
    // 此处使用 name 来作为交换机的身份标识. (唯一的)
    private String name;
    // 交换机类型, DIRECT(专属红包), FANOUT(群发红包), TOPIC(指令红包)
    private ExchangeType type = ExchangeType.DIRECT;
    private boolean durable = false;
    // 如果当前交换机, 没人使用了, 就会自动被删除.
    // 这个属性暂时先列在这里, 后续的代码中并没有真的实现这个自动删除功能~~ (RabbitMQ 是有的)
    private boolean autoDelete = false;
    // arguments 表示的是创建交换机时指定的一些额外的参数选项. 后续代码中并没有真的实现对应的功能, 先列出来. (RabbitMQ 也是有的)
    // 为了把这个 arguments 存到数据库中, 就需要把 Map 转成 json 格式的字符串.
    private Map<String, Object> arguments = new HashMap<>();
    public String getName() { return name;}
    public void setName(String name) {this.name = name;}
    public ExchangeType getType() {
        return type;
    }
    public void setType(ExchangeType type) {
        this.type = type;
    }
    public boolean isDurable() {
        return durable;
    }
    public void setDurable(boolean durable) {
        this.durable = durable;
    }
    public boolean isAutoDelete() {
        return autoDelete;
    }
    public void setAutoDelete(boolean autoDelete) {
        this.autoDelete = autoDelete;
    }
    public Map<String, Object> getArguments() {
        return arguments;
    }
    public void setArguments(Map<String, Object> arguments) {
        this.arguments = arguments;
    }

/*
这是一个枚举类，里面表示的是交换机的类型
 */
public enum ExchangeType {
    DIRECT(0),
    FANOUT(1),
    TOPIC(2);
    private final int type;
    //private封装，不想让外界访问
    private ExchangeType(int type) {
        this.type = type;
    }
    public int getType() {
        return type;
    }
}

/*
这表示一个队列
 */
public class MSGQueue {
    // 表示队列的身份标识.
    private String name;
    // 表示队列是否持久化, true 表示持久化保存, false 表示不持久化.
    private boolean durable = false;
    // 这个属性为 true, 表示这个队列只能被一个消费者使用(别人用不了). 如果为 false 则是大家都能使用
    // 这个 独占 功能, 也是先把字段列在这里, 具体的独占功能暂时先不实现.
    private boolean exclusive = false;
    // 为 true 表示没有人使用之后, 就自动删除. false 则是不会自动删除.
    // 这个 自动删除 功能, 也是先把字段列在这里, 具体的自动删除功能暂时先不实现.
    private boolean autoDelete = false;
    // 也是表示扩展参数. 当前也是先列在这里, 先暂时不实现
    private Map<String, Object> arguments = new HashMap<>();
    public String getName() {
        return name;
    }
    public void setName(String name) {
        this.name = name;
    }
    public boolean isDurable() {
        return durable;
    }
    public void setDurable(boolean durable) {
        this.durable = durable;
    }
    public boolean isExclusive() {
        return exclusive;
    }
    public void setExclusive(boolean exclusive) {
        this.exclusive = exclusive;
    }
    public boolean isAutoDelete() {
        return autoDelete;
    }
    public void setAutoDelete(boolean autoDelete) {
        this.autoDelete = autoDelete;
    }
    public Map<String, Object> getArguments() {
        return arguments;
    }
    public void setArguments(Map<String, Object> arguments) {
        this.arguments = arguments;
    }
}

*
表示队列和交换机之间的关系
 */
public class Binding {
    //交换机名字
    private String exchangeName;
    //队列名字
    private String queueName;
    // bindingKey对应的交换机类型为Topic，主题匹配
    // 相当于就是在出题, 要求领红包的人要画个 "桌子" 出来~~
    private String bindingKey;
    //此处没有设置是否持久化存储的原因
    // Binding 这个东西, 是依附于 Exchange 和 Queue 的!!!
    // 比如, 对于持久化来说, 如果 Exchange 和 Queue 任何一个都没有持久化,
    // 此时你针对 Binding 持久化是没有任何意义的
    public String getExchangeName() {
        return exchangeName;
    }
    public void setExchangeName(String exchangeName) {
        this.exchangeName = exchangeName;
    }
    public String getQueueName() {
        return queueName;
    }
    public void setQueueName(String queueName) {
        this.queueName = queueName;
    }
    public String getBindingKey() {
        return bindingKey;
    }
    public void setBindingKey(String bindingKey) {
        this.bindingKey = bindingKey;
    }
}

/*
 * 表示要传递的消息
 * 注意!!! 此处的 Message 对象, 是需要能够在网络上传输, 并且也需要能写入到文件中.
 * 此时就需要针对 Message 进行序列化和反序列化.
 * 此处使用 标准库 自带的 序列化/反序列化 操作.
 */
public class Message implements Serializable {
    //消息的属性
    private BasicProperties basicProperties = new BasicProperties();
    //消息的正文
    private byte[] body;
    // 下面的属性则是辅助用的属性.
    // Message 后续会存储到文件中(如果持久化的话).
    // 一个文件中会存储很多的消息. 如何找到某个消息, 在文件中的具体位置呢?
    // 使用下列的两个偏移量来进行表示. [offsetBeg, offsetEnd)
    // 这俩属性并不需要被序列化保存到文件中~~ 此时消息一旦被写入文件之后, 所在的位置就固定了. 并不需要单独存储.
    // 这俩属性存在的目的, 主要就是为了让内存中的 Message 对象, 能够快速找到对应的硬盘上的 Message 的位置.
    private transient long offsetBeg = 0;  // 消息数据的开头距离文件开头的位置偏移(字节)
    private transient long offsetEnd = 0;  // 消息数据的结尾距离文件开头的位置偏移(字节)
    // 使用这个属性表示该消息在文件中是否是有效消息. (针对文件中的消息, 如果删除, 使用逻辑删除的方式)
    // 0x1 表示有效. 0x0 表示无效.
    private byte isValid = 0x1;
    // 创建一个工厂方法, 让工厂方法帮我们封装一下创建 Message 对象的过程.
    // 这个方法中创建的 Message 对象, 会自动生成唯一的 MessageId
    // 万一 routingKey 和 basicProperties 里的 routingKey 冲突, 以外面的为主.
    public Message createMessageWithId(String routingKey, BasicProperties basicProperties, byte[] body) {
        Message message = new Message();
        if(basicProperties != null) {
            message.setBasicProperties(basicProperties);
        }
        //此处生成的 MessageId 以 M- 作为前缀,与其他id进行区分
        message.setMessageId("M-" + UUID.randomUUID());
        message.setRoutingKey(routingKey);
        message.body = body;
        // 此处是把 body 和 basicProperties 先设置出来. 他俩是 Message 的核心内容.
        // 而 offsetBeg, offsetEnd, isValid, 则是消息持久化的时候才会用到. 在把消息写入文件之前再进行设定.
        // 此处只是在内存中创建一个 Message 对象,所以不设置.
        return message;
    }
    public String getMessageId() {
        return basicProperties.getMessageId();
    }
    public void setMessageId(String messageId) {
        basicProperties.setMessageId(messageId);
    }
    public String getRoutingKey() {
        return basicProperties.getRoutingKey();
    }
    public void setRoutingKey(String routingKey) {
        basicProperties.setRoutingKey(routingKey);
    }
    public int getDeliverMode() {
        return basicProperties.getDeliverMode();
    }
    public void setDeliverMode(int mode) {
        basicProperties.setDeliverMode(mode);
    }
    public BasicProperties getBasicProperties() {
        return basicProperties;
    }
    public void setBasicProperties(BasicProperties basicProperties) {
        this.basicProperties = basicProperties;
    }
    public byte[] getBody() {
        return body;
    }
    public void setBody(byte[] body) {
        this.body = body;
    }
    public long getOffsetBeg() {
        return offsetBeg;
    }
    public void setOffsetBeg(long offsetBeg) {
        this.offsetBeg = offsetBeg;
    }
    public long getOffsetEnd() {
        return offsetEnd;
    }
    public void setOffsetEnd(long offsetEnd) {
        this.offsetEnd = offsetEnd;
    }
    public byte getIsValid() {
        return isValid;
    }
    public void setIsValid(byte isValid) {
        this.isValid = isValid;
    }
}

/*
 * 此处代码为 Message 对象相应的属性
 * 注意!!! 此处的 BasicProperties 对象, 是需要能够在网络上传输, 并且也需要能写入到文件中.
 * 此时就需要针对 BasicProperties 进行序列化和反序列化.
 * 此处使用 标准库 自带的 序列化/反序列化 操作.
 */
public class BasicProperties implements Serializable {
    // 消息的唯一身份标识. 此处为了保证 id 的唯一性, 使用 UUID 来作为 message id
    private String messageId;
    // 是一个消息上带有的内容, 和 bindingKey 做匹配.
    // 如果当前的交换机类型是 DIRECT, 此时 routingKey 就表示要转发的队列名.
    // 如果当前的交换机类型是 FANOUT, 此时 routingKey 无意义(不使用).
    // 如果当前的交换机类型是 TOPIC, 此时 routingKey 就要和 bindingKey 做匹配. 符合要求的才能转发给对应队列.
    private String routingKey;
    // 这个属性表示消息是否要持久化. 1 表示不持久化, 2 表示持久化. (RabbitMQ 就是这样搞的....)
    private int deliverMode = 1;
    //为了后续代码更加方便
    public String getMessageId() {
        return messageId;
    }
    public void setMessageId(String messageId) {
        this.messageId = messageId;
    }
    public String getRoutingKey() {
        return routingKey;
    }
    public void setRoutingKey(String routingKey) {
        this.routingKey = routingKey;
    }
    public int getDeliverMode() {
        return deliverMode;
    }
    public void setDeliverMode(int deliverMode) {
        this.deliverMode = deliverMode;
    }
}