RabbitMQ如何实现RPC远程调用消息队列

这篇文章主要介绍了RabbitMQ如何实现RPC远程调用消息队列,具有一定借鉴价值,感兴趣的朋友可以参考下,希望大家阅读完这篇文章之后大有收获,下面让小编带着大家一起了解一下。

  客户端接口

  我们创建一个客户端类来说明如何使用RPC服务,暴露一个call方法来发送RPC请求和数据获取结果。

FibonacciRpcClient fibonacciRpc = new FibonacciRpcClient();  
String result = fibonacciRpc.call("4");
System.out.println( "fib(4) is " + result);

  尽管RPC是编程中一种常见的模式,但其也常常饱受批评。因为程序员常常不知道调用的方法是本地方法还是一个RPC方法,这在调试中常常增加一些不必要的复杂性。我们应该简化代码,而不是滥用RPC导致代码变的臃肿。

  回调队列

  一般来说,通过RabbitMQ实现RPC非常简单,客户端发送一个请求消息,服务端响应消息就完成了。为了接收到响应内容,我们在请求中发送”callback“队列地址,也可以使用默认的队列。

callbackQueueName = channel.queueDeclare().getQueue();
BasicProperties props = new BasicProperties .Builder().replyTo(callbackQueueName) .build(); 
channel.basicPublish("", "rpc_queue", props, message.getBytes());

  AMQP协议中预定了14个消息属性,除了下面几个,其它的都很少使用:

  deliveryMode : 标识消息是持久化还是瞬态的。

  contentType : 描述 mime-type的编码类型,如JSON编码为”application/json“。

  replyTo : 通常在回调队列中使用。

  correlationId : 在请求中关联RPC响应时使用。

  关联Id(Correlation Id)

  在前面的方法中,要求在每个RPC请求创建回调队列,这可真是一件繁琐的事情,但幸运的是我们有个好方法-在每个客户端创建一个简单的回调队列。

  这样问题又来了,队列如何知道这些响应来自哪个请求呢?这时候correlationId就出场了。我们在每个请求中都设置一个唯一的值,这样我们在回调队列中接收消息的时候就能知道是哪个请求发送的。如果收到未知的correlationId,就废弃该消息,因为它不是我们发出的请求。

  你可能会问,为什么抛弃未知消息而不是抛出错误呢?这是由服务器竞争资源所导致的。尽管这不太可能,试想一下,如果RPC服务器在发送完响应后而在发送应答消息前死掉了,重启RPC服务器会重新发送请求。这就是我们在客户机上优雅地处理重复的反应,RPC应该是等同的。

RabbitMQ如何实现RPC远程调用消息队列

  (1)客户端启动,创建一个匿名且唯一的回调队列。

  (2)对每个RPC请求,客户端发送一个包含replyTo和correlationId两个属性的消息。

  (3)请求发送到rpc_queue队列。

  (4)RPC服务在队列中等待请求,当请求出现时,根据replyTo字段使用队列将结果发送到客户端。

  (5)客户端在回调队列中等待数据。当消息出现时,它会检查correlationId属性,如果该值匹配的话,就会返回响应结果给应用。

  示例代码

  RPCServer.java

package com.favccxx.favrabbit;

import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.QueueingConsumer;
import com.rabbitmq.client.AMQP.BasicProperties;

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) {
		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.Builder().correlationId(props.getCorrelationId())
						.build();

				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) {
				}
			}
		}
	}
}

  RPCClient.java       

package com.favccxx.favrabbit;

import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.QueueingConsumer;
import com.rabbitmq.client.AMQP.BasicProperties;
import java.util.UUID;

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.Builder().correlationId(corrId).replyTo(replyQueueName).build();

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

		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 + "'");
		} catch (Exception e) {
			e.printStackTrace();
		} finally {
			if (fibonacciRpc != null) {
				try {
					fibonacciRpc.close();
				} catch (Exception ignore) {
				}
			}
		}
	}
}

  先启动RPCServer,然后运行RPCClient,控制台输出如下内容

RPCClient[x] Requesting fib(30)

RPCClient[.] Got '832040'


RPCServer[x] Awaiting RPC requests

RPCServer[.] fib(30)



感谢你能够认真阅读完这篇文章,希望小编分享的“RabbitMQ如何实现RPC远程调用消息队列”这篇文章对大家有帮助,同时也希望大家多多支持亿速云,关注行业资讯频道,更多相关知识等着你来学习!