public class RPCClient {private static final String RPC_QUEUE_NAME = "rpc_queue";private Connection connection;private Channel channel;private String replyQueueName;private QueueingConsumer consumer;public RPCClient() throws Exception {ConnectionFactory factory = new ConnectionFactory();// 设置MabbitMQ所在主机ip或者主机名factory.setHost("192.168.65.136");factory.setUsername("rabbitmq");factory.setPassword("123456");// 创建一个连接connection = factory.newConnection();// 创建一个频道channel = connection.createChannel();//声明队列channel.queueDeclare(RPC_QUEUE_NAME, false, false, false, null);//为每一个客户端获取一个随机的回调队列replyQueueName = channel.queueDeclare().getQueue();//为每一个客户端创建一个消费者（用于监听回调队列，获取结果）consumer = new QueueingConsumer(channel);//消费者与队列关联channel.basicConsume(replyQueueName, true, consumer);}/*** 获取斐波列其数列的值** @param message* @return* @throws Exception*/public String call(String message) throws Exception{String response = null;String corrId = java.util.UUID.randomUUID().toString();//设置replyTo和correlationId属性值BasicProperties props = new BasicProperties.Builder().correlationId(corrId).replyTo(replyQueueName).build();//发送消息到rpc_queue队列channel.basicPublish("", RPC_QUEUE_NAME, props, message.getBytes());while (true) {System.out.println("OK?");QueueingConsumer.Delivery delivery = consumer.nextDelivery();System.out.println("OK");if (delivery.getProperties().getCorrelationId().equals(corrId)) {response = new String(delivery.getBody(),"UTF-8");break;}}return response;}public static void main(String[] args) throws Exception {RPCClient fibonacciRpc = new RPCClient();String result = fibonacciRpc.call("4");System.out.println( "fib(4) is " + result);}
}

public class RPCServer {private static final String RPC_QUEUE_NAME = "rpc_queue";public static void main(String[] args) throws Exception {ConnectionFactory factory = new ConnectionFactory();// 设置MabbitMQ所在主机ip或者主机名factory.setHost("192.168.65.136");factory.setUsername("rabbitmq");factory.setPassword("123456");// 创建一个连接Connection connection = factory.newConnection();// 创建一个频道Channel channel = connection.createChannel();//声明队列channel.queueDeclare(RPC_QUEUE_NAME, false, false, false, null);//限制：每次最多给一个消费者发送1条消息channel.basicQos(1);//为rpc_queue队列创建消费者，用于处理请求QueueingConsumer consumer = new QueueingConsumer(channel);channel.basicConsume(RPC_QUEUE_NAME, false, consumer);System.out.println(" [x] Awaiting RPC requests");while (true) {QueueingConsumer.Delivery delivery = consumer.nextDelivery();//获取请求中的correlationId属性值，并将其设置到结果消息的correlationId属性中BasicProperties props = delivery.getProperties();BasicProperties replyProps = new BasicProperties.Builder().correlationId(props.getCorrelationId()).build();//获取回调队列名字String callQueueName = props.getReplyTo();String message = new String(delivery.getBody(),"UTF-8");System.out.println(" [.] fib(" + message + ")");//获取结果String response = "" + fib(Integer.parseInt(message));//先发送回调结果channel.basicPublish("", callQueueName, replyProps,response.getBytes());//后手动发送消息反馈channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);}}/*** 计算斐波列其数列的第n项** @param n* @return* @throws Exception*/private static int fib(int n) throws Exception {if (n < 0)throw new Exception("参数错误，n必须大于等于0");if (n == 0)return 0;if (n == 1)return 1;return fib(n - 1) + fib(n - 2);}
}