环境

window10

虚拟机、secureCRT

Intellij IDEA

RPC

Remote Procedure Call。

远程过程调用：调用的程序或者函数，并不在本地，而是在远程计算机中。这个时候不能直接调用，需要通过网络来表达调用的语义和传达调用的数据。

要利用RabbitMQ来实现RPC功能是很简单的；客户端发送消息，服务端回复响应的消息。为了接收响应的消息，我们需要在请求中发送一个回调队列。可以使用默认的队列。

String message = "rabbitmq rpc";// 匿名回调队列String callbackQueueName = channel.queueDeclare().getQueue();// 设置回调队列AMQP.BasicProperties props = new AMQP.BasicProperties.Builder().replyTo(callbackQueueName).build();channel.basicPublish("", "rpc_queue", props, message.getBytes());

流程图：

步骤

① 客户端创建一个匿名队列作为回调队列。

② 客户端为RPC请求设置2个属性：replayTo用来告知RPC服务端回复请求时的目的队列（也就是上面的匿名队列）即回调队列；。correlationId用来标记请求的唯一性。

③ 请求发送到rpc_queue队列中。

④ RPC服务端监听rpc_queue队列中的请求，当请求来到时，服务端会处理并且把带有结果的消息发送给客户端。接收的队列就是replyTo设定的回调队列。

⑤ 客户端监听回调队列，当有消息时，检查correlationId属性，如果与请求匹配，那就是结果了。

这里贴出官网的代码：

server端：

import com.rabbitmq.client.*;public class RPCServer {       private static final String RPC_QUEUE_NAME = "rpc_queue";    private static int fib(int n) {           if (n == 0) return 0;        if (n == 1) return 1;        return fib(n - 1) + fib(n - 2);    }    public static void main(String[] argv) throws Exception {           ConnectionFactory factory = new ConnectionFactory();        factory.setHost("localhost");        try (Connection connection = factory.newConnection();             Channel channel = connection.createChannel()) {   	        // 声明一个队列	        channel.queueDeclare(RPC_QUEUE_NAME, false, false, false, null);	        // 清空队列里的内容	        channel.queuePurge(RPC_QUEUE_NAME);	        // 设置客户端最多接收未被确认ack的消息的个数	        channel.basicQos(1);            System.out.println(" [x] Awaiting RPC requests");            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");                    int n = Integer.parseInt(message);                    System.out.println(" [.] fib(" + message + ")");                    response += fib(n);                } catch (RuntimeException e) {                       System.out.println(" [.] " + e.toString());                } finally {                       channel.basicPublish("", delivery.getProperties().getReplyTo(), replyProps, response.getBytes("UTF-8"));                    channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);                    // RabbitMq consumer worker thread notifies the RPC server owner thread                    synchronized (monitor) {                           monitor.notify();                    }                }            };            channel.basicConsume(RPC_QUEUE_NAME, false, deliverCallback, (consumerTag -> {    }));            // Wait and be prepared to consume the message from RPC client.            while (true) {                   synchronized (monitor) {                       try {                           monitor.wait();                    } catch (InterruptedException e) {                           e.printStackTrace();                    }                }            }        }    }}

client端：

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("localhost");        connection = factory.newConnection();        channel = connection.createChannel();    }    public static void main(String[] argv) {           try (RPCClient fibonacciRpc = new RPCClient()) {               for (int i = 0; i < 32; i++) {                   String i_str = Integer.toString(i);                System.out.println(" [x] Requesting fib(" + i_str + ")");                String response = fibonacciRpc.call(i_str);                System.out.println(" [.] Got '" + response + "'");            }        } 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();        // 空字符就是使用默认交换器 direct类型的。        // 这个默认交换器有个特性：        // 每个队列都会自动和这个默认交换器进行绑定，绑定的路由键就是队列名        // 这个是rabbitmq自动完成的。        // 这也就可以解释：我们没有将回调队列和默认交换器进行绑定，        // 但是下面这段代码却可以将消息发送出去。        channel.basicPublish("", requestQueueName, props, message.getBytes("UTF-8"));        final BlockingQueue
   
     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();    }}

注意的地方：

① 客户端发送消息使用的是默认交换器；默认交换器有一个特性，所有的队列都会对它进行绑定，绑定的路由键就是队列名。

② 服务端消费消息后，需要返回结果的话，需要将消息发送到回调队列，并且也要将correlationId带回去。

总结

① 默认交换器有个特性：所有的队列都会对它进行绑定，绑定的路由键就是队列名。

② 取消订阅使用channel.basicCancel(ctag);其中ctag是消费者标签；channel.basicConsume()返回的就是消费者标签。

③ 实现RPC关键点：

对replayTo属性指定回调函数

指定请求唯一表示：correlationId;

参考地址：

《RabbitMQ》实战指南

转载地址：http://eryh.baihongyu.com/

你可能感兴趣的文章