SpringBoot+RabbitMQ实现RPC调用

说到RPC（RemoteProcedureCallProtocol远程过程调用协议），小伙伴们脑海中浮现的大概就是RESTfulAPI、Dubbo、WebService、JavaRMI、CORBA等。其实RabbitMQ也为我们提供了RPC功能，使用起来也非常简单。今天宋哥就通过一个简单的案例和大家分享一下SpringBoot+RabbitMQ是如何实现一个简单的RPC调用的。注意，有些朋友可能对RabbitMQ实现RPC调用有一些误解，认为这不简单？创建两个消息队列，queue_1和queue_2。首先，客户端向queue_1发送消息，服务端监听queue_1上的消息，收到后进行处理；处理完成后，服务端将消息发送给queue_2，然后客户端监听queue_2队列上的消息，这样就知道服务端的处理结果了。这个方法也不是不行，就是有点麻烦！RabbitMQ提供了现成的解决方案，可以直接使用，非常方便。接下来我们一起学习吧。1.架构首先我们来看一个简单的架构图：这个图把问题说的很清楚了：首先，Client发送消息。与普通消息相比，这条消息多了两个关键内容：一是correlation_id，表示这条消息的唯一id，内容是reply_to，表示消息回复队列的名称。服务器从消息发送队列中获取消息，并处理相应的业务逻辑。处理完成后，将处理结果发送到reply_to指定的回调队列。客户端从回调队列中读取消息以了解消息的执行情况。这种情况其实很适合处理异步调用。2.实践接下来我们通过一个具体的例子来看看这是如何实现的。2.1客户端开发首先，我们创建一个名为producer的SpringBoot项目。作为消息生产者，在创建时添加web和rabbitmq依赖，如下图：项目创建成功后，首先在application.properties中配置RabbitMQ的基本信息，如下：spring.rabbitmq.host=localhostspring.rabbitmq.port=5672spring.rabbitmq.username=guestspring.rabbitmq.password=guestspring.rabbitmq.publisher-confirm-type=correlatedspring.rabbitmq.publisher-returns=true这个配置的前四行很容易理解，所以我不会详细介绍。接下来两行：首先，配置消息确认方式。我们通过相关确认。只有启用此配置后，correlation_id才会包含在以后的消息中。只有通过correlation_id才能发送消息和链接返回值。最后一行配置是开启发送失败返回。接下来我们提供一个配置类，如下：/***@author江南一场小雨*@微信公众号江南一场小雨*@网站http://www.itboyhub.com*@国际站http://www.javaboy.org*@微信a_java_boy*@GitHubhttps://github.com/lenve*@Giteehttps://gitee.com/lenve*/@ConfigurationpublicclassRabbitConfig{publicstaticfinalStringRPC_QUEUE1="queue_1";publicstaticfinalStringRPC_QUEUE2="queue_2";publicstaticfinalStringRPC_EXCHANGE="rpc_exchange";/***设置消息发送RPC队列*/@BeanQueuemsgQueue(){returnnewQueue(RPC_QUEUE1);}/***设置返回队列*/@BeanQueuereplyQueue(){returnnewQueue(RPC_QUEUE2);}/***设置交换*/@BeanTopicExchangeexchange(){returnnewTopicExchange(RPC_EXCHANGE);}/***请求队列和exchangeBuilder绑定*/@BeanBindingmsgBinding(){returnBindingBuilder.bind(msgQueue()).to(exchange()).with(RPC_QUEUE1);}/***返回队列和交换绑定*/@BeanBindingreplyBinding(){r返回BindingBuilder.bind(replyQueue()).to(exchange()).with(RPC_QUEUE2);}/***使用RabbitTemplate收发消息*并设置回调队列地址*/@BeanRabbitTemplaterabbitTemplate(ConnectionFactoryconnectionFactory){RabbitTemplatetemplate=newRabbitTemplate(connectionFactory);template.setReplyAddress(RPC_QUEUE2);template.setReplyTimeout(6000);返回模板；}/***为返回队列设置监听器*/@BeanSimpleMessageListenerContainerreplyContainer(ConnectionFactoryListenerconnectionMtainerage){=newSimpleMessageListenerContainer();container.setConnectionFactory(connectionFactory);container.setQueueNames(RPC_QUEUE2);container.setMessageListener(rabbitTemplate(connectionFactory));返回容器；}}在这个配置类中，我们分别配置消息发送队列msgQueue和消息返回队列replyQueue，然后将这两个队列绑定到消息交换机上。这是RabbitMQ的正常运行，没什么好说的。在SpringBoot中，我们负责发送消息的工具是RabbitTemplate。默认情况下，系统会自动提供这个工具，但是这里我们需要重新自定义这个工具，主要是添加消息发送的返回队列，最后我们需要给返回队列设置一个监听器。好了，那我们就可以开始发送具体消息了：/***@author江南小雨*@微信公众号江南小雨*@Websitehttp://www.itboyhub.com*@国际站http://www.javaboy.org*@微信a_java_boy*@GitHubhttps://github.com/lenve*@Giteehttps://gitee.com/lenve*/@RestControllerpublicclassRpcClientController{privatestaticfinalLoggerlogger=LoggerFactory.getLogger(RpcClientController.class);@AutowiredprivateRabbitTemplaterabbitTemplate;@GetMapping("/send")publicStringsend(Stringmessage){//创建消息对象MessagenewMessage=MessageBuilder.withBody(message.getBytes()).build();logger.info("客户端发送：{}",newMessage);//客户端发送消息Messageresult=rabbitTemplate.sendAndReceive(RabbitConfig.RPC_EXCHANGE,RabbitConfig.RPC_QUEUE1,newMessage);字符串响应="";if(result!=null){//获取发送消息的correlationIdStringcorrelationId=newMessage.getMessageProperties().getCorrelationId();logger.info("合作relationId:{}",correlationId);//获取响应头信息HashMapheaders=(HashMap)result.getMessageProperties().getHeaders();//获取返回的消息id服务器StringmsgId=(String)headers.get("spring_returned_message_correlation");if(msgId.equals(correlationId)){response=newString(result.getBody());logger.info("clientreceive:{}",response);}}returnresponse;}}这块代码其实就是一些常规代码，我挑几个关键节点说一下：消息发送调用sendAndReceive方法，有自己的返回值，returnvalue是服务器返回的，在消息服务器返回的消息中，header信息中包含spring_returned_message_correlation字段，即消息发送时的correlation_id，通过消息发送时的correlation_id和spring_returned_message_correlation的值返回的消息头中的字段，我们可以通过比较返回的消息内容和发送的消息绑定在一起，确认返回的内容是针对发送的消息。这是整个客户端的开发。其实核心就是sendAndReceive方法的调用。调用虽然简单，但是准备工作还是够多的比如我们不在application.properties中配置correlation，那么发送的消息中就没有correlation_id，这样返回的消息内容就无法与发送的消息内容相关联.2.2服务端开发我们再来看看服务端开发。首先，创建一个名为consumer的SpringBoot项目。项目中添加的依赖与客户端开发创建的依赖一致，这里不再赘述。然后配置application.properties配置文件。该文件的配置也与客户端中的配置一致，这里不再赘述。接下来，提供了一个RabbitMQ配置类。这个配置类比较简单。简单的配置消息队列，绑定消息开关，如下：本站http://www.javaboy.org*@WeChata_java_boy*@GitHubhttps://github.com/lenve*@Giteehttps://gitee.com/lenve*/@ConfigurationpublicclassRabbitConfig{publicstaticfinalStringRPC_QUEUE1=“队列_1”；publicstaticfinalStringRPC_QUEUE2="queue_2";publicstaticfinalStringRPC_EXCHANGE="rpc_exchange";/***配置消息发送队列*/@BeanQueuemsgQueue(){returnnewQueue(RPC_QUEUE1);}/***设置返回队列*/@BeanQueuereplyQueue(){returnnewQueue(RPC_QUEUE2);}/***设置开关*/@BeanTopicExchangeexchange(){returnnewTopicExchange(RPC_EXCHANGE);}/***请求队列和交换绑定*/@BeanBindingmsgBinding(){returnBindingBuilder.bind(msgQueue()).to(exchange()).with(RPC_QUEUE1);}/***返回队列和交换绑定*/@Bean绑定replyBinding(){returnBindingBuilder.bind(replyQueue()).to(exchange()).with(RPC_QUEUE2);}}最后我们来看消息的消费：@ComponentpublicclassRpcServerController{privatestaticfinalLoggerlogger=LoggerFactory.getLogger(RpcServerController.class);@AutowiredprivateRabbitTemplaterabbitTemplate;@RabbitListener(queues=RabbitConfig.RPC_QUEUE1)publicvoidprocess(Messagemsg){logger.info("serverreceive:{}",msg.toString());消息响应=MessageBuilder.withBody(("我收到了："+newString(msg.getBody())).getBytes()).build();CorrelationDatacorrelationData=newCorrelationData(msg.getMessageProperties().getCorrelationId());rabbitTemplate.sendAndReceive(RabbitConfig.RPC_EXCHANGE,RabbitConfig.RPC_QUEUE2,response,correlationData);}}这里的逻辑比较简单：服务端先收到消息，打印出来。服务器提取原始消息中的correlation_id。服务端调用sendAndReceive方法将消息发送到参数为correlation_id的RPC_QUEUE2队列。服务器发送消息后，客户端将收到服务器返回的结果。好的，你完成了。2.3测试接下来我们进行一个简单的测试。首先启动RabbitMQ。接下来分别启动producer和consumer，然后在postman中调用producer的接口进行测试，如下：可以看到已经收到了服务端的返回信息。我们看一下生产者的运行日志：可以看到消息发出后，也收到了消费者返回的消息。可以看到，消费者也收到了客户端的消息。3.总结好了，一个小案例，让小伙伴们体验一个RabbitMQ实现RPC调用。公众号江南一点鱼后台回复mq_rpc即可获取本文案例~有兴趣的朋友可以试试~