RabbitMinQ(模擬實現消息隊列項目)02

十.整合數據庫和文件數據

創建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;
}