rabbitmq学习：远程结果调用

public class RPCClient {
    private final static String RPC_Queue_Name = "rpc_queue";
    public static void main(String[] args) throws IOException, TimeoutException, InterruptedException {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost");
        Connection connection = factory.newConnection();
        Channel channel = connection.createChannel();
        //声明队列
        channel.queueDeclare(RPC_Queue_Name, false, false, false, null);

        //为每一个客户端获取一个随机的回调队列
        String replyQueueName = channel.queueDeclare().getQueue();
        //为每一个客户端创建一个消费者（用于监听回调队列，获取结果）
        QueueingConsumer consumer = new QueueingConsumer(channel);
        //消费者与队列关联
        channel.basicConsume(replyQueueName, true, consumer);

        String response = null;
        String corrId = java.util.UUID.randomUUID().toString();

        //设置replyTo和correlationId属性值
        AMQP.BasicProperties props = new AMQP.BasicProperties.Builder().correlationId(corrId).replyTo(replyQueueName).build();

        //发送消息到rpc_queue队列
        channel.basicPublish("", RPC_Queue_Name, props, "8".getBytes());

        while (true) {
            QueueingConsumer.Delivery delivery = consumer.nextDelivery();
            if (delivery.getProperties().getCorrelationId().equals(corrId)) {
                response = new String(delivery.getBody(),"UTF-8");
                break;
            }
        }
        System.out.println( "fib(8) is " + response);
    }
}

public class RPCServer {

    private final static String RPC_Queue_Name = "rpc_queue";

    public static void main(String[] args) throws IOException, TimeoutException, InterruptedException {

        ConnectionFactory factory = new ConnectionFactory();

        factory.setHost("localhost");

        Connection connection = factory.newConnection();

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

            QueueingConsumer.Delivery delivery = consumer.nextDelivery();

            //获取请求中的correlationId属性值，并将其设置到结果消息的correlationId属性中

            BasicProperties props = delivery.getProperties();

            AMQP.BasicProperties replyProps = new AMQP.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);

        }

    }

  

    private static int fib(int i)

    {

        if(i==0) return 0;

        if (i==1) return 1;

        return fib(i-1) +fib(i-2);

    }

}