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

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);

    }

}