使用 Netty 实现 RPC 通信框架

🕗 发布于 2024-12-12 08:04 rpc 网络协议 netty

使用 Netty 实现 RPC 通信框架

远程过程调用（RPC，Remote Procedure Call） 是分布式系统中非常重要的通信机制。它允许客户端调用远程服务器上的方法，就像调用本地方法一样。RPC 的核心在于屏蔽底层通信细节，使开发者关注业务逻辑。

Netty 作为一个高性能的网络通信框架，非常适合实现 RPC 框架。本篇文章将介绍如何使用 Netty 实现一个简单的 RPC 通信框架。

1. RPC 通信框架基本原理

1.1 核心组成

RPC 框架的核心模块通常包括：

服务注册与发现：
- 将服务接口及其实现类的地址注册到中心（如注册中心或简单的服务端映射）。
序列化与反序列化：
- 将方法调用、参数等序列化成字节流，传输到远程服务器，服务器再反序列化进行处理。
网络通信：
- 使用 Netty 实现客户端和服务端之间的数据传输。
动态代理：
- 使用动态代理拦截客户端对接口的调用，将调用信息发送到服务端并返回结果。

1.2 RPC 调用流程

客户端：
- 客户端调用代理对象的方法。
- 代理对象将方法、参数打包成 RPC 请求，发送到服务器。
服务器：
- 服务器解析 RPC 请求，定位到具体的方法和参数。
- 调用本地方法，获取结果后返回给客户端。
客户端：
- 接收服务器的响应，将结果返回给调用者。

2. Netty 实现 RPC 通信框架

2.1 项目结构设计

src/main/java/
├── common/         // 通用模块
│   ├── RpcRequest.java       // RPC 请求封装
│   ├── RpcResponse.java      // RPC 响应封装
│   ├── Serializer.java       // 序列化接口
│   ├── JsonSerializer.java   // JSON 序列化实现
├── server/         // 服务端模块
│   ├── RpcServer.java         // RPC 服务端
│   ├── ServiceRegistry.java   // 服务注册表
├── client/         // 客户端模块
│   ├── RpcClient.java         // RPC 客户端
│   ├── RpcProxy.java          // 客户端动态代理

2.2 核心代码实现

2.2.1 通用模块

(1) RPC 请求与响应类

RpcRequest 和 RpcResponse 用于封装客户端发送的请求和服务器的响应。

public class RpcRequest {
    private String methodName; // 方法名
    private String className;  // 类名
    private Object[] parameters; // 参数
    private Class<?>[] paramTypes; // 参数类型

    // Getters and setters
}

public class RpcResponse {
    private Object result; // 方法调用结果
    private String error;  // 错误信息（如果有）
    
    // Getters and setters
}

(2) 序列化接口

为确保传输的数据可以跨网络传递，定义序列化与反序列化的接口。

public interface Serializer {
    byte[] serialize(Object obj);   // 序列化
    <T> T deserialize(byte[] bytes, Class<T> clazz); // 反序列化
}

(3) JSON 序列化实现

使用 Jackson 实现简单的 JSON 序列化。

import com.fasterxml.jackson.databind.ObjectMapper;

public class JsonSerializer implements Serializer {
    private static final ObjectMapper objectMapper = new ObjectMapper();

    @Override
    public byte[] serialize(Object obj) {
        try {
            return objectMapper.writeValueAsBytes(obj);
        } catch (Exception e) {
            throw new RuntimeException("Serialization failed", e);
        }
    }

    @Override
    public <T> T deserialize(byte[] bytes, Class<T> clazz) {
        try {
            return objectMapper.readValue(bytes, clazz);
        } catch (Exception e) {
            throw new RuntimeException("Deserialization failed", e);
        }
    }
}

2.2.2 服务端模块

(1) 服务注册表

ServiceRegistry 用于存储服务接口与实现类的映射。

import java.util.HashMap;
import java.util.Map;

public class ServiceRegistry {
    private final Map<String, Object> services = new HashMap<>();

    public void register(String className, Object serviceImpl) {
        services.put(className, serviceImpl);
    }

    public Object getService(String className) {
        return services.get(className);
    }
}

(2) RPC 服务端

服务端接收 RPC 请求并调用对应的服务实现。

import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.codec.LengthFieldBasedFrameDecoder;
import io.netty.handler.codec.LengthFieldPrepender;

public class RpcServer {
    private final int port;
    private final ServiceRegistry serviceRegistry;

    public RpcServer(int port, ServiceRegistry serviceRegistry) {
        this.port = port;
        this.serviceRegistry = serviceRegistry;
    }

    public void start() throws InterruptedException {
        EventLoopGroup bossGroup = new NioEventLoopGroup();
        EventLoopGroup workerGroup = new NioEventLoopGroup();

        try {
            ServerBootstrap bootstrap = new ServerBootstrap();
            bootstrap.group(bossGroup, workerGroup)
                     .channel(NioServerSocketChannel.class)
                     .childHandler(new ChannelInitializer<Channel>() {
                         @Override
                         protected void initChannel(Channel ch) throws Exception {
                             ChannelPipeline pipeline = ch.pipeline();
                             pipeline.addLast(new LengthFieldBasedFrameDecoder(65536, 0, 4, 0, 4));
                             pipeline.addLast(new LengthFieldPrepender(4));
                             pipeline.addLast(new RpcServerHandler(serviceRegistry));
                         }
                     });

            ChannelFuture future = bootstrap.bind(port).sync();
            System.out.println("RPC Server started on port " + port);
            future.channel().closeFuture().sync();
        } finally {
            bossGroup.shutdownGracefully();
            workerGroup.shutdownGracefully();
        }
    }
}

class RpcServerHandler extends SimpleChannelInboundHandler<RpcRequest> {
    private final ServiceRegistry serviceRegistry;

    public RpcServerHandler(ServiceRegistry serviceRegistry) {
        this.serviceRegistry = serviceRegistry;
    }

    @Override
    protected void channelRead0(ChannelHandlerContext ctx, RpcRequest request) throws Exception {
        Object service = serviceRegistry.getService(request.getClassName());
        if (service == null) {
            ctx.writeAndFlush(new RpcResponse(null, "Service not found"));
            return;
        }

        // 调用服务实现
        Object result = service.getClass()
                               .getMethod(request.getMethodName(), request.getParamTypes())
                               .invoke(service, request.getParameters());
        ctx.writeAndFlush(new RpcResponse(result, null));
    }
}

2.2.3 客户端模块

(1) RPC 客户端

import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;

public class RpcClient {
    private final String host;
    private final int port;

    public RpcClient(String host, int port) {
        this.host = host;
        this.port = port;
    }

    public RpcResponse send(RpcRequest request) throws InterruptedException {
        EventLoopGroup group = new NioEventLoopGroup();
        RpcResponse response = new RpcResponse();

        try {
            Bootstrap bootstrap = new Bootstrap();
            bootstrap.group(group)
                     .channel(NioSocketChannel.class)
                     .handler(new ChannelInitializer<Channel>() {
                         @Override
                         protected void initChannel(Channel ch) throws Exception {
                             ChannelPipeline pipeline = ch.pipeline();
                             pipeline.addLast(new RpcClientHandler(response));
                         }
                     });

            Channel channel = bootstrap.connect(host, port).sync().channel();
            channel.writeAndFlush(request).sync();
            channel.closeFuture().sync();
        } finally {
            group.shutdownGracefully();
        }

        return response;
    }
}

class RpcClientHandler extends SimpleChannelInboundHandler<RpcResponse> {
    private final RpcResponse response;

    public RpcClientHandler(RpcResponse response) {
        this.response = response;
    }

    @Override
    protected void channelRead0(ChannelHandlerContext ctx, RpcResponse msg) throws Exception {
        response.setResult(msg.getResult());
        response.setError(msg.getError());
    }
}

(2) 动态代理

import java.lang.reflect.Proxy;

public class RpcProxy {
    private final RpcClient client;

    public RpcProxy(RpcClient client) {
        this.client = client;
    }

    @SuppressWarnings("unchecked")
    public <T> T create(Class<T> serviceClass) {
        return (T) Proxy.newProxyInstance(serviceClass.getClassLoader(), new Class<?>[]{serviceClass}, (proxy, method, args) -> {
            RpcRequest request = new RpcRequest();
            request.setClassName(serviceClass.getName());
            request.setMethodName(method.getName());
            request.setParameters(args);
            request.setParamTypes(method.getParameterTypes());

            RpcResponse response = client.send(request);
            if (response.getError() != null) {
                throw new RuntimeException(response.getError());
            }
            return response.getResult();
        });
    }
}

2.3 测试示例

定义服务接口和实现：

public interface HelloService {
    String sayHello(String name);
}

public class HelloServiceImpl implements HelloService {
    @Override
    public String sayHello(String name) {
        return "Hello, " + name;
    }
}

服务端注册服务并启动：

ServiceRegistry registry = new ServiceRegistry();
registry.register(HelloService.class.getName(), new HelloServiceImpl());

RpcServer server = new RpcServer(8080, registry);
server.start();

客户端调用服务：

RpcClient client = new RpcClient("localhost", 8080);
RpcProxy proxy = new RpcProxy(client);

HelloService service = proxy.create(HelloService.class);
String result = service.sayHello("Netty");
System.out.println(result); // 输出: Hello, Netty

3. 总结

通过上述代码，我们实现了一个简单的基于 Netty 的 RPC 通信框架，涵盖了服务注册、序列化、网络通信和动态代理等核心模块。

关键点回顾：

服务端：通过 ServiceRegistry 注册服务，并使用 Netty 接收和处理 RPC 请求。
客户端：通过动态代理封装 RPC 调用，简化客户端使用。
序列化：使用 JSON 进行数据的序列化和反序列化。

该框架可以作为一个简单的模板，在实际应用中可扩展为支持注册中心（如 Zookeeper）、负载均衡、异步调用等高级功能的完整 RPC 框架。

原文地址：https://blog.csdn.net/u012561308/article/details/144408416

免责声明：本站文章内容转载自网络资源，如本站内容侵犯了原著者的合法权益，可联系本站删除。更多内容请关注自学内容网（zxcms.com）！

上一篇：学习日志023---初始opencv
下一篇：若依将数据库更改为SQLite

.NET(C#) 如何配置用户首选项及保存用户设置
.NET(C#) 如何配置用户首选项及保存用户设置
阅读更多2024-12-14
【最新】北大数字普惠金融指数数据集-省市县（2011-2023年）
郭峰,王靖一,王芳,孔涛,张勋,程志云.测度中国数字普惠金融发展:指数编制与空间特征[J].经济学(季刊),2020,19(04):1401-1418.时间跨度：省级和城市级指数时间跨度为2011-2
阅读更多2024-12-14
GESP202412 四级【Recamán】题解（AC）
a11ak−1−kkakak−1−kak−1k小杨想知道 Recamán 数列的前n项从小到大排序后的结果。手动计算非常困难，小杨希望你能帮他解决这个问题。
阅读更多2024-12-14
IDEA遇到EasyConnect中的网络资源无法访问的问题
版权声明：本文为博主原创文章，遵循 CC 4.0 BY-SA 版权协议，转载请附上原文出处链接和本声明。原文链接：https://blog.csdn.net/wanshanyu_/article/de
阅读更多2024-12-14
双目摄像头标定方法
此时已经完成标定，左下角为反投影误差，右边为外参可视化。将双目左右目拍的图像上传（左右目最好不少于20张）此时回到主页面，即可看到成功导出。把这些误差大的删除即可。
阅读更多2024-12-14
Servlet、omcat服务器架构与工作原理
Servlet是运行在服务器端的Java程序，它的主要职责之一是接收并处理来自客户端（如浏览器）的HTTP请求。当客户端发送一个请求到服务器时，Servlet可以解析请求中的信息，例如请求的URL路径
阅读更多2024-12-14
Vue生命周期钩子函数：深入解析与实践
作为高级Vue前端开发人员，对Vue组件的生命周期钩子函数有着深刻的理解是至关重要的。生命周期钩子函数是指在Vue组件的创建、更新、销毁等过程中，Vue自动调用的一系列方法。通过这些钩子函数，我们可以
阅读更多2024-12-14
安卓开发--使用android studio发布APP
app发布
阅读更多2024-12-14
数据结构与算法学习笔记----拓扑排序
@ author: 明月清了个风。
阅读更多2024-12-14
python 将数据保存到现有的Excel文件的新工作表
out_file = ‘query.xlsx’df1 = pd.DataFrame(out_data)若直接写入：df1.to_excel(out_file, index=False, sheet_n
阅读更多2024-12-14

使用 Netty 实现 RPC 通信框架