RabbitMQ RPC

喜欢ヅ旅行 2022-08-12 01:21 174阅读 0赞

在《 [rabbitmq学习2:Work Queues][rabbitmq_2_Work Queues] 》中我们已经知道了在多个worker如何分配耗时的任务。如果我现在要在远程的机器上运行然后得到结果，那应当怎么做呢？那就要用到RPC（ *Remote Procedure Call* or *RPC* ）了！

关于RPC的介绍请参考百度百科里的关于RPC的介绍：[http://baike.baidu.com/view/32726.htm\#sub32726][http_baike.baidu.com_view_32726.htm_sub32726]

现在来看看来看看Rabbitmq中RPC吧！RPC的工作示意图如下：

![7c4514e0-e860-33bb-8787-02a6b37815fc.png][]

上图中的C代表客户端，S表示服务器端；Rabbitmq中的RPC流程如下：

1、首先客户端发送一个reply\_to和corrention\_id的请求,发布到RPC队列中；

2、服务器端处理这个请求，并把处理结果发布到一个回调Queue,此Queue的名称应当与reply\_to的名称一致

3、客户端从回调Queue中得到先前corrention\_id设定的值的处理结果。如果碰到和先前不一样的corrention\_id的值，将会忽略而不是抛出异常。

对于上面所提到的回调Queue中的消费处理使用的是BasicProperties类；而消息 属性在AMQP的协议中规定有14个；而很多大部分我们没有用到。常用的几个属性有：

English代码 [ ![复制代码][icon_copy.gif]][_icon_copy.gif]  ![收藏代码][icon_star.png]![spinner.gif][]

1.  Message properties  
2.  The AMQP protocol predefine a set of 14 properties that go with a message. Most of the properties are rarely used, with the exception of the following:  
3.    
4.  delivery\_mode: Marks a message as persistent (with a value of 2) or transient (any other value). You may remember this property from the second tutorial.   
5.  content\_type: Used to describe the mime-type of the encoding. For example for the often used JSON encoding it is a good practice to set this property to: application/json.   
6.  reply\_to: Commonly used to name a callback queue.   
7.  correlation\_id: Useful to correlate RPC responses with requests.

Message properties
    The AMQP protocol predefine a set of 14 properties that go with a message. Most of the properties are rarely used, with the exception of the following:
    
    delivery_mode: Marks a message as persistent (with a value of 2) or transient (any other value). You may remember this property from the second tutorial. 
    content_type: Used to describe the mime-type of the encoding. For example for the often used JSON encoding it is a good practice to set this property to: application/json. 
    reply_to: Commonly used to name a callback queue. 
    correlation_id: Useful to correlate RPC responses with requests.

delivery\_mode : 标记消息是持久性消息还是瞬态信息。在前面的“Work Queue”中我们已经提到过；

content\_type : 用来描述MIME的类型。如把其类型设定为JSON；

reply\_to : 用于命名一个回调Queue；

correlation\_id : 用于与相关联的请求的RPC响应.

现在我们就开始RPC的程序吧！

client的代码如下：

Java代码 [ ![复制代码][icon_copy.gif]][_icon_copy.gif]  ![收藏代码][icon_star.png]![spinner.gif][]

1.  package com.abin.rabbitmq;  
2.    
3.  import java.util.UUID;  
4.    
5.  import com.rabbitmq.client.AMQP.BasicProperties;  
6.  import com.rabbitmq.client.Channel;  
7.  import com.rabbitmq.client.Connection;  
8.  import com.rabbitmq.client.ConnectionFactory;  
9.  import com.rabbitmq.client.QueueingConsumer;  
10.   
11. public class RPCClient \{  
12.     private Connection connection;  
13.     private Channel channel;  
14.     private String requestQueueName = "rpc\_queue";  
15.     private String replyQueueName;  
16.     private QueueingConsumer consumer;  
17.   
18.     public RPCClient() throws Exception \{  
19.         ConnectionFactory factory = new ConnectionFactory();  
20.         factory.setHost("localhost");  
21.         connection = factory.newConnection();  
22.         channel = connection.createChannel();  
23.   
24.         replyQueueName = channel.queueDeclare().getQueue();  
25.         consumer = new QueueingConsumer(channel);  
26.         channel.basicConsume(replyQueueName, true, consumer);  
27.     \}  
28.   
29.     public String call(String message) throws Exception \{  
30.         String response = null;  
31.         String corrId = UUID.randomUUID().toString();  
32.   
33.         BasicProperties props = new BasicProperties();  
34.         props.setReplyTo(replyQueueName);  
35.         props.setCorrelationId(corrId);  
36.   
37.         channel.basicPublish("", requestQueueName, props, message.getBytes());  
38.   
39.         while (true) \{  
40.             QueueingConsumer.Delivery delivery = consumer.nextDelivery();  
41.             if (delivery.getProperties().getCorrelationId().equals(corrId)) \{  
42.                 response = new String(delivery.getBody(), "UTF-8");  
43.                 break;  
44.             \}  
45.         \}  
46.   
47.         return response;  
48.     \}  
49.   
50.     public void close() throws Exception \{  
51.         connection.close();  
52.     \}  
53.   
54.     public static void main(String\[\] argv) \{  
55.         RPCClient fibonacciRpc = null;  
56.         String response = null;  
57.         try \{  
58.             fibonacciRpc = new RPCClient();  
59.   
60.             System.out.println(" \[x\] Requesting fib(30)");  
61.             response = fibonacciRpc.call("30");  
62.             System.out.println(" \[.\] Got '" + response + "'");  
63.             System.out.println(" \[x\] Requesting fib(-1)");  
64.             response = fibonacciRpc.call("-1");  
65.             System.out.println(" \[.\] Got '" + response + "'");  
66.             System.out.println(" \[x\] Requesting fib(a)");  
67.             response = fibonacciRpc.call("a");  
68.             System.out.println(" \[.\] Got '" + response + "'");  
69.         \} catch (Exception e) \{  
70.             e.printStackTrace();  
71.         \} finally \{  
72.             if (fibonacciRpc != null) \{  
73.                 try \{  
74.                     fibonacciRpc.close();  
75.                 \} catch (Exception ignore) \{  
76.                 \}  
77.             \}  
78.         \}  
79.     \}  
80. \}

package com.abin.rabbitmq;
    
    import java.util.UUID;
    
    import com.rabbitmq.client.AMQP.BasicProperties;
    import com.rabbitmq.client.Channel;
    import com.rabbitmq.client.Connection;
    import com.rabbitmq.client.ConnectionFactory;
    import com.rabbitmq.client.QueueingConsumer;
    
    public class RPCClient {
    	private Connection connection;
    	private Channel channel;
    	private String requestQueueName = "rpc_queue";
    	private String replyQueueName;
    	private QueueingConsumer consumer;
    
    	public RPCClient() throws Exception {
    		ConnectionFactory factory = new ConnectionFactory();
    		factory.setHost("localhost");
    		connection = factory.newConnection();
    		channel = connection.createChannel();
    
    		replyQueueName = channel.queueDeclare().getQueue();
    		consumer = new QueueingConsumer(channel);
    		channel.basicConsume(replyQueueName, true, consumer);
    	}
    
    	public String call(String message) throws Exception {
    		String response = null;
    		String corrId = UUID.randomUUID().toString();
    
    		BasicProperties props = new BasicProperties();
    		props.setReplyTo(replyQueueName);
    		props.setCorrelationId(corrId);
    
    		channel.basicPublish("", requestQueueName, props, message.getBytes());
    
    		while (true) {
    			QueueingConsumer.Delivery delivery = consumer.nextDelivery();
    			if (delivery.getProperties().getCorrelationId().equals(corrId)) {
    				response = new String(delivery.getBody(), "UTF-8");
    				break;
    			}
    		}
    
    		return response;
    	}
    
    	public void close() throws Exception {
    		connection.close();
    	}
    
    	public static void main(String[] argv) {
    		RPCClient fibonacciRpc = null;
    		String response = null;
    		try {
    			fibonacciRpc = new RPCClient();
    
    			System.out.println(" [x] Requesting fib(30)");
    			response = fibonacciRpc.call("30");
    			System.out.println(" [.] Got '" + response + "'");
    			System.out.println(" [x] Requesting fib(-1)");
    			response = fibonacciRpc.call("-1");
    			System.out.println(" [.] Got '" + response + "'");
    			System.out.println(" [x] Requesting fib(a)");
    			response = fibonacciRpc.call("a");
    			System.out.println(" [.] Got '" + response + "'");
    		} catch (Exception e) {
    			e.printStackTrace();
    		} finally {
    			if (fibonacciRpc != null) {
    				try {
    					fibonacciRpc.close();
    				} catch (Exception ignore) {
    				}
    			}
    		}
    	}
    }

server的代码如下：

Java代码 [ ![复制代码][icon_copy.gif]][_icon_copy.gif]  ![收藏代码][icon_star.png]![spinner.gif][]

1.  package com.abin.rabbitmq;  
2.    
3.  import com.rabbitmq.client.AMQP.BasicProperties;  
4.  import com.rabbitmq.client.Channel;  
5.  import com.rabbitmq.client.Connection;  
6.  import com.rabbitmq.client.ConnectionFactory;  
7.  import com.rabbitmq.client.QueueingConsumer;  
8.    
9.  public class RPCServer \{  
10.     private static final String RPC\_QUEUE\_NAME = "rpc\_queue";  
11.   
12.     private static int fib(int n) \{  
13.         if (n > 1)  
14.             return fib(n - 1) + fib(n - 2);  
15.         else  
16.             return n;  
17.     \}  
18.   
19.     public static void main(String\[\] argv) \{  
20.         Connection connection = null;  
21.         Channel channel = null;  
22.         try \{  
23.             ConnectionFactory factory = new ConnectionFactory();  
24.             factory.setHost("localhost");  
25.   
26.             connection = factory.newConnection();  
27.             channel = connection.createChannel();  
28.   
29.             channel.queueDeclare(RPC\_QUEUE\_NAME, false, false, false, null);  
30.   
31.             channel.basicQos(1);  
32.   
33.             QueueingConsumer consumer = new QueueingConsumer(channel);  
34.             channel.basicConsume(RPC\_QUEUE\_NAME, false, consumer);  
35.   
36.             System.out.println(" \[x\] Awaiting RPC requests");  
37.   
38.             while (true) \{  
39.                 String response = null;  
40.   
41.                 QueueingConsumer.Delivery delivery = consumer.nextDelivery();  
42.   
43.                 BasicProperties props = delivery.getProperties();  
44.                 BasicProperties replyProps = new BasicProperties();  
45.                 replyProps.setCorrelationId(props.getCorrelationId());  
46.   
47.                 try \{  
48.                     String message = new String(delivery.getBody(), "UTF-8");  
49.                     int n = Integer.parseInt(message);  
50.   
51.                     System.out.println(" \[.\] fib(" + message + ")");  
52.                     response = "" + fib(n);  
53.                 \} catch (Exception e) \{  
54.                     System.out.println(" \[.\] " + e.toString());  
55.                     response = "";  
56.                 \} finally \{  
57.                     channel.basicPublish("", props.getReplyTo(), replyProps,  
58.                             response.getBytes("UTF-8"));  
59.   
60.                     channel.basicAck(delivery.getEnvelope().getDeliveryTag(),  
61.                             false);  
62.                 \}  
63.             \}  
64.         \} catch (Exception e) \{  
65.             e.printStackTrace();  
66.         \} finally \{  
67.             if (connection != null) \{  
68.                 try \{  
69.                     connection.close();  
70.                 \} catch (Exception ignore) \{  
71.                 \}  
72.             \}  
73.         \}  
74.     \}  
75. \}

package com.abin.rabbitmq;
    
    import com.rabbitmq.client.AMQP.BasicProperties;
    import com.rabbitmq.client.Channel;
    import com.rabbitmq.client.Connection;
    import com.rabbitmq.client.ConnectionFactory;
    import com.rabbitmq.client.QueueingConsumer;
    
    public class RPCServer {
    	private static final String RPC_QUEUE_NAME = "rpc_queue";
    
    	private static int fib(int n) {
    		if (n > 1)
    			return fib(n - 1) + fib(n - 2);
    		else
    			return n;
    	}
    
    	public static void main(String[] argv) {
    		Connection connection = null;
    		Channel channel = null;
    		try {
    			ConnectionFactory factory = new ConnectionFactory();
    			factory.setHost("localhost");
    
    			connection = factory.newConnection();
    			channel = connection.createChannel();
    
    			channel.queueDeclare(RPC_QUEUE_NAME, false, false, false, null);
    
    			channel.basicQos(1);
    
    			QueueingConsumer consumer = new QueueingConsumer(channel);
    			channel.basicConsume(RPC_QUEUE_NAME, false, consumer);
    
    			System.out.println(" [x] Awaiting RPC requests");
    
    			while (true) {
    				String response = null;
    
    				QueueingConsumer.Delivery delivery = consumer.nextDelivery();
    
    				BasicProperties props = delivery.getProperties();
    				BasicProperties replyProps = new BasicProperties();
    				replyProps.setCorrelationId(props.getCorrelationId());
    
    				try {
    					String message = new String(delivery.getBody(), "UTF-8");
    					int n = Integer.parseInt(message);
    
    					System.out.println(" [.] fib(" + message + ")");
    					response = "" + fib(n);
    				} catch (Exception e) {
    					System.out.println(" [.] " + e.toString());
    					response = "";
    				} finally {
    					channel.basicPublish("", props.getReplyTo(), replyProps,
    							response.getBytes("UTF-8"));
    
    					channel.basicAck(delivery.getEnvelope().getDeliveryTag(),
    							false);
    				}
    			}
    		} catch (Exception e) {
    			e.printStackTrace();
    		} finally {
    			if (connection != null) {
    				try {
    					connection.close();
    				} catch (Exception ignore) {
    				}
    			}
    		}
    	}
    }

先运行服务器端，运行结果如下：

Java代码 [ ![复制代码][icon_copy.gif]][_icon_copy.gif]  ![收藏代码][icon_star.png]![spinner.gif][]

1.  \[x\] Awaiting RPC requests

[x] Awaiting RPC requests

再运行运行客户端，运行结果如下：

Java代码 [ ![复制代码][icon_copy.gif]][_icon_copy.gif]  ![收藏代码][icon_star.png]![spinner.gif][]

1.  \[x\] Requesting fib(30)  
2.  \[.\] Got '832040'  
3.  \[x\] Requesting fib(-1)  
4.  \[.\] Got '-1'  
5.  \[x\] Requesting fib(a)  
6.  \[.\] Got ''

[x] Requesting fib(30)
     [.] Got '832040'
     [x] Requesting fib(-1)
     [.] Got '-1'
     [x] Requesting fib(a)
     [.] Got ''

在服务器还可以出现：

Java代码 [ ![复制代码][icon_copy.gif]][_icon_copy.gif]  ![收藏代码][icon_star.png]![spinner.gif][]

1.  \[.\] fib(30)  
2.  \[.\] fib(-1)  
3.  \[.\] java.lang.NumberFormatException: For input string: "a"

原文：

[http://wubin850219.iteye.com/blog/1006300][http_wubin850219.iteye.com_blog_1006300]

[rabbitmq_2_Work Queues]: http://wubin850219.iteye.com/blog/1003840
[http_baike.baidu.com_view_32726.htm_sub32726]: http://baike.baidu.com/view/32726.htm#sub32726
[7c4514e0-e860-33bb-8787-02a6b37815fc.png]: /images/20220810/45fd0dd87a314bd588a5efc8cccfdd91.png
[icon_copy.gif]: http://wubin850219.iteye.com/images/icon_copy.gif
[_icon_copy.gif]: http://wubin850219.iteye.com/blog/1006300#
[icon_star.png]: http://wubin850219.iteye.com/images/icon_star.png
[spinner.gif]: /images/20220810/b7eb5e9d6daa45f1bbec3b31c9e37f97.png
[http_wubin850219.iteye.com_blog_1006300]: http://wubin850219.iteye.com/blog/1006300