RabbitMQ学习笔记:实现简单的RPC功能

2024-08-27 15:48

本文主要是介绍RabbitMQ学习笔记:实现简单的RPC功能,希望对大家解决编程问题提供一定的参考价值,需要的开发者们随着小编来一起学习吧!

环境

window10
虚拟机、secureCRT
Intellij IDEA

RPC

Remote Procedure Call。

远程过程调用:调用的程序或者函数,并不在本地,而是在远程计算机中。这个时候不能直接调用,需要通过网络来表达调用的语义和传达调用的数据。

要利用RabbitMQ来实现RPC功能是很简单的;客户端发送消息,服务端回复响应的消息。为了接收响应的消息,我们需要在请求中发送一个回调队列。可以使用默认的队列。

String message = "rabbitmq rpc";
// 匿名回调队列
String callbackQueueName = channel.queueDeclare().getQueue();
// 设置回调队列
AMQP.BasicProperties props = 
new AMQP.BasicProperties.Builder().replyTo(callbackQueueName).build();
channel.basicPublish("", "rpc_queue", props, message.getBytes());

流程图:
在这里插入图片描述

步骤

① 客户端创建一个匿名队列作为回调队列。
② 客户端为RPC请求设置2个属性:replayTo用来告知RPC服务端回复请求时的目的队列(也就是上面的匿名队列)即回调队列;。correlationId用来标记请求的唯一性。
③ 请求发送到rpc_queue队列中。
④ RPC服务端监听rpc_queue队列中的请求,当请求来到时,服务端会处理并且把带有结果的消息发送给客户端。接收的队列就是replyTo设定的回调队列。
⑤ 客户端监听回调队列,当有消息时,检查correlationId属性,如果与请求匹配,那就是结果了。

这里贴出官网的代码:

server端:

import com.rabbitmq.client.*;public class RPCServer {private static final String RPC_QUEUE_NAME = "rpc_queue";private static int fib(int n) {if (n == 0) return 0;if (n == 1) return 1;return fib(n - 1) + fib(n - 2);}public static void main(String[] argv) throws Exception {ConnectionFactory factory = new ConnectionFactory();factory.setHost("localhost");try (Connection connection = factory.newConnection();Channel channel = connection.createChannel()) {// 声明一个队列channel.queueDeclare(RPC_QUEUE_NAME, false, false, false, null);// 清空队列里的内容channel.queuePurge(RPC_QUEUE_NAME);// 设置客户端最多接收未被确认ack的消息的个数channel.basicQos(1);System.out.println(" [x] Awaiting RPC requests");Object monitor = new Object();DeliverCallback deliverCallback = (consumerTag, delivery) -> {AMQP.BasicProperties replyProps = new AMQP.BasicProperties.Builder().correlationId(delivery.getProperties().getCorrelationId()).build();String response = "";try {String message = new String(delivery.getBody(), "UTF-8");int n = Integer.parseInt(message);System.out.println(" [.] fib(" + message + ")");response += fib(n);} catch (RuntimeException e) {System.out.println(" [.] " + e.toString());} finally {channel.basicPublish("", delivery.getProperties().getReplyTo(), replyProps, response.getBytes("UTF-8"));channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);// RabbitMq consumer worker thread notifies the RPC server owner threadsynchronized (monitor) {monitor.notify();}}};channel.basicConsume(RPC_QUEUE_NAME, false, deliverCallback, (consumerTag -> { }));// Wait and be prepared to consume the message from RPC client.while (true) {synchronized (monitor) {try {monitor.wait();} catch (InterruptedException e) {e.printStackTrace();}}}}}
}

client端:

import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;import java.io.IOException;
import java.util.UUID;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.TimeoutException;public class RPCClient implements AutoCloseable {private Connection connection;private Channel channel;private String requestQueueName = "rpc_queue";public RPCClient() throws IOException, TimeoutException {ConnectionFactory factory = new ConnectionFactory();factory.setHost("localhost");connection = factory.newConnection();channel = connection.createChannel();}public static void main(String[] argv) {try (RPCClient fibonacciRpc = new RPCClient()) {for (int i = 0; i < 32; i++) {String i_str = Integer.toString(i);System.out.println(" [x] Requesting fib(" + i_str + ")");String response = fibonacciRpc.call(i_str);System.out.println(" [.] Got '" + response + "'");}} catch (IOException | TimeoutException | InterruptedException e) {e.printStackTrace();}}public String call(String message) throws IOException, InterruptedException {final String corrId = UUID.randomUUID().toString();String replyQueueName = channel.queueDeclare().getQueue();AMQP.BasicProperties props = new AMQP.BasicProperties.Builder().correlationId(corrId).replyTo(replyQueueName).build();// 空字符就是使用默认交换器 direct类型的。// 这个默认交换器有个特性:// 每个队列都会自动和这个默认交换器进行绑定,绑定的路由键就是队列名// 这个是rabbitmq自动完成的。// 这也就可以解释:我们没有将回调队列和默认交换器进行绑定,// 但是下面这段代码却可以将消息发送出去。channel.basicPublish("", requestQueueName, props, message.getBytes("UTF-8"));final BlockingQueue<String> response = new ArrayBlockingQueue<>(1);// 监听匿名队列,消费消息String ctag = channel.basicConsume(replyQueueName, true, (consumerTag, delivery) -> {if (delivery.getProperties().getCorrelationId().equals(corrId)) {response.offer(new String(delivery.getBody(), "UTF-8"));}}, consumerTag -> {});String result = response.take();// 取消订阅channel.basicCancel(ctag);return result;}public void close() throws IOException {connection.close();}
}

注意的地方:

① 客户端发送消息使用的是默认交换器;默认交换器有一个特性,所有的队列都会对它进行绑定,绑定的路由键就是队列名。
② 服务端消费消息后,需要返回结果的话,需要将消息发送到回调队列,并且也要将correlationId带回去。

总结

① 默认交换器有个特性:所有的队列都会对它进行绑定,绑定的路由键就是队列名。
② 取消订阅使用channel.basicCancel(ctag);其中ctag是消费者标签;channel.basicConsume()返回的就是消费者标签。
③ 实现RPC关键点:

  • 对replayTo属性指定回调函数
  • 指定请求唯一表示:correlationId;

参考地址:

《RabbitMQ》实战指南

这篇关于RabbitMQ学习笔记:实现简单的RPC功能的文章就介绍到这儿,希望我们推荐的文章对编程师们有所帮助!



http://www.chinasem.cn/article/1112100

相关文章

使用Python和OpenCV库实现实时颜色识别系统

《使用Python和OpenCV库实现实时颜色识别系统》:本文主要介绍使用Python和OpenCV库实现的实时颜色识别系统,这个系统能够通过摄像头捕捉视频流,并在视频中指定区域内识别主要颜色(红... 目录一、引言二、系统概述三、代码解析1. 导入库2. 颜色识别函数3. 主程序循环四、HSV色彩空间详解

PostgreSQL中MVCC 机制的实现

《PostgreSQL中MVCC机制的实现》本文主要介绍了PostgreSQL中MVCC机制的实现,通过多版本数据存储、快照隔离和事务ID管理实现高并发读写,具有一定的参考价值,感兴趣的可以了解一下... 目录一 MVCC 基本原理python1.1 MVCC 核心概念1.2 与传统锁机制对比二 Postg

SpringBoot整合Flowable实现工作流的详细流程

《SpringBoot整合Flowable实现工作流的详细流程》Flowable是一个使用Java编写的轻量级业务流程引擎,Flowable流程引擎可用于部署BPMN2.0流程定义,创建这些流程定义的... 目录1、流程引擎介绍2、创建项目3、画流程图4、开发接口4.1 Java 类梳理4.2 查看流程图4

C++中零拷贝的多种实现方式

《C++中零拷贝的多种实现方式》本文主要介绍了C++中零拷贝的实现示例,旨在在减少数据在内存中的不必要复制,从而提高程序性能、降低内存使用并减少CPU消耗,零拷贝技术通过多种方式实现,下面就来了解一下... 目录一、C++中零拷贝技术的核心概念二、std::string_view 简介三、std::stri

C++高效内存池实现减少动态分配开销的解决方案

《C++高效内存池实现减少动态分配开销的解决方案》C++动态内存分配存在系统调用开销、碎片化和锁竞争等性能问题,内存池通过预分配、分块管理和缓存复用解决这些问题,下面就来了解一下... 目录一、C++内存分配的性能挑战二、内存池技术的核心原理三、主流内存池实现:TCMalloc与Jemalloc1. TCM

OpenCV实现实时颜色检测的示例

《OpenCV实现实时颜色检测的示例》本文主要介绍了OpenCV实现实时颜色检测的示例,通过HSV色彩空间转换和色调范围判断实现红黄绿蓝颜色检测,包含视频捕捉、区域标记、颜色分析等功能,具有一定的参考... 目录一、引言二、系统概述三、代码解析1. 导入库2. 颜色识别函数3. 主程序循环四、HSV色彩空间

苹果macOS 26 Tahoe主题功能大升级:可定制图标/高亮文本/文件夹颜色

《苹果macOS26Tahoe主题功能大升级:可定制图标/高亮文本/文件夹颜色》在整体系统设计方面,macOS26采用了全新的玻璃质感视觉风格,应用于Dock栏、应用图标以及桌面小部件等多个界面... 科技媒体 MACRumors 昨日(6 月 13 日)发布博文,报道称在 macOS 26 Tahoe 中

Python实现精准提取 PDF中的文本,表格与图片

《Python实现精准提取PDF中的文本,表格与图片》在实际的系统开发中,处理PDF文件不仅限于读取整页文本,还有提取文档中的表格数据,图片或特定区域的内容,下面我们来看看如何使用Python实... 目录安装 python 库提取 PDF 文本内容:获取整页文本与指定区域内容获取页面上的所有文本内容获取

基于Python实现一个Windows Tree命令工具

《基于Python实现一个WindowsTree命令工具》今天想要在Windows平台的CMD命令终端窗口中使用像Linux下的tree命令,打印一下目录结构层级树,然而还真有tree命令,但是发现... 目录引言实现代码使用说明可用选项示例用法功能特点添加到环境变量方法一:创建批处理文件并添加到PATH1

Java使用HttpClient实现图片下载与本地保存功能

《Java使用HttpClient实现图片下载与本地保存功能》在当今数字化时代,网络资源的获取与处理已成为软件开发中的常见需求,其中,图片作为网络上最常见的资源之一,其下载与保存功能在许多应用场景中都... 目录引言一、Apache HttpClient简介二、技术栈与环境准备三、实现图片下载与保存功能1.