Rabbit MQ篇：RabbitMQ实现RPC调用

亦凉 2024-03-17 13:29 67阅读 0赞

`Remote Procedure Call：`远程过程调用，一次远程过程调用的流程即客户端发送一个请求到服务端，服务端根据请求信息进行处理后返回响应信息，客户端收到响应信息后结束。

[![img][]][img 1]

这里生产者作为客户端来调用，消费者作为服务端接收请求然后响应给生产者。

### 1、同步调用 ###

##### 1.1、绑定队列 #####

`@Configuration public class RPCRabbitConfig { @Bean public Queue RPCQueue() { return new Queue("RPCQueue", true, false, false); } @Bean public DirectExchange RPCExchange() { return new DirectExchange("RPCExchange", true, false); } @Bean public Binding bindingRPC() { return BindingBuilder.bind(RPCQueue()).to(RPCExchange()).with("RPC"); } }`

##### 1.2、消费者(服务端) #####

`@Component @RabbitListener(queues = "RPCQueue") @Slf4j public class RPCReceiver { @RabbitHandler public String process(String message) { log.info("接收远程调用请求消息:[{}]", message); return "remote procedure call success!"; } }`

##### 1.3、生产者(客户端) #####

\`\`\` @RestController @Slf4j public class RPCController \{ @Autowired private RabbitTemplate rabbitTemplate;

@PostConstruct
    public void init() {
        // 同步调用设置远程调用响应超时时间，单位：毫秒
        rabbitTemplate.setReplyTimeout(60000);
    }
    
    @PostMapping("/syncRPC")
    public String syncRPC() {
        Object response = rabbitTemplate.convertSendAndReceive("RPCExchange", "RPC", "RPC同步调用");
        String respMsg = response.toString();
        log.info("远程调用响应:[{}]", respMsg);
        return respMsg;
    }

\} \`\`\`

可以通过`setReplyTimeout(long milliseconds)`函数设置超时时间。

##### 1.4、结果 #####

`接收远程调用请求消息:[RPC同步调用] 远程调用响应:[remote procedure call success!]`

### 2、异步调用 ###

##### 2.1、配置Bean #####

`/** * 配置AsyncRabbitTemplate SpringBoot 没有默认的AsyncRabbitTemplate注入，所以这里需要自己配置 * * @param rabbitTemplate * @return */ @Bean public AsyncRabbitTemplate asyncRabbitTemplate(RabbitTemplate rabbitTemplate) { return new AsyncRabbitTemplate(rabbitTemplate); }`

##### 2.2、生产者(客户端) #####

\`\`\` @RestController @Slf4j public class RPCController \{ @Autowired private AsyncRabbitTemplate asyncRabbitTemplate;

@PostMapping("/asyncRPC")
    public String asyncRPC() {
        AsyncRabbitTemplate.RabbitConverterFuture<Object> future = asyncRabbitTemplate.convertSendAndReceive("RPCExchange", "RPC", "RPC异步调用");
        future.addCallback(new ListenableFutureCallback<Object>() {
            @Override
            public void onFailure(Throwable throwable) {
                log.error("异步调用失败", throwable);
            }
    
            @Override
            public void onSuccess(Object o) {
                log.info("异步调用响应:[{}}", o.toString());
            }
        });
        return "ok";
    }

\} \`\`\`

##### 2.3、结果 #####

`SimpleConsumer [queue=amq.rabbitmq.reply-to, consumerTag=amq.ctag-nHw71SucAmOUHb6hGVjaZA identity=5fbed23f] started 接收远程调用请求消息:[RPC异步调用] 异步调用响应:[remote procedure call success!}`

### 参考链接 ###

*  [SpringBoot中使用RabbitMQ(三) RPC][SpringBoot_RabbitMQ_ RPC]

### 代码地址 ###

*  github：[https://github.com/senlinmu1008/rabbitmq-action/tree/master/rabbitmq][https_github.com_senlinmu1008_rabbitmq-action_tree_master_rabbitmq]
 *  gitee：[https://gitee.com/ppbin/rabbitmq-action/tree/master/rabbitmq][https_gitee.com_ppbin_rabbitmq-action_tree_master_rabbitmq] \-

[img]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/15/ae35158f7a6846018bc30eb5478d85f1.png
[img 1]: https://zhaoxiaobin.net/pic/post/rabbitmq/rabbitmq-12-1.jpg
[SpringBoot_RabbitMQ_ RPC]: https://blog.csdn.net/weixin_34383618/article/details/93168325
[https_github.com_senlinmu1008_rabbitmq-action_tree_master_rabbitmq]: https://github.com/senlinmu1008/rabbitmq-action/tree/master/rabbitmq
[https_gitee.com_ppbin_rabbitmq-action_tree_master_rabbitmq]: https://gitee.com/ppbin/rabbitmq-action/tree/master/rabbitmq