当前位置：首页 > news >正文

RabbitMinQ(模拟实现消息队列项目)02

news 2025/9/2 6:01:41

十.整合数据库和文件数据

创建DiskDataManager类

十一.内存结构设计

创建MeneryDataCenter类:

实现集合操作:

对MemoryDataCenter类功能测试:

十二.整合内存和磁盘数据

创建VirtualHost类:

Exchange:

MSGQueue:

Binding:

创建Router类

对Router类的TOPIC匹配进行测试:

发送消息:

创建ConsumerManager类:

订阅消息:

创建ConsumerEnv类:

创建Consumer接口:

为MSGQueue类新增加二个属性和方法,用于管理订阅队列的消费者集合:

在ConsumerManager类中实现添加消费者方法:

十三.网络通信协议设计

设计应用层协议:

创建request类:

创建response类:

创建参数父类:

创建响应父类:

创建设备功能的参数类:

十四.实现BrokerServer

十五.实现客户端

创建ConnectionFactory.

创建Connection类:

创建Channel类:

客户端代码测试:

十六.完成

成果测试:

启动消息队列服务器:

创建生产者发送消息:

创建消费者消费消息:

十.整合数据库和文件数据

上面的代码中,使用数据库存储了Exchange,Queue,Binding,使用文件存储了Message,

下面对数据库和文件中的数据进行整合.进行统一管理.

创建DiskDataManager类

/*** 对数据库中的Exchange,Queue,Binding和文件中的Message数据进行整合* 统一管理,后续上层代码直接调用该类中的方法即可,无需再向下层数据结构调用*/
public class DiskDataManager {private DataBaseManager dataBaseManager = new DataBaseManager();private MessageFileManager messageFileManager = new MessageFileManager();public void init() throws JsonProcessingException {dataBaseManager.init();messageFileManager.init();}//交换机://添加交换机public void insertExchange(Exchange exchange){dataBaseManager.insertExchange(exchange);}//删除交换机public void deleteExchange(String exchangeName){dataBaseManager.deleteExchange(exchangeName);}//查找交换机public List<Exchange> selectAllExchanges(){return dataBaseManager.selectAllExchanges();}//队列//添加队列public void insertQueue(MSGQueue queue) throws IOException, MqException {dataBaseManager.insertQueue(queue);//创建队列后,不仅要将队列写入到数据库中,还要创建出对应的目录和文件messageFileManager.createQueueFile(queue.getName());}//删除队列public void deleteQueue(String queueName) throws IOException {dataBaseManager.deleteQueue(queueName);//删除队列后,还要讲对应的目录和文件删除messageFileManager.destoryQueueFile(queueName);}//查找队列public List<MSGQueue> selectAllQueues(){return dataBaseManager.selectAllQueues();}//绑定关系//添加绑定关系public void insertBinding(Binding binding){dataBaseManager.insertBinding(binding);}//删除绑定关系public void deleteBinding(String bingingKey){dataBaseManager.deleteBindings(bingingKey);}//查找绑定public List<Binding> selectAllBindings(){return dataBaseManager.selectAllBindings();}//消息//发送消息public void sendMessage(MSGQueue queue, Message message) throws IOException, MqException {messageFileManager.sendMessage(queue,message);}//删除消息public void deleteMessageFromQueue(MSGQueue queue,Message message) throws IOException, ClassNotFoundException, MqException {messageFileManager.deleteMessageFromFile(queue,message);//删除消息后,查看是否需要进行GCif(messageFileManager.checkGC(queue.getName())){messageFileManager.GC(queue);}}//加载所有的消息到内存中public List<Message> loadAllMessageFromQueue(String queueName) throws IOException, MqException, ClassNotFoundException {return messageFileManager.loadAllMessage(queueName);}
}

十一.内存结构设计

将数据存储到数据库和文件,是为了实现其持久性,但数据还是要存储在内存上的,这样才能更快的访问到数据.

创建MeneryDataCenter类:

这里通过设计不同的数据集合来存储数据在内存中.

/*** 将数据存储在内存中,创建不同的数据集合来管理* 要管理的数据有:* 交换机* 队列* 绑定关系* 消息* 队列中的消息集合* 待确认消息队列中的消息集合*/
public class MemoryDataCenter {//key:exchangeNameprivate ConcurrentHashMap<String, Exchange> exchangeMap = new ConcurrentHashMap<>();//key:queueNameprivate ConcurrentHashMap<String, MSGQueue> queueMap = new ConcurrentHashMap<>();
//   key1:exchangeName key2:queueNameprivate ConcurrentHashMap<String,ConcurrentHashMap<String, Binding>> bindingsMap = new ConcurrentHashMap<>();//key: messageIdprivate ConcurrentHashMap<String, Message> messageMap = new ConcurrentHashMap<>();
//  key:queueName  List:messageprivate ConcurrentHashMap<String, LinkedList<Message>> queueMessageMap = new ConcurrentHashMap<>();//存储在手动确认模式下,管理待确认的消息和队列,在未收到确认消息时,要先将数据存储到这个数据集合中,
//  key1:queueName   key2:messageIdprivate ConcurrentHashMap<String,ConcurrentHashMap<String,Message>> WaitAckQueueMessageMap = new ConcurrentHashMap<>();
}

实现集合操作:

 //交换机://插入public void insertExchange(Exchange exchange){exchangeMap.put(exchange.getName(),exchange);System.out.println("[MemoryDataCenter] 新增交换机成功 exchangeName:"+exchange.getName());}//删除public void deleteExchange(String exchangeName){exchangeMap.remove(exchangeName);System.out.println("[MemoryDataCenter] 删除交换机成功 exchangeName:"+exchangeName);}//查找public Exchange getExchange(String exchangeName){Exchange exchange = exchangeMap.get(exchangeName);return exchange;}//队列//插入public void insertQueue(MSGQueue queue){queueMap.put(queue.getName(),queue);System.out.println("[MemoryDataCenter] 新增队列成功! queueName: "+queue.getName());}//删除public void deleteQueue(String queueName){queueMap.remove(queueName);System.out.println("[MemoryDataCenter] 队列删除成功! queueName: "+queueName);}//查找public MSGQueue getQueue(String queueName){return queueMap.get(queueName);}//绑定关系://新增public void insertBinding(Binding binding) throws MqException {
//        //绑定关系不存在时,创建一个,存在时,进行覆盖
//        ConcurrentHashMap<String, Binding> bindingMap = bindingsMap.get(binding.getQueueName());
//        if(bindingMap==null){
//            bindingMap = new ConcurrentHashMap<>();
//        }
//        bindingMap.put(binding.getQueueName(),binding);
//        bindingsMap.put(binding.getExchangeName(),bindingMap);//这个方法是ConcurrentMap方法用来判断对应的哈希表是否存在,不存在就执行第二个参数,存在就直接赋值,和上面的逻辑是一样的//且该方法是原子的,不存在线程安全问题ConcurrentHashMap<String, Binding> bindingMap = bindingsMap.computeIfAbsent(binding.getExchangeName(),f -> new ConcurrentHashMap<>());//此处可能会存在线程安全问题,以绑定关系为基准进行上锁synchronized (binding){Binding binding1 = bindingMap.get(binding.getQueueName());//当绑定关系已经存在时,抛出异常,只有新的绑定插入时,才会成功if(binding1!=null){throw new MqException("[MemoryDataCenter] 绑定已存在 exchangeName: "+binding.getExchangeName()+" ,queueName: "+binding.getQueueName());}bindingMap.put(binding.getQueueName(),binding);}System.out.println("[MemoryDataCenter] 新的绑定创建成功! +exchangeName:"+binding.getExchangeName()+" ,queueName: "+binding.getQueueName()+" ,bindingKey: "+binding.getBindingKey());}//删除public void deleteBinding(Binding binding) throws MqException {//先判断交换机是否存在绑定,不存在时,无法删除.抛出异常ConcurrentHashMap<String, Binding> bindingMap = bindingsMap.get(binding.getExchangeName());if(bindingMap==null){throw new MqException("[MemoryDataCenter] 无绑定关系,删除失败 exchangeName: "+binding.getExchangeName()+" ,queueName: "+binding.getQueueName());}bindingMap.remove(binding.getQueueName());System.out.println("[MemoryDataCenter] 绑定删除成功 exchangeName: "+binding.getExchangeName()+" ,queueName: "+binding.getQueueName()+" ,bindingKey:"+binding.getBindingKey());}//查找public Binding getBinding(String exchangeName,String queueName) throws MqException {ConcurrentHashMap<String, Binding> bindingMap = bindingsMap.get(exchangeName);if(bindingMap==null){return null;}Binding binding = bindingMap.get(queueName);return binding;}//消息//插入public void insertMessage(Message message){messageMap.put(message.getMessageId(),message);System.out.println("[MemoryDataCenter] 新增消息成功 messageID:"+message.getMessageId());}//删除public void deleteMessage(String messageId){messageMap.remove(messageId);System.out.println("[MemoryDataCenter] 消息删除陈功 messageId: "+messageId);}//查找public Message getMessage(String messageId){return messageMap.get(messageId);}//队列消息集合//发送消息到指定队列public void sendMessage(MSGQueue queue,Message message){//1.先查找队列对应的集合是否存在,不存在时创建消息集合LinkedList<Message> messages = queueMessageMap.computeIfAbsent(queue.getName(), f -> new LinkedList<>());//这里当多个线程同时执行插入操作时,可能会覆盖消息,要以集合为维度进行上锁synchronized(messages){messages.add(message);}//将消息也存入到消息集合中messageMap.put(message.getMessageId(),message);System.out.println("[MemoryDataCenter] 发送消息到队列成功 queueName:"+queue.getName()+" ,messageId:"+message.getMessageId());}//从队列中取消息public Message pollMessage(String queueName){LinkedList<Message> messages = queueMessageMap.get(queueName);if(messages==null || messages.isEmpty()){return null;}Message message = messages.remove(0);System.out.println("[MemoryDataCenter] 从队列中取消息成功 queueName:"+queueName+" ,messageId:"+message.getMessageId());return message;}//获取队列中的消息个数public int getMessageCountFromQueue(String queueName){LinkedList<Message> messages = queueMessageMap.get(queueName);if(messages==null) return 0;//此处获取集合中元素个数可能存在线程安全问题,对集合进行上锁synchronized(messages){return messages.size();}}//待确认消息集合//发送消息到待确认消息集合public void sendWaitMessage(String queueName,Message message){ConcurrentHashMap<String, Message> waitMessagesMap = queueMessageWaitAckMap.computeIfAbsent(queueName, f -> new ConcurrentHashMap<>());//此处向待确认消息集合中插入数据时,也可能存在线程安全问题,以集合为维度加锁synchronized (waitMessagesMap){waitMessagesMap.put(message.getMessageId(),message);}System.out.println("[MemoryDataCenter] 发送待确认消息到队列成功 queueName:"+queueName+" ,messageId:"+message.getMessageId());}//从队列中取待确认消息public Message pollWaitMessage(String queueName,String messageId){ConcurrentHashMap<String, Message> waitMessagesMap = queueMessageWaitAckMap.get(queueName);if(waitMessagesMap==null){return null;}Message message = waitMessagesMap.get(messageId);if(message==null){return null;}System.out.println("[MemoryDataCenter] 从队列中取代确认消息成功 messageId:"+messageId+" ,queueName:"+queueName);return message;}//从队列中删除待确认消息public void deleteWaitMessage(String queueName,String messageId){ConcurrentHashMap<String, Message> waitMessagesMap = queueMessageWaitAckMap.get(queueName);if(waitMessagesMap==null){System.out.println("[MemoryDataCenter] 待确认消息队列不存在,消息删除失败 messageId:"+messageId+" ,queueName:"+queueName);}waitMessagesMap.remove(messageId);System.out.println("[MemoryDataCenter] 待确认消息删除成功 messageId:"+messageId+" ,queueName:"+queueName);}//恢复所有硬盘中的数据//当服务器重启后,内存中的数据都不存在了,要从磁盘中获取数据public void recovery(DiskDataManager diskDataManager) throws IOException, MqException, ClassNotFoundException {//先将内存中的集合都清空,防止存在残留数据exchangeMap.clear();queueMap.clear();bindingsMap.clear();messageMap.clear();queueMessageMap.clear();queueMessageWaitAckMap.clear();//恢复交换机数据List<Exchange> exchanges = diskDataManager.selectAllExchanges();for(Exchange e:exchanges){String exchangeName = e.getName();exchangeMap.put(exchangeName,e);}//恢复队列数据List<MSGQueue> queues = diskDataManager.selectAllQueues();for(MSGQueue q:queues){String queueName = q.getName();queueMap.put(queueName,q);}//恢复绑定关系List<Binding> bindings = diskDataManager.selectAllBindings();for(Binding b:bindings){ConcurrentHashMap<String, Binding> bindingMap = bindingsMap.computeIfAbsent(b.getExchangeName(), f -> new ConcurrentHashMap<>());bindingMap.put(b.getQueueName(),b);}//恢复消息for(MSGQueue q:queueMap.values()){List<Message> messages = diskDataManager.loadAllMessageFromQueue(q.getName());for(Message m:messages){messageMap.put(m.getMessageId(),m);}}//对于未确认消息,当服务器重启后,服务器中所有的消息都要重新发送,未被确认的消息就都成了未被取走的消息了,//对于未确认的消息, 就不需要回复这些数据了}

对MemoryDataCenter类功能测试:

@SpringBootTest
public class MemoryDataCenter {private MemoryDataCenter memoryDataCenter;@BeforeEachpublic void setUp(){memoryDataCenter = new MemoryDataCenter();System.out.println("前置工作已经准备后!");}@AfterEachpublic void tearDown(){memoryDataCenter = null;System.out.println("收尾工作以完成!");}
}

测试功能:

//测试交换机相关操作private Exchange createExchange(String exchangeName){Exchange exchange = new Exchange();exchange.setName(exchangeName);exchange.setType(ExchangeType.DIRECT);exchange.setDurable(true);exchange.setAutoDelete(false);return exchange;}@Testvoid testExchange(){Exchange exchange = createExchange("exchangeTest");memoryDataCenter.insertExchange(exchange);Exchange act = memoryDataCenter.getExchange(exchange.getName());Assertions.assertEquals(exchange.getName(),act.getName());Assertions.assertEquals(exchange.getType(),act.getType());Assertions.assertEquals(exchange.isDurable(),act.isDurable());Assertions.assertEquals(exchange.isAutoDelete(),act.isAutoDelete());memoryDataCenter.deleteExchange(exchange.getName());act = memoryDataCenter.getExchange(exchange.getName());Assertions.assertNull(act);}//测试队列相关操作private MSGQueue createQueue(String queueName){MSGQueue queue = new MSGQueue();queue.setName(queueName);queue.setDurable(true);queue.setAutoDelete(false);return queue;}@Testvoid testQueue(){MSGQueue queue = createQueue("queueTest");memoryDataCenter.insertQueue(queue);MSGQueue act = memoryDataCenter.getQueue(queue.getName());Assertions.assertEquals(queue.getName(),act.getName());Assertions.assertEquals(queue.isDurable(),act.isDurable());Assertions.assertEquals(queue.isAutoDelete(),act.isAutoDelete());memoryDataCenter.deleteQueue(queue.getName());act  = memoryDataCenter.getQueue(queue.getName());Assertions.assertNull(act);}//测试绑定关系相关操作private Binding createBinding(String exchangeName,String queueName,String bindingKey){Binding binding = new Binding();binding.setExchangeName(exchangeName);binding.setQueueName(queueName);binding.setBindingKey(bindingKey);return binding;}@Testvoid testBinding() throws MqException {//要先创建队列和交换机Exchange exchange = createExchange("exchangeTest");MSGQueue queue = createQueue("queueTest");Binding binding = createBinding(exchange.getName(), queue.getName(), "bindingKeyTest");memoryDataCenter.insertBinding(binding);Binding act = memoryDataCenter.getBinding(exchange.getName(), queue.getName());Assertions.assertEquals(binding.getExchangeName(),act.getExchangeName());Assertions.assertEquals(binding.getQueueName(),act.getQueueName());Assertions.assertEquals(binding.getBindingKey(),act.getBindingKey());memoryDataCenter.deleteBinding(binding);act = memoryDataCenter.getBinding(act.getExchangeName(),act.getQueueName());Assertions.assertNull(act);}//测试消息操作private Message createMessage(String body){Message message = new Message();return message.createMessageById("routingKeyTest",null,body.getBytes());}@Testpublic void testMessage(){Message expectedMessage = createMessage("testMessage");memoryDataCenter.insertMessage(expectedMessage);Message actualMessage = memoryDataCenter.getMessage(expectedMessage.getMessageId());Assertions.assertEquals(expectedMessage,actualMessage);//删除消息memoryDataCenter.deleteMessage(expectedMessage.getMessageId());actualMessage = memoryDataCenter.getMessage(expectedMessage.getMessageId());Assertions.assertNull(actualMessage);}//测试队列中的消息集合@Testvoid testQueueMessage(){MSGQueue queue = createQueue("queueTest");Message message = createMessage("hello");memoryDataCenter.sendMessage(queue,message);int n = memoryDataCenter.getMessageCountFromQueue(queue.getName());Assertions.assertEquals(1,n);Message act = memoryDataCenter.pollMessage(queue.getName());n = memoryDataCenter.getMessageCountFromQueue(queue.getName());Assertions.assertEquals(0,n);Assertions.assertEquals(message.getMessageId(),act.getMessageId());Assertions.assertArrayEquals(message.getBody(),act.getBody());Assertions.assertEquals(message.getIsVaild(),act.getIsVaild());Assertions.assertEquals(message.getDeliveryMode(),act.getDeliveryMode());}//测试待确认队列集合@Testvoid testWaitMessageQueue(){MSGQueue queue = createQueue("queueTest");Message message = createMessage("hello");memoryDataCenter.sendWaitMessage(queue.getName(), message);Message act = memoryDataCenter.pollWaitMessage(queue.getName(), message.getMessageId());Assertions.assertEquals(message.getMessageId(),act.getMessageId());Assertions.assertArrayEquals(message.getBody(),act.getBody());Assertions.assertEquals(message.getIsVaild(),act.getIsVaild());Assertions.assertEquals(message.getDeliveryMode(),act.getDeliveryMode());memoryDataCenter.deleteWaitMessage(queue.getName(), message.getMessageId());act = memoryDataCenter.pollWaitMessage(queue.getName(), message.getMessageId());Assertions.assertNull(act);}//测试加载磁盘所有数据到内存@Testvoid testRecovery() throws IOException, MqException, ClassNotFoundException {//这里需要使用到mybatis,需要进行了类加载,先启动SpringApplicationMq02Application.context = SpringApplication.run(Mq02Application.class);//在磁盘上构造好数据:DiskDataManager diskDataCenter = new DiskDataManager();diskDataCenter.init("");//创建交换机:Exchange exchange = createExchange("testExchange");diskDataCenter.insertExchange(exchange);//创建队列:MSGQueue queue = createQueue("testQueue");diskDataCenter.insertQueue(queue);//创建绑定Binding binding = new Binding();binding.setExchangeName(exchange.getName());binding.setQueueName(queue.getName());binding.setBindingKey("bindingKey");diskDataCenter.insertBinding(binding);//创建消息Message message = createMessage("testContext");diskDataCenter.sendMessage(queue,message);//执行恢复:memoryDataCenter.recovery(diskDataCenter);//结果比对://交换机:Exchange actualExchange = memoryDataCenter.getExchange(exchange.getName());Assertions.assertEquals(exchange.getName(),actualExchange.getName());Assertions.assertEquals(exchange.getType(),actualExchange.getType());Assertions.assertEquals(exchange.isDurable(),actualExchange.isDurable());Assertions.assertEquals(exchange.isAutoDelete(),actualExchange.isAutoDelete());//队列:MSGQueue actualQueue = memoryDataCenter.getQueue(queue.getName());Assertions.assertEquals(queue.getName(),actualQueue.getName());Assertions.assertEquals(queue.isDurable(),actualQueue.isDurable());Assertions.assertEquals(queue.isAutoDelete(),actualQueue.isAutoDelete());//绑定:Binding actulaBinding = memoryDataCenter.getBinding(exchange.getName(), queue.getName());Assertions.assertEquals(binding.getExchangeName(),actulaBinding.getExchangeName());Assertions.assertEquals(binding.getQueueName(),actulaBinding.getQueueName());Assertions.assertEquals(binding.getBindingKey(),actulaBinding.getBindingKey());//消息:Message actualMessage = memoryDataCenter.getMessage(message.getMessageId());Assertions.assertEquals(message.getMessageId(),actualMessage.getMessageId());Assertions.assertEquals(message.getDeliveryMode(),actualMessage.getDeliveryMode());Assertions.assertEquals(message.getRoutingKey(),actualMessage.getRoutingKey());Assertions.assertArrayEquals(message.getBody(),actualMessage.getBody());//   清除文件//清理之前要先关闭文件Mq02Application.context.close();File file = new File("./data");FileUtils.deleteDirectory(file);}

十二.整合内存和磁盘数据

将内存和磁盘上的数据进行整合,用"虚拟机"这个概念将其整合起来. 不同虚拟机中的交换机队列,绑定关系,消息都是不互通的. 此处为了简化,仅实现单台虚拟主机,但在数据结构上设置不同虚拟主句名
为区分不同的虚拟主机上的设备,通过配置设备名区别:(以虚拟机名为前缀)
规定:
* exchangeName = virtualHostName+exchangeName;
* queueName = virtualHostName+queueName;
并且将调用的方法抛出的异常都在这个类中进行处理,不再向上抛出

创建VirtualHost类:

@Data
public class VirtualHost {private String virtualHostName;private MemoryDataCenter memoryDataCenter = new MemoryDataCenter();private DiskDataManager diskDataManager = new DiskDataManager();public VirtualHost(String virtualHostName){this.virtualHostName = virtualHostName;//初始化磁盘数据:diskDataManager.init();//初始化内存数据try {memoryDataCenter.recovery(diskDataManager);} catch (IOException | MqException | ClassNotFoundException e) {System.out.println("[VirtualHost] 内存数据恢复失败");e.printStackTrace();}}
}

Exchange的声明和删除:

 //在对交换机在内存和磁盘上插入和删除数据时,可能存在线程安全问题,要以交换机为维度对其上锁//交换机锁对象:private final Object exchangeLocker = new Object();//交换机操作://创建交换机,//创建后,将其保存到内存和磁盘上public boolean exchangeDeclare(String exchangeName, ExchangeType type, boolean durable,boolean autoDelete, Map<String,Object> args){//   先根据约定 设置交换机名exchangeName = virtualHostName + exchangeName;synchronized(exchangeLocker){//先在内存上查找,若已存在,则直接返回Exchange exchange = memoryDataCenter.getExchange(exchangeName);if(exchange!=null){System.out.println("[VirtualHost] 交换机已经存在,不再创建 exchangeName:"+exchangeName);return true;}exchange = new Exchange();exchange.setName(exchangeName);exchange.setType(type);exchange.setDurable(durable);exchange.setAutoDelete(autoDelete);//这里对args参数的设置.要在Exchange类中再为args关于Map参数添加set和get方法exchange.setArgs(args);//先存入数据库,再存入内存中,//这个顺序是:插入数据库操作比较容易出现异常,存内存出现异常的可能小较小//         若插入数据库失败,则不再存入内存中;//         若是转换顺序,当存数据库出现异常时,还要将内存中的数据再删了,比较麻烦if(durable){//当交换机设置为持久化时,将其存入内存:diskDataManager.insertExchange(exchange);}//存入内存memoryDataCenter.insertExchange(exchange);System.out.println("[VirtualHost] 交换机创建成功 exchangeName:"+exchangeName);return true;}}//删除交换机//在内存和磁盘上将数据删除public boolean exchangeDelete(String exchangeName){exchangeName = virtualHostName + exchangeName;try {synchronized (exchangeLocker){Exchange exchange = memoryDataCenter.getExchange(exchangeName);if(exchange==null){throw new MqException("[VirtualHost] 要删除的交换机不存在 exchangeName:"+exchangeName);}//删除内存数据:memoryDataCenter.deleteExchange(exchangeName);//删除磁盘数据:boolean durable = exchange.isDurable();if(durable){diskDataManager.deleteExchange(exchangeName);}System.out.println("[VirtualHost] 交换机删除成功 exchangeName:"+exchangeName);return true;}} catch (MqException e) {System.out.println("[VirtualHost] 交换机删除失败 exchangeName:"+exchangeName);e.printStackTrace();}return false;}

在Exchange类中关于args属性上,再增加关于Map参数类型的set方法:

 public void setArgs(Map<String,Object> args){this.args = args;}

MSGQueue的声明和删除:

//在对队列在内存和磁盘上插入和删除数据时,可能存在线程安全问题,要以队列为维度对其上锁//创建 队列锁对象:private final Object queueLocker = new Object();/**队列* 创建队列:创建队列并将其存入到磁盘和内存中*/public boolean queueDeclare(String queueName,boolean isDurable,boolean autoDelete, Map<String,Object> args){queueName = virtualHostName+queueName;try {synchronized(queueLocker){MSGQueue existsQueue = memoryDataCenter.getQueue(queueName);if(existsQueue!=null){System.out.println("[VirtualHost] 队列已经存在 queueName:"+queueName);return true;}MSGQueue queue = new MSGQueue();queue.setName(queueName);queue.setDurable(isDurable);queue.setAutoDelete(autoDelete);//此处在MSGQueue类中,针对args属性,要实现关于Map类型的set方法queue.setArgs(args);//存入磁盘if(isDurable) {diskDataManager.insertQueue(queue);}//存入内存memoryDataCenter.insertQueue(queue);System.out.println("[VirtualHost] 创建队列成功 !");}return true;} catch (IOException | MqException e) {System.out.println("[VirtualHost] 创建队列失败 queueName:"+queueName);e.printStackTrace();return false;}}//删除队列:从磁盘和内存中 删除队列public boolean queueDelete(String queueName){queueName = virtualHostName+queueName;try{synchronized(queueLocker){MSGQueue existsQueue = memoryDataCenter.getQueue(queueName);if(existsQueue==null){throw new MqException("[VirtualHost] 队列不存在,删除队列失败 queueName:"+queueName);}if(existsQueue.isDurable()){diskDataManager.deleteQueue(queueName);}memoryDataCenter.deleteQueue(queueName);System.out.println("[VirtualHost] 删除队列成功 queueName:"+queueName);}return true;}catch (Exception e) {System.out.println("[VirtualHost] 队列删除失败! queueName:" +queueName);e.printStackTrace();return false;}}

同样,在MSGQueue类中关于args属性上,再增加关于Map参数类型的set方法:

 public void setArgs(Map<String,Object> args){this.args = args;}

Binding的创建和删除:


//  该类实现和绑定相关的操作private Router router = new Router();//绑定的插入和删除//插入绑定public boolean bindingDeclare(String exchangeName,String queueName,String bindingKey){exchangeName = virtualHostName + exchangeName;queueName = virtualHostName + queueName;try{//1.验证绑定是否存在,不存在再创建Binding binding = memoryDataCenter.getBinding(exchangeName, queueName);if(binding!=null){throw new MqException("[VirtualHost] 绑定已存在 exchangeName:"+exchangeName+" ,queueName:"+queueName);}//这里再创建一个类router,实现关于绑定相关的操作//2.判断bindingKey格式是否正确boolean ok = router.checkBindingKey(bindingKey);if(!ok){throw new MqException("[VirtualHost] 绑定格式有误 bindingKey:"+bindingKey);}//3.创建绑binding = new Binding();binding.setExchangeName(exchangeName);binding.setQueueName(queueName);binding.setBindingKey(bindingKey);//4.验证绑定的队列和交换机是否存在MSGQueue queue = memoryDataCenter.getQueue(queueName);if(queue==null){throw new MqException("[VirtualHost] 要绑定的队列不存在 queueName:"+queueName+" ,bindingKey:"+bindingKey);}Exchange exchange = memoryDataCenter.getExchange(exchangeName);if(exchange==null){throw new MqException("[VirtualHost] 要绑定的交换机不存在 exchangeName:"+exchangeName+" ,bindingKey:"+bindingKey);}//5,存入磁盘//当队列和交换机同时设置持久化时,将该绑定关系存入磁盘if(queue.isDurable() && exchange.isDurable()){diskDataManager.insertBinding(binding);}//6.存入内存memoryDataCenter.insertBinding(binding);System.out.println("[VirtualHost] 创建绑定成功 bindingKey: "+bindingKey);return true;}catch (MqException e) {System.out.println("[VirtualHost] 创建绑定失败 bindingKey:"+bindingKey);e.printStackTrace();}return false;}//删除绑定public boolean bindingDelete(String exchangeName,String queueName){exchangeName = virtualHostName + exchangeName;queueName = virtualHostName + queueName;try {Binding binding = memoryDataCenter.getBinding(exchangeName, queueName);if(binding==null){throw new MqException("[VirtualHost] 绑定不存在 queueName:"+queueName+" ,exchangeName:"+exchangeName);}//从内存删除memoryDataCenter.deleteBinding(binding);//从磁盘删除//此处可能绑定没有保存在磁盘上,删除失败,但没有关系,没有影响diskDataManager.deleteBinding(binding.getBindingKey());System.out.println("[VirtualHost] 删除绑定成功 exchangeName:"+exchangeName+" ,queueName:"+queueName +" , bindingKey:"+binding.getBindingKey());return true;}catch(Exception e){System.out.println("[VirtualHost] 删除绑定失败 exchangeName:"+exchangeName+" ,queueName:"+queueName);e.printStackTrace();return false;}}

创建Router类:

实现匹配判断功能:

/*** 该类实现和绑定相关的操作//路由规定://routingKey: 只能由 数字 字母(大小写) 下划线 构成,使用.作为分割//bindingKey:只能包含 数字 字母 下划线 * #,以 . 作为分割,* #只能作为独立的分段*/
public class Router {//判断消息携带的绑定格式是否正确public boolean checkRoutingKey(String routingKey){char[] ch = routingKey.toCharArray();for(char i:ch){if(i>='a' && i<='z') continue;if(i>='A' && i<='Z') continue;if(i>='0' && i<='9') continue;if(i=='.' || i=='_') continue;else return false;}return true;}//判断绑定格式是否正确public boolean checkBindingKey(String bindingKey){char[] ch = bindingKey.toCharArray();for(char c:ch){if(c>='A' && c<='Z') continue;if(c>='a' && c<='z') continue;if(c>='0' && c<='9') continue;if(c=='_' || c=='*' || c=='#' || c=='.') continue;else return false;}//规定不能让* #相连,即出现以下情况规定不成立:// *.#   #.*  #.#//以 . 对字符串进行分隔,判断String[] s = bindingKey.split("\\.");for(int i=0;i<s.length-1;i++){if (s[i].equals("*") && s[i+1].equals("#") ||s[i].equals("#") && s[i+1].equals("*") ||s[i].equals("#") && s[i+1].equals("#")) {return false;}}return true;}//判断bindingKey与routingKey是否匹配成功public boolean isRouting(ExchangeType type,String routingKey,String bindingKey) throws MqException {//判断当前交换机类型:fanout/topicif(type==ExchangeType.FANOUT){//匹配到绑定交换机的所有队列//直接返回即可return true;}else if(type==ExchangeType.TOPIC){//进行routingKey和BindingKey的匹配判断return routingTopic(routingKey,bindingKey);}else{throw new MqException("[Router] 交换机类型有误 type:"+type);}}/*** 规定:rotingKey匹配bindingKey*  *:匹配任意单个字符串*  #:匹配任意个任意字符串* @param routingKey 消息携带的匹配字符串* @param bindingKey 交换机和队列的绑定关系* @return*/private boolean routingTopic(String routingKey, String bindingKey) {String[] b = bindingKey.split("\\.");String[] r = routingKey.split("\\.");int n1 = b.length;int n2 = r.length;int i = 0;int j = 0;while(i<n1 && j<n2){if(b[i].equals("*")){//可以匹配routingKey的任意单个字符//直接向后走就行:i++;j++;}else if(b[i].equals("#")){//匹配routingKey的任意个任意字符if(i==n1-1){//当bindingKey的最后一个字符为#时,可以匹配routingKey后面的所有字符串,直接返回true即可:return true;}else{//当b的#不是最后一个字符时,就找r之后的字符串中是否有b的下一个字符串的下标,当找不到时,就返回-1:i++;j = checkNext(b[i],r,j);//当在r中找不到b的下一个字符串时,一定匹配失败,直接返回if(j==-1) return false;else{i++;j++;}}}else{//b为普通字符串时if(!b[i].equals(r[j])) return false;else {i++;j++;}}}//b / r有一个已经匹配到结尾了,只有两个都完全匹配完,才算匹配成功if(i!=n1 || j!=n2){return false;}return true;}private int checkNext(String next, String[] r, int j) {for(int k=j;k<r.length;k++){if(r[k].equals(next)) return k;}return -1;}}

对Router类的TOPIC匹配进行测试:

@SpringBootTest
public class RouterTopicTest {private Router router = new Router();// [测试用例]// binding key          routing key         result// aaa                  aaa                 true// aaa.bbb              aaa.bbb             true// aaa.bbb              aaa.bbb.ccc         false// aaa.bbb              aaa.ccc             false// aaa.bbb.ccc          aaa.bbb.ccc         true// aaa.*                aaa.bbb             true// aaa.*.bbb            aaa.bbb.ccc         false// *.aaa.bbb            aaa.bbb             false// #                    aaa.bbb.ccc         true// aaa.#                aaa.bbb             true// aaa.#                aaa.bbb.ccc         true// aaa.#.ccc            aaa.ccc             true// aaa.#.ccc            aaa.bbb.ccc         true// aaa.#.ccc            aaa.aaa.bbb.ccc     true// #.ccc                ccc                 true// #.ccc                aaa.bbb.ccc         true@Testvoid test01() throws MqException {boolean ok = router.isRouting(ExchangeType.TOPIC, "aaa", "aaa");Assertions.assertTrue(ok);}@Testvoid test02() throws MqException {boolean ok = router.isRouting(ExchangeType.TOPIC,"aaa.bbb", "aaa.bbb");Assertions.assertTrue(ok);}@Testvoid test03() throws MqException {boolean ok = router.isRouting(ExchangeType.TOPIC,"aaa.bbb.ccc", "aaa.bbb");Assertions.assertFalse(ok);}@Testvoid test04() throws MqException {boolean ok = router.isRouting(ExchangeType.TOPIC,"aaa.ccc", "aaa.bbb");Assertions.assertFalse(ok);}@Testvoid test05() throws MqException {boolean ok = router.isRouting(ExchangeType.TOPIC,"aaa.bb.cc", "aaa.bb.cc");Assertions.assertTrue(ok);}@Testvoid test06() throws MqException {boolean ok = router.isRouting(ExchangeType.TOPIC,"aaa.bb", "aaa.*");Assertions.assertTrue(ok);}@Testvoid test07() throws MqException {boolean ok = router.isRouting(ExchangeType.TOPIC,"aaa.bb.cc", "aaa.*.bb");Assertions.assertFalse(ok);}@Testvoid test08() throws MqException {boolean ok = router.isRouting(ExchangeType.TOPIC,"aaa.bb", "*.aaa.bb");Assertions.assertFalse(ok);}@Testvoid test09() throws MqException {boolean ok = router.isRouting(ExchangeType.TOPIC,"aaa.bb.cc", "#");Assertions.assertTrue(ok);}@Testvoid test10() throws MqException {boolean ok = router.isRouting(ExchangeType.TOPIC,"aaa.bb", "aaa.#");Assertions.assertTrue(ok);}@Testvoid test11() throws MqException {boolean ok = router.isRouting(ExchangeType.TOPIC,"aaa.bb.cc", "aaa.#");Assertions.assertTrue(ok);}@Testvoid test12() throws MqException {boolean ok = router.isRouting(ExchangeType.TOPIC,"aaa.cc", "aaa.#.cc");Assertions.assertTrue(ok);}@Testvoid test13() throws MqException {boolean ok = router.isRouting(ExchangeType.TOPIC,"aaa.bb.cc", "aaa.#.cc");Assertions.assertTrue(ok);}@Testvoid test14() throws MqException {boolean ok = router.isRouting(ExchangeType.TOPIC,"aaa.aaa.bb.cc", "aaa.#.cc");Assertions.assertTrue(ok);}@Testvoid test15() throws MqException {boolean ok = router.isRouting(ExchangeType.TOPIC,"cc", "#.cc");Assertions.assertTrue(ok);}    @Testvoid test16() throws MqException {boolean ok = router.isRouting(ExchangeType.TOPIC,"aaa.bb.cc", "#.cc");Assertions.assertTrue(ok);}

发送消息:

//发送消息到队列public boolean basicPublish(String exchangeName,String routingKey,BasicProperties basicProperties,byte[] body){exchangeName = virtualHostName + exchangeName;try {//1.判断交换机是否存在Exchange exchange = memoryDataCenter.getExchange(exchangeName);if(exchange==null){throw new MqException("[VirtualHost] 交换机不存在 exchangeName:"+exchangeName);}//2.判断routingKey格式是否正确boolean ok = router.checkRoutingKey(routingKey);if(!ok) {throw new MqException("[VirtualHost] routingKey格式有误 routingKey:"+routingKey);}//3.根据交换机的类型进行路由匹配,分发消息if(exchange.getType()==ExchangeType.DIRECT){//直接交换机,routingKey就是队列名,bindingKey无用,将消息路由到指定的队列上//获取到指定队列String queueName = virtualHostName + routingKey;MSGQueue queue = memoryDataCenter.getQueue(queueName);if(queue==null){throw new MqException("[VirtualHost] 队列不存在 queueName:"+queueName);}//构造消息对象Message message = new Message();message = message.createMessageById(null,basicProperties,body);//发送消息到队列,再构造一个方法实现sendMessage(queue,message);}else{//当交换机类型为fanout/topic时://遍历交换机所有的绑定ConcurrentHashMap<String, Binding> bindings = memoryDataCenter.getBindings(exchangeName);for(Binding b:bindings.values()){MSGQueue queue = memoryDataCenter.getQueue(b.getQueueName());//判断交换机绑定的队列是否存在:if(queue==null){System.out.println("[VirtualHost] 队列不存在 queueName:"+b.getQueueName());continue;}//构造消息对象Message message = new Message().createMessageById(routingKey, basicProperties, body);//判断routingKey与binding是否成功if(!router.isRouting(exchange.getType(),message.getRoutingKey(),b.getBindingKey())){//匹配失败:System.out.println("[VirtualHost] routingKey和BindingKey不匹配 routingKey:"+routingKey+" , bindingKey:"+b.getBindingKey());continue;}//匹配成功时,就将消息转发sendMessage(queue,message);System.out.println("[VirtualHost] 消息发送成功 queueName:"+queue.getName()+" ,messageId:"+message.getMessageId());}}return true;}catch (Exception e){System.out.println("[VirtualHost]消息发送失败 ");e.printStackTrace();return false;}}//消费者管理对象:private ConsumerManager consumerManager = new ConsumerManager(this);private void sendMessage(MSGQueue queue, Message message) throws IOException, MqException, InterruptedException {//存入磁盘//是否持久化//1:持久化 0:非持久化if(message.getDeliveryMode()==1){diskDataManager.sendMessage(queue,message);}//存入内存:memoryDataCenter.sendMessage(queue,message);//消息已经到达队列,通知订阅队列的消费者消费消息consumerManager.notifyConsumer(queue.getName());System.out.println("[VirtualHost] 发送消息成功");}

创建ConsumerManager类:

对消费者进行管理:


/*** 消费者管理类*/
public class ConsumerManager {//持有上层的VirtualHost对象的引用,用来操作数据private VirtualHost virtualHost;//    使⽤⼀个线程池⽤来执⾏消息回调private ExecutorService workerPool = Executors.newFixedThreadPool(4);//存放令牌(队列名)的队列:那个队列当前有消息了,就将队列名加入到阻塞队列中//然后扫描线程通过该队列中存放的队列名找到对应的消息和订阅者,将信息打包放到线程池中进行消费private BlockingQueue<String> tokenQueue = new LinkedBlockingQueue<>();//扫描线程private Thread scannerThread = null;//通知消费者消费消息://调用时机:发送方发送消息成功后,//当队列中有消息了,就将其放到阻塞队列中,然后就要通知消费者消费消息了public void notifyConsumer(String queueName) throws InterruptedException {tokenQueue.put(queueName);}
}

订阅消息:

    /**//订阅消息//添加一个订阅者:* @param consumerTag 消费者身份标识* @param queueName 队列名* @param autoAck 是否自动确认消息* @param consumer 回调函数* @return*/public boolean basicConsume(String consumerTag, String queueName,boolean autoAck, Consumer consumer){queueName = virtualHostName + queueName;try {//通过消费者管理类实现添加消费者功能consumerManager.addConsumer(consumerTag,queueName,autoAck,consumer);System.out.println("[VirtualHost] basicConsumer 成功 queueName:"+queueName);return true;}  catch (MqException e) {System.out.println("[VirtualHost] basicConsumer 失败 queueName:"+queueName);e.printStackTrace();return false;}}

创建ConsumerEnv类:

消费者完整环境类:


/*** 表示消费者(完整的执行环境)*/
@Data
public class ConsumerEnv {//消费者唯一标识private String consumerTag;//订阅队列的队列名字private String queueName;//是否自动确认消息private boolean autoAck;//要执行的具体功能,通过一个接口,由调用者自己实现其方法体private Consumer consumer;
}

创建Consumer接口:

实现消费者的回调函数接口:通过lambda表达式,让消费者自己实现对消息的处理

/*** 函数式接口,回调函数,当消费者收到消息后,要处理消息,调用者通过这个接口实现具体的功能*/
@FunctionalInterface
public interface Consumer {//deliver:投递的意思,这个方法在每次服务器收到发送来的消息后,调用//通过这个方法把消息推送给对应的消费者void handlerDeliver(String consumerTag, BasicProperties basicProperties, byte[] body) throws MqException, IOException;
}

为MSGQueue类新增加二个属性和方法,用于管理订阅队列的消费者集合:

//此处再添加一个属性:订阅该队列的消费者集合private List<ConsumerEnv> consumerEnvList = new ArrayList<>();//当订阅队列的消费者不止一个时 , 规定以轮训的方式消费消息//再添加一个属性,记录当前轮到哪个消费者消费消息了//这里使用AtomicInteger类来实现,目的是不让手动修改,且要实现自增的功能private AtomicInteger atomicInteger = new AtomicInteger(0);//添加一个新的订阅者(消费者)public void addConsumerEnv(ConsumerEnv consumerEnv){consumerEnvList.add(consumerEnv);}//挑选一个订阅者,消费当前消息,按照轮训的方式public ConsumerEnv chooseConsumerEnv(){if(consumerEnvList.isEmpty()){//当前该队列还没有消费者订阅System.out.println("[MSGQueue] 当前该队列没有订阅者");return null;}//按照轮训的方式获取一个要消费消息的订阅者下标int index = atomicInteger.get()%consumerEnvList.size();//让轮训值 自增atomicInteger.getAndIncrement();return consumerEnvList.get(index);}

在ConsumerManager类中实现添加消费者方法:

 //添加新的消费者,并消费队列中当前存在的消息public void addConsumer(String consumerTag, String queueName, boolean autoAck, Consumer consumer) throws MqException {//1.找到对应的队列MSGQueue queue = virtualHost.getMemoryDataCenter().getQueue(queueName);if(queue==null){throw new MqException("[ConsumerManager] 队列不存在 queueName:"+queueName);}//2.创建一个消费者ConsumerEnv consumerEnv = new ConsumerEnv();consumerEnv.setConsumerTag(consumerTag);consumerEnv.setQueueName(queueName);consumerEnv.setAutoAck(autoAck);consumerEnv.setConsumer(consumer);//3.将订阅者加入到队列的订阅者队列中queue.addConsumerEnv(consumerEnv);//4.当队列中已经有一些消息时,要将其消费掉synchronized (queue){int n = virtualHost.getMemoryDataCenter().getMessageCountFromQueue(queueName);for(int i=0;i<n;i++){//这个方法调用一次就消费一条消息consumerMessage(queue);}}}//  消费消息:调用消息的回调函数,并将消息从队列中删除//从队列中获取一个消息,并让消费者消费,// 当消费者不止一个时,按照轮训的方式让消费者依次消费消息private void consumerMessage(MSGQueue queue) throws MqException {//1.从队列的订阅者中挑选一个订阅者ConsumerEnv consumerEnv = queue.chooseConsumerEnv();if(consumerEnv==null){//当前队列号没有订阅者,无法消费消息System.out.println("[ConsumerManager] 当前队列中还没有订阅者");return;}//2.消费消息Message message = virtualHost.getMemoryDataCenter().pollMessage(queue.getName());if(message==null){//当前队列中还没有消息,不需要消费System.out.println("当前队列中还没有消息");return;}//将消息带到消费者的回调方法中,给线程池执行workerPool.submit(()->{try{//1.在执行回调之前,先将消息放到待确认队列集合中,一旦消息被消费失败了.就重新发送消息virtualHost.getMemoryDataCenter().sendWaitMessage(queue.getName(),message);//2.执行订阅者的回调方法consumerEnv.getConsumer().handlerDeliver(consumerEnv.getConsumerTag(),message.getBasicProperties(),message.getBody());//3.根据消费者的确认消息方式及消费者消费消息的情况,执行删除消息操作//  这里完成为自动确认模式下的操作,手动模式下,在basicAck方法中实现if(consumerEnv.isAutoAck()){//4.删除磁盘中的数据//  是否持久化//  1:非持久化 0:持久化if(message.getDeliveryMode()==0) {virtualHost.getDiskDataManager().deleteMessageFromQueue(queue, message);}//5.删除未确认消息队列中的消息virtualHost.getMemoryDataCenter().deleteWaitMessage(queue.getName(), message.getMessageId());//6.删除消息集合中的消息virtualHost.getMemoryDataCenter().deleteMessage(message.getMessageId());System.out.println("[ConsumerManager] 消息被成功消费 ");}}catch (Exception e){System.out.println("[ConsumerManager] 消费消息失败");e.printStackTrace();}});}

在ConsumerManager类中,添加扫描线程,不停扫描阻塞令牌队列,查看是否有新的消息到来,需要消费者及时消费:

//先获取到令牌,根据令牌找到指定的队列,从队列中获取消息进行消费public ConsumerManager(VirtualHost parent){virtualHost = parent;//为推的模式.不断的扫描令牌队列,一但有消息进入队列,就将其推送给消费者Thread t = new Thread(()->{while(true){try {//1.获取令牌String queueName = tokenQueue.take();//2.根据令牌,找到指定的队列MSGQueue queue = virtualHost.getMemoryDataCenter().getQueue(queueName);if(queue==null){throw new MqException("[ConsumerManager] 获取令牌时,发现队列不存在");}synchronized (queue){//3.从队列中获取一个消息并进行消费consumerMessage(queue);}} catch (InterruptedException | MqException e) {throw new RuntimeException(e);}}});//将线程设为后台线程//当前台线程执行结束了,后台线程也就结束了,//若设为前台线程,那么只有当前台线程执行完了,整个进程才会结束,// 这里的循环是while(true)会一直卡着执行结束不了,因此要设成后台线程t.setDaemon(true);//启动线程t.start();}

十三.网络通信协议设计

生产者和消费者都是客户端,需要通过网络和消息队列服务器进行通信.

此处我们使⽤TCP协议,来作为通信的底层协议.同时在这个基础上⾃定义应⽤层协议,完成客⼾端对服务器这边功能的远程调⽤.

设计应用层协议:

使⽤⼆进制的⽅式设定协议.

请求数据格式:

响应数据格式:

其中 type 表⽰请求响应不同的功能. 取值如下:

• 0x1 创建 channel

• 0x2 关闭 channel

• 0x3 创建 exchange

• 0x4 销毁 exchange

• 0x5 创建 queue

• 0x6 销毁 queue

• 0x7 创建 binding

• 0x8 销毁 binding

• 0x9 发送 message

• 0xa 订阅 message

• 0xb 返回 ack

• 0xc 服务器给客⼾端推送的消息. (被订阅的消息) 响应独有的

对于请求来说,payload是各种请求方法的参数信息

对响应来说,payload是方法的返回数据信息.

创建request类:


/*** 表示一个网络通信中的请求对象*/
@Data
public class Request {/** type 表⽰请求响应不同的功能. 取值如下*  0x1  创建 channel* • 0x2  关闭 channel* • 0x3  创建 exchange* • 0x4  销毁 exchange* • 0x5  创建 queue* • 0x6  销毁 queue* • 0x7  创建 binding* • 0x8  销毁 binding* • 0x9  发送 message* • 0xa  订阅 message* • 0xb  返回 ack* • 0xc  服务器给客⼾端推送的消息. (被订阅的消息) 响应独有的*///请求类型,设定占4字节private int type;//请求的数据长度,占4字节private int length;//请求体 payload 表⽰这次⽅法调⽤的各种参数信息private byte[] payload;
}

创建response类:

/*** 表示一个响应对象*/
@Data
public class Response {//按照自己的定义,响应类型,4字节private int type;//响应的数据长度,4字节private int length;//响应体private byte[] payload;
}

创建参数父类:

//定义参数⽗类
//构造⼀个类表⽰⽅法的参数, 作为 Request 的 payload.
//不同的⽅法中, 参数形态各异, 但是有些信息是通⽤的, 使⽤⼀个⽗类表⽰出来. 具体每个⽅法的参数再
//通过继承的⽅式体现
@Data
public class BasicArgs implements Serializable {//表示一次请求的身份标识,用来和该请求 对应的返回的响应相对照protected String rid;//每一次请求需要建立连接,通过TCP建立连接,一个连接可以发送多次消息,每条消息通过信道传送//一条信道可以发送多条消息//这次通信的信道channel的身份标识protected String channelId;
}

创建响应父类:

/*** 定义payload的返回数据*/
@Data
public class BasicReturns implements Serializable {//一次请求或相应的身份标识protected String rid;//标识一个channelprotected String channelId;//表示方法的执行结果  payload 表⽰这次⽅法调⽤的返回值.protected boolean ok;
}

创建设备功能的参数类:

exchangeDeclareArgs:

/*** 这个类表示调用声明交换机方法的参数*/
@Data
public class ExchangeDeclareArgs extends BasicArgs implements Serializable {private String exchangeName;private ExchangeType type;private boolean isDurable;private boolean autoDelete;private Map<String,Object> args;
}

exchangeDeleteArgs:

@Data
public class ExchangeDeleteArgs extends BasicArgs implements Serializable {private String exchangeName;
}

queueDeclareArgs:

@Data
public class QueueDeclareArgs extends BasicArgs implements Serializable {private String queueName;private boolean isDurable;private boolean autoDelete;private Map<String,Object> args;
}

queueDeleteArgs:

@Data
public class QueueDeleteArgs extends BasicArgs implements Serializable {private String queueName;
}

bindingDeclareArgs:

@Data
public class BindingDeclareArgs extends BasicArgs implements Serializable {private String ExchangeName;private String queueName;private String bindingKey;
}

bindingDeleteArgs:

@Data
public class BindingDeleteArgs extends BasicArgs implements Serializable {private String exchangeName;private String queueName;
}

basicPublishArgs:

@Data
public class BasicPublishArgs extends BasicArgs implements Serializable {private String exchangeName;private String routingKey;private BasicProperties basicProperties;private byte[] body;
}

basicConsumerArgs:

@Data
public class BasicConsumerArgs extends BasicArgs implements Serializable {private String consumerTag;private String queueName;private boolean autoAck;//这个类对应的BasicConsumer方法还有一个参数 consumer,是一个回到参数//消费者客户端收到服务器发送的消息后,针对自己的业务,实现这个回调接口就行了,//无需再将回调参数传给服务器,因此解救不需要在这里写这个参数了//并且,这个 回调参数也无法通过网络传输给服务器
}

basicAckArgs:

/*** 手动响应数据*/
@Data
public class BasicAckArgs extends BasicArgs implements Serializable {private String queueName;private String messageId;
}

subscribeReturns:

/*** 这里类表示返回数据的具体参数* 是服务器给消费者提供的订阅消息* consumerTag其实是channelId.* basicProperties和body共同构成了Message.*/
@Data
public class SubScribeReturns extends BasicReturns implements Serializable {private String consumerTag;private BasicProperties basicProperties;private byte[] body;
}