Rabbitmq远程过程调用（RPC）模式

秒速五厘米 2023-06-27 04:00 35阅读 0赞

**目录**

[idea创建java项目整合rabbitmq][idea_java_rabbitmq]

[Rabbitmq简单模式][Rabbitmq]

[Rabbitmq工作队列模式（任务队列模式）][Rabbitmq 1]

[Rabbitmq发布/订阅模式][Rabbitmq 2]

[Rabbitmq直接(direct)路由模式][Rabbitmq_direct]

[Rabbitmq主题(topic)路由模式][Rabbitmq_topic]

[Rabbitmq发布者确认模式][Rabbitmq 3]

[Rabbitmq远程过程调用（RPC）模式][Rabbitmq_RPC]

前面的几种模式都是生产者发布了消息到达队列之后，这些消息如何被处理就不管了，现在RPC模式就是客户端发送RPC请求并进入阻塞状态，直到收到服务端处理这些消息之后并返回答案为止，在远程计算机上运行功能并等待结果这种模式通常称为“ 远程过程调用”或“ RPC”。

回调队列：客户端发送请求消息到服务器使用了一个队列，然而服务器要向客户端发送响应消息也需要一个队列。所以我们需要客户端要使用replyTo方法设置一个临时队列给服务端使用。

relatedId（关联ID）：将为每个请求将其设置为唯一值,当我们在客户端收到响应的消息时，响应的relatedId与请求的relatedId进行比较。如果不相等值，我们可以放心地丢弃该消息-它不属于我们的请求。

例如下面代码中的if (delivery.getProperties().getCorrelationId().equals(corrId))

开始演示：

第一步、执行服务端RPCServer.java，监听客户端RPC请求。

第二步、执行客户端RPCClient.java，发送5个RPC请求，并阻塞等待服务端响应返回数据。

客户端打印的结果：

客户端开始发送消息：0

远程服务端响应的内容：本服务已处理的消息【0】

\---------------------------------------------------------------------------

客户端开始发送消息：1

远程服务端响应的内容：本服务已处理的消息【1】

\---------------------------------------------------------------------------

客户端开始发送消息：2

远程服务端响应的内容：本服务已处理的消息【2】

\---------------------------------------------------------------------------

客户端开始发送消息：3

远程服务端响应的内容：本服务已处理的消息【3】

\---------------------------------------------------------------------------

客户端开始发送消息：4

远程服务端响应的内容：本服务已处理的消息【4】

服务端打印的结果：

服务器启动成功，正等待客户端RPC请求......

服务端收到的RPC请求：0

服务端收到的RPC请求：1

服务端收到的RPC请求：2

服务端收到的RPC请求：3

服务端收到的RPC请求：4

演示结论：客户端发送RPC请求并进入阻塞状态，直到收到服务端处理这些消息之后并返回答案为止。

客户端代码：

package com.demo;

import com.rabbitmq.client.AMQP;

import com.rabbitmq.client.Channel;

import com.rabbitmq.client.Connection;

import com.rabbitmq.client.ConnectionFactory;

import java.io.IOException;

import java.util.UUID;

import java.util.concurrent.ArrayBlockingQueue;

import java.util.concurrent.BlockingQueue;

import java.util.concurrent.TimeoutException;

public class RPCClient implements AutoCloseable \{

private Connection connection;

private Channel channel;

private String requestQueueName = "rpc\_queue";

public RPCClient() throws IOException, TimeoutException \{

ConnectionFactory factory = new ConnectionFactory();

factory.setHost("127.0.0.1");

factory.setPort(5672);

factory.setUsername("guest");

factory.setPassword("guest");

connection = factory.newConnection();

channel = connection.createChannel();

public static void main(String\[\] argv) \{

try (RPCClient fibonacciRpc = new RPCClient()) \{

for (int i = 0; i < 5; i++) \{

String i\_str = Integer.toString(i);

System.out.println("客户端开始发送消息：" + i\_str );

String response = fibonacciRpc.call(i\_str);

System.out.println("远程服务端响应的内容：" + response);

System.out.println("---------------------------------------------------------------------------");

\} catch (IOException | TimeoutException | InterruptedException e) \{

e.printStackTrace();

public String call(String message) throws IOException, InterruptedException \{

final String corrId = UUID.randomUUID().toString();

String replyQueueName = channel.queueDeclare().getQueue();

AMQP.BasicProperties props = new AMQP.BasicProperties

.Builder()

.correlationId(corrId)

.replyTo(replyQueueName)

.build();

channel.basicPublish("", requestQueueName, props, message.getBytes("UTF-8"));

final BlockingQueue<String> response = new ArrayBlockingQueue<>(1);

String ctag = channel.basicConsume(replyQueueName, true, (consumerTag, delivery) -> \{

if (delivery.getProperties().getCorrelationId().equals(corrId)) \{

response.offer(new String(delivery.getBody(), "UTF-8"));

\}, consumerTag -> \{

\});

String result = response.take();

channel.basicCancel(ctag);

return result;

public void close() throws IOException \{

connection.close();

服务端代码：

package com.demo;

import com.rabbitmq.client.\*;

public class RPCServer \{

private static final String RPC\_QUEUE\_NAME = "rpc\_queue";

public static void main(String\[\] argv) throws Exception \{

ConnectionFactory factory = new ConnectionFactory();

factory.setHost("127.0.0.1");

factory.setPort(5672);

factory.setUsername("guest");

factory.setPassword("guest");

try (Connection connection = factory.newConnection();

Channel channel = connection.createChannel()) \{

channel.queueDeclare(RPC\_QUEUE\_NAME, false, false, false, null);

channel.queuePurge(RPC\_QUEUE\_NAME);

channel.basicQos(1);

System.out.println("服务器启动成功，正等待客户端RPC请求......");

Object monitor = new Object();

DeliverCallback deliverCallback = (consumerTag, delivery) -> \{

AMQP.BasicProperties replyProps = new AMQP.BasicProperties

.Builder()

.correlationId(delivery.getProperties().getCorrelationId())

.build();

String response = "";

try \{

String message = new String(delivery.getBody(), "UTF-8");

System.out.println("服务端收到的RPC请求：" + message);

response += "本服务已处理的消息【"+message+"】";

\} catch (RuntimeException e) \{

e.printStackTrace();

\} finally \{

channel.basicPublish("", delivery.getProperties().getReplyTo(), replyProps, response.getBytes("UTF-8"));

channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);

synchronized (monitor) \{

monitor.notify();

\};

channel.basicConsume(RPC\_QUEUE\_NAME, false, deliverCallback, (consumerTag -> \{ \}));

while (true) \{

synchronized (monitor) \{

try \{

monitor.wait();

\} catch (InterruptedException e) \{

e.printStackTrace();

[idea_java_rabbitmq]: http://www.yayihouse.com/yayishuwu/chapter/2342
[Rabbitmq]: http://www.yayihouse.com/yayishuwu/chapter/2343
[Rabbitmq 1]: http://www.yayihouse.com/yayishuwu/chapter/2344
[Rabbitmq 2]: http://www.yayihouse.com/yayishuwu/chapter/2345
[Rabbitmq_direct]: http://www.yayihouse.com/yayishuwu/chapter/2346
[Rabbitmq_topic]: http://www.yayihouse.com/yayishuwu/chapter/2347
[Rabbitmq 3]: http://www.yayihouse.com/yayishuwu/chapter/2348
[Rabbitmq_RPC]: http://www.yayihouse.com/yayishuwu/chapter/2361