模型▲

模型相当于一对一跟调用方法一样能拿到方法返回的结果这就是典型的RPC模型不仅要传参数还需要拿到返回值

reply回答、答复

通过使用两个独享队列可以让RabbitMQ实现RPC远程过程调用通信模型

其通信过程其实也很简单客户端向服务器消费的独享队列发送一条消息服务器收到该消息后对该消息进行处理然后将处理结果发送给客户端消费的独享队列。

服务器端消费的独享队列负责保存调用参数客户端消费的独享队列负责保存调用的返回值。

使用独享队列可以避免其他连接来读取队列的消息、只有当前连接才能读取该队列的消息这样才能保证服务器能读到客户端发送的每条消息客户端也能读到服务器返回的每条消息。

为了让服务器知道客户端所消费的独享队列客户端发送消息时应该将自己监听的队列名以

reply_to属性

▲为了能准确识别服务器应答消息返回值与客户端请求消息调用参数之间的对应关系

还需要为每条消息都增加一个

reply_to该属性指定了服务器要将返回的消息送回到哪个队列。

correlation_id该属性指定了服务器返回的消息也要添加相同的correlation_id属性。

1服务器启动时它会创建一个名为“rpc_queue”的独享队列名称可以随意并使用服务器端的消费者监听该独享队列的消息。

所有的RPC

调用一定都是先从服务器端的启动开始的。

2客户端启动时它会创建一个匿名默认由RabbitMQ命名的

独享队列并使用客户端的消费者监听该独享队列的消息。

这个独享队列的名字也是

reply_to

属性的属性值3客户端发送带有两个属性的消息一个是代表应答队列名的

reply_to属性该属性值就是第2步客户端所创建的独享队列名另一个是代表消息标识的

correlation_id

属性。

4将消息发送到服务器监听的rpc_queue队列中。

5服务器从rpc_queue队列中读取消息服务器调用处理程序对该消息进行计算将计算结果以消息发送给

reply_to属性

属性值匹配将它返回给应用。

————上面过程其实就是对P2P模型的应用因此无需使用自己的Exchange而是使用系统自动创建的默认Exchange即可。

代码演示

总体流程解释

方法声明队列时不指定具体的消息队列的参数全凭默认生成客户端监听着这个默认的消息队列。

replyTo

这个默认消息队列作用是指定了服务器要将返回的消息送回到这个默认队列

correlationId

解释如果消息发布者指定默认的Exchange那么Exchange就会根据消息发布者发来的消息中携带的路由key假如路由key叫

aaa

去找是否有同样名字叫aaa的消息队列有的话就把消息分发给消息队列没有的话该消息就会被丢弃

rpc_queue

这个消息队列的时候把这个消息队列设置为独享类型exclusivetrue那么

rpc_queue

这个消息队列里面的消息就只能被这个服务端消费不能被其他消费者获取到消息。

因为服务端监听这个

这个消息队列所以服务端拿到这个消息队列的消息之后就会把消息中的

replyTo

业务逻辑处理完消息后服务端需要把这些处理后的消息返回给客户端。

correlationId

标识所以服务端在返回回去时需要把处理好的消息的原本的correlationId

比如客户端发来消息

就是拿出来处理完消息再放回去这样客户端在接收服务端返回来的处理过后的A消息时才能根据

correlationId

服务端返回处理过的消息给客户端也是一个发送消息的过程所以发送消息指定的消息队列就是这个

这个默认消息队列这个replyTo

也是从客户端发送来的消息中获取获取一个属性作用是指定了服务端要将返回的消息送回到这个默认队列。

amq.gen-3xxx

这个默认消息队列所以客户端就能得到自己一开始发送给服务端然后服务端处理完成后返回来的消息。

然后客户端就能根据

这个标识准确找到每个被处理修改过后的消息而不至于找混。

再根据需求去对处理过的消息进行业务操作。

客户端声明并监听着默认队列

amq.gen-3xxx然后发送消息到客户端消息携带有correlationId

replyto

两个属性路由key是rpc_queueexchange是默认的。

服务端声明并监听

消息队列从该队列得到消息messageA后从每个消息中获取该消息对应的

correlationId

两个属性的属性值然后处理消息对于处理完的消息resultMessageA需要把correlationId

replyto

两个属性的属性值重新设置回给resultMessageA在通过Exchange分发回给

replyto

默认的消息队列中获取服务端处理并返回回来的消息进行对应的消费。

ConstantUtil

服务器端----------------//消息队列public

final

------------topic类型的Exchange需要的相关常量----------public

final

www.crazyit.cn,edu.crazyit.org,

crazyit.org,

edu.org};//-------------------------------------------------------//

final

com.rabbitmq.client.Connection;

import

com.rabbitmq.client.ConnectionFactory;import

java.io.IOException;

java.util.concurrent.TimeoutException;//连接工具

public

TimeoutException{//创建连接工厂----这个ConnectionFactory源码可以看出有构造器所以直接new一个出来ConnectionFactory

connectionFactory

ConnectionFactory();//设置连接信息connectionFactory.setHost(localhost);connectionFactory.setPort(5672);connectionFactory.setUsername(ljh);connectionFactory.setPassword(123456);connectionFactory.setVirtualHost(/);

connection

connectionFactory.newConnection();//返回连接return

connection;}

cn.ljh.rabbitmq.producer;import

cn.ljh.rabbitmq.util.ConnectionUtil;

import

cn.ljh.rabbitmq.util.ConstantUtil;

import

java.nio.charset.StandardCharsets;

import

java.util.concurrent.TimeoutException;//消息队列实现

服务器端

TimeoutException{//1、创建连接Connection

conn

ConnectionUtil.getConnection();//2、通过Connection获取Channel。

Channel

channel

conn.createChannel();//3、使用系统自动创建的默认Exchange无需声明Exchange//消息队列设置为独占exclusive:true--------------------------------------------------------------------------------------声明消息队列channel.queueDeclare(ConstantUtil.RPC_QUEUE,true,

*/true,

是否只允许只有这个消息队列的消息消费者才可以消费这个消息队列的消息

*/false,

*///不需要关闭资源因为它也要监听自己消费消息的队列//4、调用Channel

basicConsume()方法开始消费消息----------------------------------------------------------------------------1、服务端监听并消费消息channel.basicConsume(ConstantUtil.RPC_QUEUE,

*/true,new

DefaultConsumer(channel){//处理消息当这个

消息队列收到消息的时候这个方法就会被触发。

重写这个方法Overridepublic

void

/*消息所在的信封,存放消息的exchange、路由key这些*/,AMQP.BasicProperties

properties

IOException{//把消息体中的消息拿出来此处读取到的消息就相当于调用参数-------------------------------------------------------2、服务端获取消息队列的消息String

param

StandardCharsets.UTF_8);//之前只需要用到消息现在需要额外读取消息里面携带的两个属性reply_to

AMQP.BasicProperties

properties.getReplyTo();System.err.println(replyTo:

replyTo);String

properties.getCorrelationId();System.err.println(correlationId:

correlationId);//调用服务器的处理消息的方法最终得到处理后的结果。

该方法可以是任意的业务处理,该方法的返回值result是要被送回客户端的。

------3、服务端处理消费消息String

result

收到来自Exchange为【%s】、路由key为【%s】的消息消息内容为%s%n,envelope.getExchange(),

param);//发送消息的方法需要把返回值result发送回客户端-------------------------------------------------------4、服务端处理消费完的消息返回客户端的操作channel.basicPublish(,

*/replyTo,/*

属性获取到的correlationId再作为参数传回去用于客户端和服务器的匹配。

new

AMQP.BasicProperties().builder().correlationId(correlationId)

correlation_id

属性因为服务器端返回的消息一定要有这个correlationId。

*/result.getBytes(StandardCharsets.UTF_8));}});}//模拟服务器端消费消息要做的处理业务逻辑操作public

static

name){//此处模拟让服务器处理这里的业务有快有慢的情况看correlation_id

rand

cn.ljh.rabbitmq.consumer;import

cn.ljh.rabbitmq.util.ConnectionUtil;

import

cn.ljh.rabbitmq.util.ConstantUtil;

import

java.nio.charset.StandardCharsets;

import

java.util.concurrent.ConcurrentHashMap;

import

java.util.concurrent.TimeoutException;//消息队列实现

客户端

ConcurrentHashMap();//客户端发送的消息参数public

static

TimeoutException{//1、创建连接工厂设置连接信息然后再通过连接工厂获取连接Connection

conn

ConnectionUtil.getConnection();//2、通过Connection获取Channel

channel

自动创建的、自动命名的、持久化的、独享的、会自动删除的【默认队列】AMQP.Queue.DeclareOk

declareOk

channel.queueDeclare();System.out.println(declareOk

用于得到默认队列的返回值也就是默认队列的名字之前声明是我们自己设置队列名这里用默认的队列就用.getQueue()

queueName

declareOk.getQueue();System.out.println(queueName:

basicConsume()方法开始处理消费消息-----------------------------------------------------------------2、客户端监听服务端处理完消息后返回来的消息channel.basicConsume(queueName

/*消息的确认模式是否自动确认该消息已经被消费完成并返回确认消息给消息队列*/,new

DefaultConsumer(channel){//处理消息当这个消息队列收到消息的时候这个方法就会被触发。

重写这个方法Overridepublic

void

/*消息所在的信封,存放消息的exchange、路由key这些*/,AMQP.BasicProperties

properties

IOException{//把消息体中的消息拿出来String

resultMessage

StandardCharsets.UTF_8);//此处需要指定每个返回值对应的是哪个参数靠的就是correlation_idString

correlationId

properties.getCorrelationId();//根据服务器端返回的消息中的correlation_id

param

paramMap.get(correlationId);System.err.println(客户端发出去的消息内容param

服务端处理后返回来的消息内容resultMessage);//printf格式化输出函数

%s输出字符串

收到来自Exchange为【%s】、路由key为【%s】的消息消息内容为%s%n,envelope.getExchange(),

resultMessage);//得到服务器的返回值之后整个调用过程就完成了此时就应该从

Map

)。

paramMap.remove(correlationId);}});//客户端发送消息---------------------------------------------------------------------------代码运行后先执行这段--------------------1、客户端发送消息for

(int

客户端发送消息携带的路由key是服务端监听的消息队列的名字且使用了默认的Exchange这就意味着消息会被发送给服务器监听的那个消息队列

*/new

AMQP.BasicProperties().builder().correlationId(i

correlation_id

correlation_id该属性指定了服务器返回的消息也要添加相同的correlation_id属性*/.replyTo(queueName)

reply_to:

构建这个BasicProperties对象这个对象主要存这个correlationId属性

*/params[i].getBytes(StandardCharsets.UTF_8));}}

pom.xml

xmlnshttp://maven.apache.org/POM/4.0.0xmlns:xsihttp://www.w3.org/2001/XMLSchema-instancexsi:schemaLocationhttp://maven.apache.org/POM/4.0.0

http://maven.apache.org/xsd/maven-4.0.0.xsdmodelVersion4.0.0/modelVersiongroupIdcn.ljh/groupIdartifactIdrabbitmq_rpc/artifactIdversion1.0.0/versionnamerabbitmq_rpc/name!--

--propertiesmaven.compiler.source11/maven.compiler.sourcemaven.compiler.target11/maven.compiler.targetproject.build.sourceEncodingUTF-8/project.build.sourceEncodingjava.version11/java.version/properties!--

--dependencies!--

--dependencygroupIdcom.rabbitmq/groupIdartifactIdamqp-client/artifactIdversion5.13.0/version/dependencydependencygroupIdorg.projectlombok/groupIdartifactIdlombok/artifactIdversion1.18.16/versionscopecompile/scope/dependency/dependencies/project