RabbitMQ — 六种工作模式（三）

怼烎@ 2022-11-27 10:21 247阅读 0赞

### 文章目录 ###

*  RPC模式
 *   *  1.1 客户端
     *  1.2 回调队列 Callback Queue
     *  1.3 关联id (correlationId)：
     *  1.4 小结
     *  1.5 RPC的工作方式是这样的:
     *  1.6 完成的代码
     *   *  1.6.1 服务器端
         *  1.6.2 客户端
     *  1.7 virtual host
     *   *  1.7.1 创建virtual host: \`/pd\`
         *  1.7.2 设置虚拟机的用户访问权限

# RPC模式 #

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxMTgwODgxNDAyNQ_size_16_color_FFFFFF_t_70_pic_center]  
如果我们需要在远程电脑上运行一个方法，并且还要等待一个返回结果该怎么办？这和前面的例子不太一样, 这种模式我们通常称为远程过程调用,即RPC.

在本节中，我们将会学习使用RabbitMQ去搭建一个RPC系统：一个客户端和一个可以升级(扩展)的RPC服务器。为了模拟一个耗时任务，我们将创建一个返回斐波那契数列的虚拟的RPC服务。

## 1.1 客户端 ##

在客户端定义一个RPCClient类,并定义一个call()方法,这个方法发送一个RPC请求,并等待接收响应结果

RPCClient client = new RPCClient();
    String result = client.call("4");
    System.out.println( "第四个斐波那契数是: " + result);

## 1.2 回调队列 Callback Queue ##

使用RabbitMQ去实现RPC很容易。一个客户端发送请求信息，并得到一个服务器端回复的响应信息。为了得到响应信息，我们需要在请求的时候发送一个“回调”队列地址。我们可以使用默认队列。下面是示例代码：

//定义回调队列,
    //自动生成对列名,非持久,独占,自动删除
    callbackQueueName = ch.queueDeclare().getQueue();
    
    //用来设置回调队列的参数对象
    BasicProperties props = new BasicProperties
                                .Builder()
                                .replyTo(callbackQueueName)
                                .build();
    //发送调用消息
    ch.basicPublish("", "rpc_queue", props, message.getBytes());

> 消息属性 Message Properties  
> AMQP 0-9-1协议定义了消息的14个属性。大部分属性很少使用，下面是比较常用的4个:  
> `deliveryMode`:将消息标记为持久化(值为2)或非持久化(任何其他值)。  
> `contentType`:用于描述mime类型。例如，对于经常使用的JSON格式，将此属性设置为:application/json。  
> `replyTo`:通常用于指定回调队列。  
> `correlationId`:将RPC响应与请求关联起来非常有用。

## 1.3 关联id (correlationId)： ##

在上面的代码中，我们会为每个RPC请求创建一个回调队列。 这是非常低效的，这里还有一个更好的方法:让我们为每个客户端创建一个回调队列。

这就提出了一个新的问题，在队列中得到一个响应时，我们不清楚这个响应所对应的是哪一条请求。这时候就需要使用关联id（correlationId）。我们将为每一条请求设置唯一的的id值。稍后，当我们在回调队列里收到一条消息的时候，我们将查看它的id属性，这样我们就可以匹配对应的请求和响应。如果我们发现了一个未知的id值，我们可以安全的丢弃这条消息，因为它不属于我们的请求。

## 1.4 小结 ##

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxMTgwODgxNDAyNQ_size_16_color_FFFFFF_t_70_pic_center 1]

## 1.5 RPC的工作方式是这样的: ##

*  对于RPC请求，客户端发送一条带有两个属性的消息:replyTo,设置为仅为请求创建的匿名独占队列,和correlationId,设置为每个请求的惟一id值。
 *  请求被发送到rpc\_queue队列。
 *  RPC工作进程(即:服务器)在队列上等待请求。当一个请求出现时，它执行任务,并使用replyTo字段中的队列将结果发回客户机。
 *  客户机在回应消息队列上等待数据。当消息出现时，它检查correlationId属性。如果匹配请求中的值，则向程序返回该响应数据。

## 1.6 完成的代码 ##

### 1.6.1 服务器端 ###

package rabbitmq.rpc;
    
    import java.io.IOException;
    import java.util.Random;
    import java.util.Scanner;
    
    import com.rabbitmq.client.AMQP;
    import com.rabbitmq.client.BuiltinExchangeType;
    import com.rabbitmq.client.CancelCallback;
    import com.rabbitmq.client.Channel;
    import com.rabbitmq.client.Connection;
    import com.rabbitmq.client.ConnectionFactory;
    import com.rabbitmq.client.DeliverCallback;
    import com.rabbitmq.client.Delivery;
    import com.rabbitmq.client.AMQP.BasicProperties;
    
    public class RPCServer { 
    	public static void main(String[] args) throws Exception { 
    		ConnectionFactory f = new ConnectionFactory();
    		f.setHost("192.168.64.140");
    		f.setPort(5672);
    		f.setUsername("admin");
    		f.setPassword("admin");
    		
    		Connection c = f.newConnection();
    		Channel ch = c.createChannel();
    		/* * 定义队列 rpc_queue, 将从它接收请求信息 * * 参数: * 1. queue, 对列名 * 2. durable, 持久化 * 3. exclusive, 排他 * 4. autoDelete, 自动删除 * 5. arguments, 其他参数属性 */
    		ch.queueDeclare("rpc_queue",false,false,false,null);
    		ch.queuePurge("rpc_queue");//清除队列中的内容
    		
    		ch.basicQos(1);//一次只接收一条消息
    		
    		
    		//收到请求消息后的回调对象
    		DeliverCallback deliverCallback = new DeliverCallback() { 
    			@Override
    			public void handle(String consumerTag, Delivery message) throws IOException { 
    				//处理收到的数据(要求第几个斐波那契数)
    				String msg = new String(message.getBody(), "UTF-8");
    				int n = Integer.parseInt(msg);
    				//求出第n个斐波那契数
    				int r = fbnq(n);
    				String response = String.valueOf(r);
    				
    				//设置发回响应的id, 与请求id一致, 这样客户端可以把该响应与它的请求进行对应
    				BasicProperties replyProps = new BasicProperties.Builder()
    						.correlationId(message.getProperties().getCorrelationId())
    						.build();
    				/* * 发送响应消息 * 1. 默认交换机 * 2. 由客户端指定的,用来传递响应消息的队列名 * 3. 参数(关联id) * 4. 发回的响应消息 */
    				ch.basicPublish("",message.getProperties().getReplyTo(), replyProps, response.getBytes("UTF-8"));
    				//发送确认消息
    				ch.basicAck(message.getEnvelope().getDeliveryTag(), false);
    			}
    		};
    		
    		//
    		CancelCallback cancelCallback = new CancelCallback() { 
    			@Override
    			public void handle(String consumerTag) throws IOException { 
    			}
    		};
    		
    		//消费者开始接收消息, 等待从 rpc_queue接收请求消息, 不自动确认
    		ch.basicConsume("rpc_queue", false, deliverCallback, cancelCallback);
    	}
    
    	protected static int fbnq(int n) { 
    		if(n == 1 || n == 2) return 1;
    		
    		return fbnq(n-1)+fbnq(n-2);
    	}
    }

### 1.6.2 客户端 ###

package rabbitmq.rpc;
    
    import java.io.IOException;
    import java.util.Scanner;
    import java.util.UUID;
    import java.util.concurrent.ArrayBlockingQueue;
    import java.util.concurrent.BlockingQueue;
    
    import com.rabbitmq.client.BuiltinExchangeType;
    import com.rabbitmq.client.CancelCallback;
    import com.rabbitmq.client.Channel;
    import com.rabbitmq.client.Connection;
    import com.rabbitmq.client.ConnectionFactory;
    import com.rabbitmq.client.DeliverCallback;
    import com.rabbitmq.client.Delivery;
    import com.rabbitmq.client.AMQP.BasicProperties;
    
    public class RPCClient { 
    	Connection con;
    	Channel ch;
    	
    	public RPCClient() throws Exception { 
    		ConnectionFactory f = new ConnectionFactory();
    		f.setHost("192.168.64.140");
    		f.setUsername("admin");
    		f.setPassword("admin");
    		con = f.newConnection();
    		ch = con.createChannel();
    	}
    	
    	public String call(String msg) throws Exception { 
    		//自动生成对列名,非持久,独占,自动删除
    		String replyQueueName = ch.queueDeclare().getQueue();
    		//生成关联id
    		String corrId = UUID.randomUUID().toString();
    		
    		//设置两个参数:
    		//1. 请求和响应的关联id
    		//2. 传递响应数据的queue
    		BasicProperties props = new BasicProperties.Builder()
    				.correlationId(corrId)
    				.replyTo(replyQueueName)
    				.build();
    		//向 rpc_queue 队列发送请求数据, 请求第n个斐波那契数
    		ch.basicPublish("", "rpc_queue", props, msg.getBytes("UTF-8"));
    		
    		//用来保存结果的阻塞集合,取数据时,没有数据会暂停等待
    		BlockingQueue<String> response = new ArrayBlockingQueue<String>(1);
    		
    		//接收响应数据的回调对象
    		DeliverCallback deliverCallback = new DeliverCallback() { 
    			@Override
    			public void handle(String consumerTag, Delivery message) throws IOException { 
    				//如果响应消息的关联id,与请求的关联id相同,我们来处理这个响应数据
    				if (message.getProperties().getCorrelationId().contentEquals(corrId)) { 
    					//把收到的响应数据,放入阻塞集合
    					response.offer(new String(message.getBody(), "UTF-8"));
    				}
    			}
    		};
    
    		CancelCallback cancelCallback = new CancelCallback() { 
    			@Override
    			public void handle(String consumerTag) throws IOException { 
    			}
    		};
    		
    		//开始从队列接收响应数据
    		ch.basicConsume(replyQueueName, true, deliverCallback, cancelCallback);
    		//返回保存在集合中的响应数据
    		return response.take();
    	}
    	
    	public static void main(String[] args) throws Exception { 
    		RPCClient client = new RPCClient();
    		while (true) { 
    			System.out.print("求第几个斐波那契数:");
    			int n = new Scanner(System.in).nextInt();
    			String r = client.call(""+n);
    			System.out.println(r);
    		}
    	}
    }

## 1.7 virtual host ##

在RabbitMQ中叫做虚拟消息服务器VirtualHost，每个VirtualHost相当于一个相对独立的RabbitMQ服务器，每个VirtualHost之间是相互隔离的。exchange、queue、message不能互通

### 1.7.1 创建virtual host: `/pd` ###

*  进入虚拟机管理界面  
    ![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxMTgwODgxNDAyNQ_size_16_color_FFFFFF_t_70_pic_center 2]
 *  添加新的虚拟机’/pd’,名称必须以"/"开头  
    ![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxMTgwODgxNDAyNQ_size_16_color_FFFFFF_t_70_pic_center 3]
 *  查看添加的结果  
    ![在这里插入图片描述][20200814233412801.png_pic_center]

### 1.7.2 设置虚拟机的用户访问权限 ###

点击 /pd 虚拟主机, 设置用户 admin 对它的访问权限  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxMTgwODgxNDAyNQ_size_16_color_FFFFFF_t_70_pic_center 4]

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxMTgwODgxNDAyNQ_size_16_color_FFFFFF_t_70_pic_center]: /images/20221124/3ee3a0e3e6214b1fa46c9241015e38b8.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxMTgwODgxNDAyNQ_size_16_color_FFFFFF_t_70_pic_center 1]: /images/20221124/7b8f4b76e9a04b4489c9a20fdcae5a05.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxMTgwODgxNDAyNQ_size_16_color_FFFFFF_t_70_pic_center 2]: /images/20221124/db7f4f698b9644449d8e098c3bca8a22.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxMTgwODgxNDAyNQ_size_16_color_FFFFFF_t_70_pic_center 3]: /images/20221124/22141810d16540d2bb98cfd31a365eee.png
[20200814233412801.png_pic_center]: /images/20221124/97dcb86494544a468e61cb306a51b2d2.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxMTgwODgxNDAyNQ_size_16_color_FFFFFF_t_70_pic_center 4]: /images/20221124/ba08682b49ac45a9937aba7a3ba51317.png