7.RabbitMQ RFC同步調用

RabbitMQ RFC同步調用是使用了兩個異步調用完成的，生產者調用消費者的同時，本身也做爲消費者等待某一隊列的返回消息，消費者接受到生產者的消息同時，也做爲消息發送者發送一消息給生產者。參考下圖：

調用流程以下：

其餘的消息服務器實現同步調用也是相似的原理，好比ActiveMQ。

 

下面編寫消費者類Server

生產者Client代碼

啓動一命令行，將當前目錄轉移到項目所在的目錄

在Eclipse中運行Client

 

 

代碼：

package com.test.rfc;

 

import java.io.IOException;

 

import com.rabbitmq.client.AMQP;

import com.rabbitmq.client.Channel;

import com.rabbitmq.client.Connection;

import com.rabbitmq.client.ConnectionFactory;

import com.rabbitmq.client.Consumer;

import com.rabbitmq.client.DefaultConsumer;

import com.rabbitmq.client.Envelope;

 

public class Server {

public static void main(String[] argv) {

ConnectionFactory factory = new ConnectionFactory();

    factory.setUsername("admin");

    factory.setPassword("admin");

        factory.setHost("192.168.169.142"); //使用默認端口5672

        Connection connection = null;

        

try {

connection = factory.newConnection();

final Channel channel = connection.createChannel();

String queueName = "queue_rpc";

channel.queueDeclare(queueName, false, false, false, null);

Consumer consumer = new DefaultConsumer(channel) {

 

@Override

public void handleDelivery(String consumerTag,

Envelope envelope, AMQP.BasicProperties properties,

byte[] body) throws IOException 

{

System.out.println("test1" + new String(body));

AMQP.BasicProperties replyProps = new AMQP.BasicProperties.Builder()

.correlationId(properties.getCorrelationId())

.build();

String response = "hello client,I'm rfc server";

channel.basicPublish("", properties.getReplyTo(),

replyProps, response.getBytes());

channel.basicAck(envelope.getDeliveryTag(), false);

}

};

channel.basicConsume(queueName, false, consumer);

} catch (IOException e) {

e.printStackTrace();

}

}

 

}

 

package com.test.rfc;

 

import java.io.IOException;

import java.util.UUID;

import java.util.concurrent.ArrayBlockingQueue;

import java.util.concurrent.BlockingQueue;

 

import com.rabbitmq.client.*;

 

public class Client {

public static void main(String[] argv) {

try

{

//發送消息的隊列,Server在這個隊列上接受消息

String queueName = "queue_rpc";

ConnectionFactory factory = new ConnectionFactory();

    factory.setUsername("admin");

    factory.setPassword("admin");

       factory.setHost("192.168.169.142"); //使用默認端口5672

       Connection connection = null;

connection = factory.newConnection();

Channel channel = connection.createChannel();

//生成臨時的隊列,Client在這隊列上等待Server返回信息,Server向這個隊列發消息

String replyQueueName = channel.queueDeclare().getQueue();

//生成惟一ID

final String corrId = UUID.randomUUID().toString();

AMQP.BasicProperties props = new AMQP.BasicProperties.Builder()

.correlationId(corrId).replyTo(replyQueueName).build();

//客戶端發送RFC請求

channel.basicPublish("", queueName, props, "GetUserInfo".getBytes());

//Server返回消息

final BlockingQueue response = new ArrayBlockingQueue(1);

channel.basicConsume(replyQueueName, true,

new DefaultConsumer(channel) {

@Override

public void handleDelivery(String consumerTag,

Envelope envelope, AMQP.BasicProperties properties,

byte[] body) throws IOException {

System.out.println(properties.getCorrelationId());

if (properties.getCorrelationId().equals(corrId)) {

response.offer(body);

}

}

});

byte[] b = response.take();

System.out.println(new String(b));

channel.close();

connection.close();

}

catch(Exception e)

{

e.printStackTrace();

}

}

}