7.RabbitMQ RFC同步调用

Posted 奋斗的一线码农

tags:

篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了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();
}
}
}
 








以上是关于7.RabbitMQ RFC同步调用的主要内容,如果未能解决你的问题,请参考以下文章

PO配置调用SAP RFC

SAP中IDOC和BAPI和RFC 之间的区别

还在写SQL做SAP二开?通过RFC调用NetWeaver,让HANA数据库操作更可靠

Python 封装SNMP调用接口

python使用mq-fw包

SAP的RFC接口的发布与JAVA调用