Java项目--仿RabbitMQ的消息队列--网络通讯协议设计

目次

一、引言
二、设计
三、代码
1.Request
2.Response
3.BasicArguments
4.BasicReturns
四、方法类
1.创建交换机
2.删除交换机 
3.创建队列 
4.删除队列 
 5.创建绑定
6.删除绑定 
7.消息发布 
8.消费消息 
9.会合返回 
五、实现Broker Server类
六、实现连接
  1.connectionFactory类
2.connection类
3.channel类
七、总结

---

一、引言

  本篇文章就介绍一下本次项目的末了一个大的部分，网络通讯协议的设计。
二、设计

  生产者和消费者都是客户端，都需要通过网络和Broker Server举行通讯。

  此处使用TCP协议，来作为通讯的底层协议，同时在这个基础上自定义应用层协议，完成客户端对服务器这边功能的远程调用。
  

 

三、代码

1.Request

1. public class Request {
2.     private int type;
3.     private int length;
4.     private byte[] payload;
6.     public int getType() {
7.         return type;
8.     }
10.     public void setType(int type) {
11.         this.type = type;
12.     }
14.     public int getLength() {
15.         return length;
16.     }
18.     public void setLength(int length) {
19.         this.length = length;
20.     }
22.     public byte[] getPayload() {
23.         return payload;
24.     }
26.     public void setPayload(byte[] payload) {
27.         this.payload = payload;
28.     }
29. }

复制代码

2.Response

1. public class Response {
2.     private int type;
3.     private int length;
4.     private byte[] payload;
6.     public int getType() {
7.         return type;
8.     }
10.     public void setType(int type) {
11.         this.type = type;
12.     }
14.     public int getLength() {
15.         return length;
16.     }
18.     public void setLength(int length) {
19.         this.length = length;
20.     }
22.     public byte[] getPayload() {
23.         return payload;
24.     }
26.     public void setPayload(byte[] payload) {
27.         this.payload = payload;
28.     }
29. }

复制代码

3.BasicArguments

1. /*
2. Request的payload
3. */
4. public class BasicArguments implements Serializable {
5.     protected String rid;
6.     protected String channelId;
8.     public String getRid() {
9.         return rid;
10.     }
12.     public void setRid(String rid) {
13.         this.rid = rid;
14.     }
16.     public String getChannelId() {
17.         return channelId;
18.     }
20.     public void setChannelId(String channelId) {
21.         this.channelId = channelId;
22.     }
23. }

复制代码

4.BasicReturns

1. /*
2. Response的payload
3. */
4. public class BasicReturns implements Serializable {
5.     protected String rid;
6.     protected String channelId;
7.     protected boolean ok;
9.     public String getRid() {
10.         return rid;
11.     }
13.     public void setRid(String rid) {
14.         this.rid = rid;
15.     }
17.     public String getChannelId() {
18.         return channelId;
19.     }
21.     public void setChannelId(String channelId) {
22.         this.channelId = channelId;
23.     }
25.     public boolean isOk() {
26.         return ok;
27.     }
29.     public void setOk(boolean ok) {
30.         this.ok = ok;
31.     }
32. }

复制代码

四、方法类

对于每个VirtualHost提供的方法都要有一个类表现对应的参数
1.创建交换机

1. public class ExchangeDeclareArguments extends BasicArguments implements Serializable {
2.     private String exchangeName;
3.     private ExchangeType exchangeType;
4.     private boolean durable;
5.     private boolean autoDelete;
6.     private Map<String,Object> arguments;
8.    
9. }

复制代码

2.删除交换机 

1. public class ExchangeDeleteArguments extends BasicArguments implements Serializable {
2.     private String exchangeName;
3. }

复制代码

3.创建队列 

1. public class QueueDeclareArguments extends BasicArguments implements Serializable {
2.     private String queueName;
3.     private boolean durable;
4.     private boolean exclusive;
5.     private boolean autoDelete;
6.     private Map<String,Object> arguments;
7. }

复制代码

4.删除队列 

1. public class QueueDeleteArguments extends BasicArguments implements Serializable {
2.     private String queueName;
3. }

复制代码

 5.创建绑定

1. public class QueueBindArguments extends BasicArguments implements Serializable {
2.     private String exchangeName;
3.     private String queueName;
4.     private String bindingKey;
5. }

复制代码

6.删除绑定 

1. public class QueueUnBindArguments extends BasicArguments implements Serializable {
2.     private String exchangeName;
3.     private String queueName;
4. }

复制代码

7.消息发布 

1. public class BasicPublishArguments extends BasicArguments implements Serializable {
2.     private String exchangeName;
3.     private String routingKey;
4.     private BasicProperties basicProperties;
5.     private byte[] body;
6. }

复制代码

8.消费消息 

1. public class BasicConsumeArguments {
2.     private String consumerTag;
3.     private String queueName;
4.     private boolean autoAck;
5. }

复制代码

9.会合返回 

1. public class SubScribeReturns extends BasicArguments implements Serializable {
2.     private String consumerTag;
3.     private BasicProperties basicProperties;
4.     private byte[] body;
5. }

复制代码

五、实现Broker Server类

1. public class BrokerServer {
2.     private ServerSocket serverSocket = null;
3.     private VirtualHost virtualHost = new VirtualHost("default");
5.     private ConcurrentHashMap<String, Socket> sessions = new ConcurrentHashMap<String,Socket>();
7.     private ExecutorService executorService = null;
9.     private volatile boolean runnbale = true;
11.     public BrokerServer(int port) throws IOException {
12.         serverSocket = new ServerSocket(port);
13.     }
15.     public void start() throws IOException {
16.         System.out.println("[BrokerServer] 启动！");
17.         executorService = Executors.newCachedThreadPool();
18.         try{
19.             while (runnbale){
20.                 Socket clientSocket = serverSocket.accept();
21.                 executorService.submit(() ->{
22.                    processConnection(clientSocket);
23.                 });
24.             }
25.         }catch (SocketException e){
26.             System.out.println("[BrokerServer]服务器停止运行！");
27.         }
28.     }
30.     public void stop() throws IOException {
31.         runnbale = false;
32.         executorService.shutdownNow();
33.         serverSocket.close();
34.     }
36.     private void processConnection(Socket clientSocket){
37.         try(InputStream inputStream = clientSocket.getInputStream(); OutputStream outputStream = clientSocket.getOutputStream()){
38.             try(DataInputStream dataInputStream = new DataInputStream(inputStream)
39.                 ; DataOutputStream dataOutputStream = new DataOutputStream(outputStream)){
40.                 while (true){
41.                     Request request = readRequest(dataInputStream);
42.                     Response response = process(request,clientSocket);
43.                     writeResponse(dataOutputStream,response);
44.                 }
45.             }catch (EOFException | SocketException e){
46.                 System.out.println("[BrokerServer] connection关闭！客户端地址："+clientSocket.getInetAddress().toString()
47.                         +"端口号:"+clientSocket.getPort());
48.             }
49.         }catch (Exception e){
50.             System.out.println("[BrokerServer] connection连接出现异常!");
51.             e.printStackTrace();
52.         }finally {
53.             try{
54.                 clientSocket.close();
55.                 clearClosedSession(clientSocket);
56.             }catch (IOException e){
57.                 e.printStackTrace();
58.             }
59.         }
60.     }
62.     private Response process(Request request, Socket clientSocket) throws IOException, ClassNotFoundException, MqException {
63.         // 1.把request中的payload做初步解析
64.         BasicArguments basicArguments = (BasicArguments) BinaryTool.fromBytes(request.getPayload());
65.         System.out.println("[Request] rid="+basicArguments.getRid()+",channelId=" +basicArguments.getChannelId()+",type="
66.                 +request.getType()+",length="+request.getLength());
67.         // 2.根据type的值区分要做什么操作
68.         boolean ok = true;
69.         if(request.getType()==0x1){
70.             // 创建channel
71.             sessions.put(basicArguments.getChannelId(),clientSocket);
72.             System.out.println("[BrokerServer]创建channel完成！channelId="+basicArguments.getChannelId());
73.         } else if (request.getType()==0x2) {
74.             // 销毁channel
75.             sessions.remove(basicArguments.getChannelId());
76.             System.out.println("[BrokerServer]销毁channel完成！channelId="+basicArguments.getChannelId());
77.         } else if (request.getType()==0x3) {
78.             // 创建交换机
79.             ExchangeDeclareArguments arguments = (ExchangeDeclareArguments) basicArguments;
80.             ok = virtualHost.exchangeDeclare(arguments.getExchangeName(),arguments.getExchangeType()
81.                     ,arguments.isDurable(),arguments.isAutoDelete(),arguments.getArguments());
82.         } else if (request.getType()==0x4) {
83.             ExchangeDeleteArguments arguments = (ExchangeDeleteArguments) basicArguments;
84.             ok = virtualHost.exchangeDelete(arguments.getExchangeName());
85.         } else if (request.getType()==0x5) {
86.             QueueDeclareArguments arguments = (QueueDeclareArguments) basicArguments;
87.             ok = virtualHost.queueDeclare(arguments.getQueueName(),arguments.isDurable(),arguments.isExclusive()
88.                     ,arguments.isAutoDelete(),arguments.getArguments());
89.         } else if (request.getType()==0x6) {
90.             QueueDeleteArguments arguments = (QueueDeleteArguments) basicArguments;
91.             ok = virtualHost.queueDelete(arguments.getQueueName());
92.         } else if (request.getType()==0x7) {
93.             QueueBindArguments arguments = (QueueBindArguments) basicArguments;
94.             ok = virtualHost.queueBind(arguments.getExchangeName(),arguments.getQueueName(),arguments.getBindingKey());
95.         } else if (request.getType()==0x8) {
96.             QueueUnBindArguments arguments = (QueueUnBindArguments) basicArguments;
97.             ok = virtualHost.queueUnBind(arguments.getExchangeName(),arguments.getQueueName());
98.         } else if (request.getType()==0x9) {
99.             BasicPublishArguments arguments = (BasicPublishArguments) basicArguments;
100.             ok = virtualHost.basicPublish(arguments.getExchangeName(),arguments.getRoutingKey()
101.                     ,arguments.getBasicProperties(),arguments.getBody());
102.         }else if (request.getType()==0xa){
103.             BasicConsumeArguments arguments = (BasicConsumeArguments) basicArguments;
104.             ok = virtualHost.basicConsume(arguments.getConsumerTag(), arguments.getQueueName(), arguments.isAutoAck(), new Consumer() {
105.                 @Override
106.                 // 回调函数：把服务器收到的消息直接推送回对应的消费者客户端
107.                 public void handleDelivery(String consumeTag, BasicProperties basicProperties, byte[] body) throws MqException, IOException {
108.                     // 根据consumeTag 其实是channelId 去sessions中查询，找到对应的socket对象
109.                     // 1.根据channelId找到socket对象
110.                     Socket clientSocket = sessions.get(consumeTag);
111.                     if(clientSocket==null || clientSocket.isClosed()){
112.                         throw new MqException("[BrokerServer]订阅消息的客户端已经关闭!");
113.                     }
114.                     // 2.构造响应数据
115.                     SubScribeReturns subScribeReturns = new SubScribeReturns();
116.                     subScribeReturns.setChannelId(consumeTag);
117.                     subScribeReturns.setRid("");  // 此处只有响应，没有请求，不需要去对应
118.                     subScribeReturns.setOk(true);
119.                     subScribeReturns.setConsumerTag(consumeTag);
120.                     subScribeReturns.setBasicProperties(basicProperties);
121.                     subScribeReturns.setBody(body);
122.                     byte[] payload = BinaryTool.toBytes(subScribeReturns);
123.                     Response response = new Response();
124.                     // 0xc 表示服务器给消费者客户端推送的消息数据
125.                     response.setType(0xc);
126.                     response.setLength(payload.length);
127.                     response.setPayload(payload);
128.                     // 3.把数据写回客户端
129.                     DataOutputStream dataOutputStream = new DataOutputStream(clientSocket.getOutputStream());
130.                     writeResponse(dataOutputStream,response);
131.                 }
132.             });
133.         } else if (request.getType()==0xb) {
134.             // 调用basicAck来确认消息
135.             BasicAckArguments arguments = (BasicAckArguments) basicArguments;
136.             ok = virtualHost.basicAck(arguments.getQueueName(),arguments.getMessageId());
137.         }else {
138.             // 当前的type是非法的
139.             throw new MqException("[BrokerServer] 未知的type！type="+request.getType());
140.         }
141.         // 3.构造响应
142.         BasicReturns basicReturns = new BasicReturns();
143.         basicReturns.setChannelId(basicArguments.getChannelId());
144.         basicReturns.setRid(basicArguments.getRid());
145.         basicReturns.setOk(ok);
146.         byte[] payload = BinaryTool.toBytes(basicReturns);
147.         Response response = new Response();
148.         response.setType(request.getType());
149.         response.setLength(payload.length);
150.         response.setPayload(payload);
151.         System.out.println("[Response] rid="+basicReturns.getRid()+",channelId="+basicReturns.getChannelId()
152.                 +",type="+response.getType()+",length="+response.getLength());
153.         return response;
154.     }
156.     private void writeResponse(DataOutputStream dataOutputStream,Response response) throws IOException {
157.         dataOutputStream.writeInt(response.getType());
158.         dataOutputStream.writeInt(response.getLength());
159.         dataOutputStream.write(response.getPayload());
160.         dataOutputStream.flush();
161.     }
163.     private Request readRequest(DataInputStream dataInputStream) throws IOException {
164.         Request request = new Request();
165.         request.setType(dataInputStream.readInt());
166.         request.setLength(dataInputStream.readInt());
167.         byte[] payload = new byte[request.getLength()];
168.         int n = request.getLength();
169.         if(n!=request.getLength()){
170.             throw new IOException("读取请求格式出错！");
171.         }
172.         request.setPayload(payload);
173.         return request;
174.     }
176.     private void clearClosedSession(Socket clientSocket){
177.         List<String> toDeleteChannelId = new ArrayList<>();
178.         for(Map.Entry<String,Socket> entry:sessions.entrySet()){
179.             if(entry.getValue()==clientSocket){
180.                 toDeleteChannelId.add(entry.getKey());
181.             }
182.         }
183.         for(String channelId:toDeleteChannelId){
184.             sessions.remove(channelId);
185.         }
186.         System.out.println("[BrokerServer] 清理session完成！被清理的channelId="+toDeleteChannelId);
187.     }
188. }

复制代码

六、实现连接

  1.connectionFactory类

1. public class ConnectionFactory {
2.     private String host;
3.     private int port;
5.     public Connection newConnection(){
6.         Connection connection = new Connection(host,port);
7.         return connection;
8.     }
9. }

复制代码

2.connection类

1. public class Connection {
2.     private Socket socket =null;
3.     private ConcurrentHashMap<String,Channel> channelMap = new ConcurrentHashMap<>();
5.     private InputStream inputStream;
6.     private OutputStream outputStream;
7.     private DataInputStream dataInputStream;
8.     private DataOutputStream dataOutputStream;
10.     private ExecutorService callbackPool = null;
12.     public Connection(String host,int port) throws IOException {
13.         socket = new Socket(host,port);
14.         inputStream = socket.getInputStream();
15.         outputStream = socket.getOutputStream();
16.         dataInputStream = new DataInputStream(inputStream);
17.         dataOutputStream = new DataOutputStream(outputStream);
18.         callbackPool = Executors.newFixedThreadPool(4);
20.         Thread t = new Thread(() ->{
21.            try {
22.                while(!socket.isClosed()){
23.                    Response response = readResponse();
24.                    dispatchResponse(response);
25.                }
26.            }catch (SocketException e){
27.                System.out.println("[Connection] 连接正常断开！");
28.            }catch (IOException | ClassNotFoundException | MqException e){
29.                System.out.println("[Connection] 连接异常断开！");
30.                e.printStackTrace();
31.            }
32.         });
33.     }
35.     public void close(){
36.         try {
37.             callbackPool.shutdownNow();
38.             channelMap.clear();
39.             dataOutputStream.close();
40.             dataInputStream.close();
41.             socket.close();
42.         } catch (IOException e) {
43.             e.printStackTrace();
44.         }
45.     }
47.     private void dispatchResponse(Response response) throws IOException, ClassNotFoundException, MqException {
48.         if(response.getType()==0xc){
49.             SubScribeReturns subScribeReturns = (SubScribeReturns) BinaryTool.fromBytes(response.getPayload());
50.             Channel channel = channelMap.get(subScribeReturns.getChannelId());
51.             if(channel==null){
52.                 throw new MqException("[Connection] 该消息对应的channel在客户端中不存在！channelId="+channel.getChannelId());
53.             }
54.             callbackPool.submit(() ->{
55.                try {
56.                    channel.getConsumer().handleDelivery(subScribeReturns.getConsumerTag()
57.                            , subScribeReturns.getBasicProperties(), subScribeReturns.getBody());
58.                }catch (MqException | IOException e){
59.                    e.printStackTrace();
60.                }
61.             });
62.         }else {
63.             BasicReturns basicReturns = (BasicReturns) BinaryTool.fromBytes(response.getPayload());
64.             Channel channel = channelMap.get(basicReturns.getChannelId());
65.             if(channel==null){
66.                 throw new MqException("[Connection] 该消息对应的channel在客户端中不存在！channelId="+channel.getChannelId());
67.             }
68.             channel.putReturns(basicReturns);
69.         }
70.     }
72.     public void writeRequest(Request request) throws IOException {
73.         dataOutputStream.writeInt(request.getType());
74.         dataOutputStream.writeInt(request.getLength());
75.         dataOutputStream.write(request.getPayload());
76.         dataOutputStream.flush();
77.         System.out.println("[Connection]发送请求！type="+request.getType()+",length="+request.getLength());
78.     }
80.     public Response readResponse() throws IOException {
81.         Response response = new Response();
82.         response.setType(dataInputStream.readInt());
83.         response.setLength(dataInputStream.readInt());
84.         byte[] payload = new byte[response.getLength()];
85.         int n = dataInputStream.read(payload);
86.         if(n!=response.getLength()){
87.             throw new IOException("响应的数据不完整！");
88.         }
89.         response.setPayload(payload);
90.         System.out.println("[Connection] 收到响应！type="+response.getType()+",length="+response.getType());
91.         return response;
92.     }
94.     public Channel createChannel(){
95.         String channelId = "C-"+ UUID.randomUUID().toString();
96.         Channel channel = new Channel(channelId,this);
97.         System.out.println(channelId);
98.         channelMap.put(channelId,channel);
99.         boolean ok = channel.createChannel();
100.         if(!ok){
101.             channelMap.remove(channelId);
102.             return null;
103.         }
104.         return channel;
105.     }
106. }

复制代码

3.channel类

1. public class Channel {
2.     private String channelId;
3.     private Connection connection;
4.     private ConcurrentHashMap<String, BasicReturns> basicReturnsMap = new ConcurrentHashMap<>();
5.     private Consumer consumer = null;
7.     public Channel(String channelId,Connection connection){
8.         this.channelId = channelId;
9.         this.connection = connection;
10.     }
12.     public boolean createChannel() throws IOException {
13.         BasicArguments basicArguments = new BasicArguments();
14.         basicArguments.setChannelId(channelId);
15.         basicArguments.setRid(generateRid());
16.         System.out.println(basicArguments.getChannelId());
17.         System.out.println(basicArguments.getRid());
18.         byte[] payload = BinaryTool.toBytes(basicArguments);
19.         System.out.println(payload);
21.         Request request = new Request();
22.         request.setType(0x1);
23.         request.setLength(payload.length);
24.         request.setPayload(payload);
26.         connection.writeRequest(request);
27.         BasicReturns basicReturns =waitResult(basicArguments.getRid());
28.         return basicReturns.isOk();
29.     }
32.     private String generateRid(){
33.         return "R-"+ UUID.randomUUID().toString();
34.     }
36.     private BasicReturns waitResult(String rid){
37.         BasicReturns basicReturns =null;
38.         while((basicReturns = basicReturnsMap.get(rid))==null){
39.             synchronized (this){
40.                 try {
41.                     wait();
42.                 }catch (InterruptedException e){
43.                     e.printStackTrace();
44.                 }
45.             }
46.         }
47.         basicReturnsMap.remove(rid);
48.         return basicReturns;
49.     }
51.     public void putReturns(BasicReturns basicReturns){
52.         basicReturnsMap.put(basicReturns.getRid(),basicReturns);
53.         synchronized (this){
54.             notifyAll();
55.         }
56.     }
58.     public boolean close() throws IOException {
59.         BasicArguments basicArguments = new BasicArguments();
60.         basicArguments.setRid(generateRid());
61.         basicArguments.setChannelId(channelId);
62.         byte[] payload = BinaryTool.toBytes(basicArguments);
63.         Request request = new Request();
64.         request.setType(0x2);
65.         request.setLength(payload.length);
66.         request.setPayload(payload);
67.         connection.writeRequest(request);
68.         BasicReturns basicReturns = waitResult(basicArguments.getRid());
69.         return basicReturns.isOk();
70.     }
72.     public boolean exchangeDeclare(String exchangeName, ExchangeType exchangeType
73.             , boolean durable, boolean autoDelete, Map<String,Object> arguments) throws IOException {
74.         ExchangeDeclareArguments exchangeDeclareArguments = new ExchangeDeclareArguments();
75.         exchangeDeclareArguments.setRid(generateRid());
76.         exchangeDeclareArguments.setChannelId(channelId);
77.         exchangeDeclareArguments.setExchangeName(exchangeName);
78.         exchangeDeclareArguments.setExchangeType(exchangeType);
79.         exchangeDeclareArguments.setDurable(durable);
80.         exchangeDeclareArguments.setAutoDelete(autoDelete);
81.         exchangeDeclareArguments.setArguments(arguments);
82.         byte[] payload = BinaryTool.toBytes(exchangeDeclareArguments);
84.         Request request = new Request();
85.         request.setType(0x3);
86.         request.setLength(payload.length);
87.         request.setPayload(payload);
89.         connection.writeRequest(request);
90.         BasicReturns basicReturns = waitResult(exchangeDeclareArguments.getRid());
91.         return basicReturns.isOk();
92.     }
94.     public boolean exchangeDelete(String exchangeName) throws IOException {
95.         ExchangeDeleteArguments exchangeDeleteArguments = new ExchangeDeleteArguments();
96.         exchangeDeleteArguments.setRid(generateRid());
97.         exchangeDeleteArguments.setChannelId(channelId);
98.         exchangeDeleteArguments.setExchangeName(exchangeName);
99.         byte[] payload = BinaryTool.toBytes(exchangeDeleteArguments);
100.         Request request = new Request();
101.         request.setType(0x4);
102.         request.setLength(payload.length);
103.         request.setPayload(payload);
104.         connection.writeRequest(request);
105.         BasicReturns basicReturns =waitResult(exchangeDeleteArguments.getRid());
106.         return basicReturns.isOk();
107.     }
109.     public boolean queueDeclare(String queueName,boolean durable,boolean exclusive,boolean autoDelete
110.             ,Map<String,Object> arguments) throws IOException {
111.         QueueDeclareArguments queueDeclareArguments = new QueueDeclareArguments();
112.         queueDeclareArguments.setRid(generateRid());
113.         queueDeclareArguments.setChannelId(channelId);
114.         queueDeclareArguments.setQueueName(queueName);
115.         queueDeclareArguments.setDurable(durable);
116.         queueDeclareArguments.setExclusive(exclusive);
117.         queueDeclareArguments.setExclusive(autoDelete);
118.         queueDeclareArguments.setArguments(arguments);
119.         byte[] payload= BinaryTool.toBytes(queueDeclareArguments);
120.         Request request = new Request();
121.         request.setType(0x5);
122.         request.setLength(payload.length);
123.         request.setPayload(payload);
124.         connection.writeRequest(request);
125.         BasicReturns basicReturns = waitResult(queueDeclareArguments.getRid());
126.         return basicReturns.isOk();
127.     }
129.     public boolean queueDelete(String queueName) throws IOException {
130.         QueueDeleteArguments queueDeleteArguments = new QueueDeleteArguments();
131.         queueDeleteArguments.setRid(generateRid());
132.         queueDeleteArguments.setChannelId(channelId);
133.         queueDeleteArguments.setQueueName(queueName);
134.         byte[] payload = BinaryTool.toBytes(queueDeleteArguments);
135.         Request request = new Request();
136.         request.setType(0x6);
137.         request.setLength(payload.length);
138.         request.setPayload(payload);
139.         connection.writeRequest(request);
140.         BasicReturns basicReturns = waitResult(queueDeleteArguments.getRid());
141.         return basicReturns.isOk();
142.     }
144.     public boolean queueBind(String exchangeName,String queueName,String bindingKey) throws IOException {
145.         QueueBindArguments queueBindArguments = new QueueBindArguments();
146.         queueBindArguments.setRid(generateRid());
147.         queueBindArguments.setChannelId(channelId);
148.         queueBindArguments.setExchangeName(exchangeName);
149.         queueBindArguments.setQueueName(queueName);
150.         queueBindArguments.setBindingKey(bindingKey);
151.         byte[] payload = BinaryTool.toBytes(queueBindArguments);
152.         Request request = new Request();
153.         request.setType(0x7);
154.         request.setLength(payload.length);
155.         request.setPayload(payload);
156.         connection.writeRequest(request);
157.         BasicReturns basicReturns =waitResult(queueBindArguments.getRid());
158.         return basicReturns.isOk();
159.     }
161.     public boolean queueUnBind(String exchangeName,String queueName) throws IOException {
162.         QueueUnBindArguments queueUnBindArguments = new QueueUnBindArguments();
163.         queueUnBindArguments.setRid(generateRid());
164.         queueUnBindArguments.setChannelId(channelId);
165.         queueUnBindArguments.setExchangeName(exchangeName);
166.         queueUnBindArguments.setQueueName(queueName);
167.         byte[] payload = BinaryTool.toBytes(queueUnBindArguments);
168.         Request request = new Request();
169.         request.setType(0x8);
170.         request.setLength(payload.length);
171.         request.setPayload(payload);
172.         connection.writeRequest(request);
173.         BasicReturns basicReturns = waitResult(queueUnBindArguments.getRid());
174.         return basicReturns.isOk();
175.     }
177.     public boolean basicPublish(String exchangeName, String routingKey, BasicProperties basicProperties,byte[] body) throws IOException {
178.         BasicPublishArguments basicPublishArguments = new BasicPublishArguments();
179.         basicPublishArguments.setRid(generateRid());
180.         basicPublishArguments.setChannelId(channelId);
181.         basicPublishArguments.setRoutingKey(routingKey);
182.         basicPublishArguments.setBasicProperties(basicProperties);
183.         basicPublishArguments.setBody(body);
184.         byte[] payload = BinaryTool.toBytes(basicPublishArguments);
185.         Request request = new Request();
186.         request.setType(0x9);
187.         request.setLength(payload.length);
188.         request.setPayload(payload);
189.         connection.writeRequest(request);
190.         BasicReturns basicReturns = waitResult(basicPublishArguments.getRid());
191.         return basicReturns.isOk();
192.     }
194.     public boolean basicConsume(String queueName,boolean autoAck,Consumer consumer) throws MqException, IOException {
195.         if(this.consumer!=null){
196.             throw new MqException("[Channel] 已经设置过回调函数了！不能重复设置！");
197.         }
198.         this.consumer = consumer;
199.         BasicConsumeArguments basicConsumeArguments = new BasicConsumeArguments();
200.         basicConsumeArguments.setRid(generateRid());
201.         basicConsumeArguments.setChannelId(channelId);
202.         basicConsumeArguments.setConsumerTag(channelId);
203.         basicConsumeArguments.setQueueName(queueName);
204.         basicConsumeArguments.setAutoAck(autoAck);
205.         byte[] payload = BinaryTool.toBytes(basicConsumeArguments);
206.         Request request = new Request();
207.         request.setType(0xa);
208.         request.setLength(payload.length);
209.         request.setPayload(payload);
210.         connection.writeRequest(request);
211.         BasicReturns basicReturns = waitResult(basicConsumeArguments.getRid());
212.         return basicReturns.isOk();
213.     }
216.     public boolean basicAck(String queueName,String messageId) throws IOException {
217.         BasicAckArguments basicAckArguments = new BasicAckArguments();
218.         basicAckArguments.setRid(generateRid());
219.         basicAckArguments.setChannelId(channelId);
220.         basicAckArguments.setQueueName(queueName);
221.         basicAckArguments.setMessageId(messageId);
222.         byte[] payload = BinaryTool.toBytes(basicAckArguments);
223.         Request request = new Request();
224.         request.setType(0xb);
225.         request.setLength(payload.length);
226.         request.setPayload(payload);
227.         connection.writeRequest(request);
228.         BasicReturns basicReturns = waitResult(basicAckArguments.getRid());
229.         return basicReturns.isOk();
230.     }
231. }

复制代码

七、总结

  本篇文章就是本次Java项目“模拟消息队列”的末了一个大的部分了，下一篇文章就是对所写的这个项目举行一个案例编写，然后对此项目举行扩展。感谢观看！

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。