GitHub - pjmike/springboot-rpc-demo: 基于Netty搭建的简易RPC

前言

现在网上有很多关于使用Netty来构建RPC框架的例子，为什么我这里还要写一篇文章进行论述呢，我很清楚我可能没有写得他们那么好。之所以还要写，有两点原因:

一是因为学过Netty之后，还需要去不断实践才能更好的把握Netty的用法，显然，基于Netty实现RPC框架是一个很好的做法；
二是因为目前市面上有很多RPC框架，比如Dubbo，这些框架通讯底层都是Netty，所以说通过这个例子，也可以更好的去体验RPC的设计。

下面我将从以下几点阐述如何基于Netty实现简易的RPC框架：

RPC是什么？
实现RPC框架需要关注哪些方面？
使用Netty如何实现？

RPC是什么？

RPC，远程过程调用，可以做到像本地调用一样调用远程服务，是一种进程间的通信方式，概念想必大家都很清楚，在看到58沈剑写的RPC文章之后，意识到其实我们可以换一种思考方式去理解RPC，也就是从本地调用出发，进而去推导RPC调用

1. 本地函数调用

本地函数是我们经常碰到的，比如下面示例：

public String sayHello(String name) {
    return "hello, " + name;
}

我们只需要传入一个参数，调用sayHello方法就可以得到一个输出，也就是输入参数——>方法体——>输出，入参、出参以及方法体都在同一个进程空间中，这就是本地函数调用

2. Socket通信

那有没有办法实现不同进程之间通信呢？调用方在进程A，需要调用方法A，但是方法A在进程B中

最容易想到的方式就是使用Socket通信，使用Socket可以完成跨进程调用，我们需要约定一个进程通信协议，来进行传参，调用函数，出参。写过Socket应该都知道，Socket是比较原始的方式，我们需要更多的去关注一些细节问题，比如参数和函数需要转换成字节流进行网络传输，也就是序列化操作，然后出参时需要反序列化；使用socket进行底层通讯，代码编程也比较容易出错。

如果一个调用方需要关注这么多问题，那无疑是个灾难，所以有没有什么简单方法，让我们的调用方不需要关注细节问题，让调用方像调用本地函数一样，只要传入参数，调用方法，然后坐等返回结果就可以了呢？

3. RPC框架

RPC框架就是用来解决上面的问题的，它能够让调用方像调用本地函数一样调用远程服务，底层通讯细节对调用方是透明的，将各种复杂性都给屏蔽掉，给予调用方极致体验。

RPC调用需要关注哪些方面

前面就已经说到RPC框架，让调用方像调用本地函数一样调用远程服务，那么如何做到这一点呢？

在使用的时候，调用方是直接调用本地函数，传入相应参数，其他细节它不用管，至于通讯细节交给RPC框架来实现。实际上RPC框架采用代理类的方式，具体来说是动态代理的方式，在运行时动态创建新的类，也就是代理类，在该类中实现通讯的细节问题，比如参数序列化。

当然不光是序列化，我们还需要约定一个双方通信的协议格式，规定好协议格式，比如请求参数的数据类型，请求的参数，请求的方法名等，这样根据格式进行序列化后进行网络传输，然后服务端收到请求对象后按照指定格式进行解码，这样服务端才知道具体该调用哪个方法，传入什么样的参数。

刚才又提到网络传输，RPC框架重要的一环也就是网络传输，服务是部署在不同主机上的，如何高效的进行网络传输，尽量不丢包，保证数据完整无误的快速传递出去？实际上，就是利用我们今天的主角——Netty，Netty是一个高性能的网络通讯框架，它足以胜任我们的任务。

前面说了这么多，再次总结下一个RPC框架需要重点关注哪几个点：

代理（动态代理）
通讯协议
序列化
网络传输

当然一个优秀的RPC框架需要关注的不止上面几点，只不过本篇文章旨在做一个简易的RPC框架，理解了上面关键的几点就够了

基于Netty实现RPC框架

终于到了本文的重头戏了，我们将根据实现RPC需要关注的几个要点（代理、序列化、协议、编解码），使用Netty进行逐一实现

1. Protocol（协议）

首先我们需要确定通信双方的协议格式，请求对象和响应对象

请求对象：

@Data
@ToString
public class RpcRequest {
    /**
     * 请求对象的ID
     */
    private String requestId;
    /**
     * 类名
     */
    private String className;
    /**
     * 方法名
     */
    private String methodName;
    /**
     * 参数类型
     */
    private Class<?>[] parameterTypes;
    /**
     * 入参
     */
    private Object[] parameters;
}

请求对象的ID是客户端用来验证服务器请求和响应是否匹配

响应对象：

@Data
public class RpcResponse {
    /**
     * 响应ID
     */
    private String requestId;
    /**
     * 错误信息
     */
    private String error;
    /**
     * 返回的结果
     */
    private Object result;
}

2. 序列化

市面上序列化协议很多，比如jdk自带的，Google的protobuf，kyro、Hessian等，只要不选择jdk自带的序列化方法，（因为其性能太差，序列化后产生的码流太大），其他方式其实都可以，这里为了方便起见，选用JSON作为序列化协议，使用fastjson作为JSON框架

为了后续扩展方便，先定义序列化接口：

public interface Serializer {
    /**
     * java对象转换为二进制
     *
     * @param object
     * @return
     */
    byte[] serialize(Object object) throws IOException;

    /**
     * 二进制转换成java对象
     *
     * @param clazz
     * @param bytes
     * @param <T>
     * @return
     */
    <T> T deserialize(Class<T> clazz, byte[] bytes) throws IOException;
}

因为我们采用JSON的方式，所以定义JSONSerializer的实现类:

public class JSONSerializer implements Serializer{

    @Override
    public byte[] serialize(Object object) {
        return JSON.toJSONBytes(object);
    }

    @Override
    public <T> T deserialize(Class<T> clazz, byte[] bytes) {
        return JSON.parseObject(bytes, clazz);
    }
}

如果后续要使用其他序列化方式，可以自行实现序列化接口

3. 编解码器

约定好协议格式和序列化方式之后，我们还需要编解码器，编码器将请求对象转换为适合于传输的格式（一般来说是字节流），而对应的解码器是将网络字节流转换回应用程序的消息格式。

编码器实现：

public class RpcEncoder extends MessageToByteEncoder {
    private Class<?> clazz;
    private Serializer serializer;

    public RpcEncoder(Class<?> clazz, Serializer serializer) {
        this.clazz = clazz;
        this.serializer = serializer;
    }


    @Override
    protected void encode(ChannelHandlerContext channelHandlerContext, Object msg, ByteBuf byteBuf) throws Exception {
        if (clazz != null && clazz.isInstance(msg)) {
            byte[] bytes = serializer.serialize(msg);
            byteBuf.writeInt(bytes.length);
            byteBuf.writeBytes(bytes);
        }
    }
}

解码器实现：

public class RpcDecoder extends ByteToMessageDecoder {
    private Class<?> clazz;
    private Serializer serializer;

    public RpcDecoder(Class<?> clazz, Serializer serializer) {
        this.clazz = clazz;
        this.serializer = serializer;
    }
    @Override
    protected void decode(ChannelHandlerContext channelHandlerContext, ByteBuf byteBuf, List<Object> list) throws Exception {
        //因为之前编码的时候写入一个Int型，4个字节来表示长度
        if (byteBuf.readableBytes() < 4) {
            return;
        }
        //标记当前读的位置
        byteBuf.markReaderIndex();
        int dataLength = byteBuf.readInt();
        if (byteBuf.readableBytes() < dataLength) {
            byteBuf.resetReaderIndex();
            return;
        }
        byte[] data = new byte[dataLength];
        //将byteBuf中的数据读入data字节数组
        byteBuf.readBytes(data);
        Object obj = serializer.deserialize(clazz, data);
        list.add(obj);
    }
}

4. Netty 客户端

下面来看看Netty客户端是如何实现的，也就是如何使用Netty开启客户端。

实际上，熟悉Netty的朋友应该都知道，我们需要注意以下几点：

编写启动方法，指定传输使用Channel
指定ChannelHandler，对网络传输中的数据进行读写处理
添加编解码器
添加失败重试机制
添加发送请求消息的方法

下面来看具体的实现代码：

@Slf4j
public class NettyClient {
    private EventLoopGroup eventLoopGroup;
    private Channel channel;
    private ClientHandler clientHandler;
    private String host;
    private Integer port;
    private static final int MAX_RETRY = 5;
    public NettyClient(String host, Integer port) {
        this.host = host;
        this.port = port;
    }
    public void connect() {
        clientHandler = new ClientHandler();
        eventLoopGroup = new NioEventLoopGroup();
        //启动类
        Bootstrap bootstrap = new Bootstrap();
        bootstrap.group(eventLoopGroup)
                //指定传输使用的Channel
                .channel(NioSocketChannel.class)
                .option(ChannelOption.SO_KEEPALIVE, true)
                .option(ChannelOption.TCP_NODELAY, true)
                .option(ChannelOption.CONNECT_TIMEOUT_MILLIS,5000)
                .handler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    protected void initChannel(SocketChannel ch) throws Exception {
                        ChannelPipeline pipeline = ch.pipeline();
                        //添加编码器
                        pipeline.addLast(new RpcEncoder(RpcRequest.class, new JSONSerializer()));
                        //添加解码器
                        pipeline.addLast(new RpcDecoder(RpcResponse.class, new JSONSerializer()));
                        //请求处理类
                        pipeline.addLast(clientHandler);
                    }
                });
        connect(bootstrap, host, port, MAX_RETRY);
    }

    /**
     * 失败重连机制，参考Netty入门实战掘金小册
     *
     * @param bootstrap
     * @param host
     * @param port
     * @param retry
     */
    private void connect(Bootstrap bootstrap, String host, int port, int retry) {
        ChannelFuture channelFuture = bootstrap.connect(host, port).addListener(future -> {
            if (future.isSuccess()) {
                log.info("连接服务端成功");
            } else if (retry == 0) {
                log.error("重试次数已用完，放弃连接");
            } else {
                //第几次重连：
                int order = (MAX_RETRY - retry) + 1;
                //本次重连的间隔
                int delay = 1 << order;
                log.error("{} : 连接失败，第 {} 重连....", new Date(), order);
                bootstrap.config().group().schedule(() -> connect(bootstrap, host, port, retry - 1), delay, TimeUnit.SECONDS);
            }
        });
        channel = channelFuture.channel();
    }

    /**
     * 发送消息
     *
     * @param request
     * @return
     */
    public RpcResponse send(final RpcRequest request) {
        try {
            channel.writeAndFlush(request).await();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        return clientHandler.getRpcResponse(request.getRequestId());
    }
    @PreDestroy
    public void close() {
        eventLoopGroup.shutdownGracefully();
        channel.closeFuture().syncUninterruptibly();
    }
}

我们对于数据的处理重点在于ClientHandler类上，它继承了ChannelDuplexHandler类，可以对出站和入站的数据进行处理

public class ClientHandler extends ChannelDuplexHandler {
    /**
     * 使用Map维护请求对象ID与响应结果Future的映射关系
     */
    private final Map<String, DefaultFuture> futureMap = new ConcurrentHashMap<>();
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        if (msg instanceof RpcResponse) {
            //获取响应对象
            RpcResponse response = (RpcResponse) msg;
            DefaultFuture defaultFuture =
            futureMap.get(response.getRequestId());
            //将结果写入DefaultFuture
            defaultFuture.setResponse(response);
        }
        super.channelRead(ctx,msg);
    }

    @Override
    public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {
        if (msg instanceof RpcRequest) {
            RpcRequest request = (RpcRequest) msg;
            //发送请求对象之前，先把请求ID保存下来，并构建一个与响应Future的映射关系
            futureMap.putIfAbsent(request.getRequestId(), new DefaultFuture());
        }
        super.write(ctx, msg, promise);
    }

    /**
     * 获取响应结果
     *
     * @param requsetId
     * @return
     */
    public RpcResponse getRpcResponse(String requsetId) {
        try {
            DefaultFuture future = futureMap.get(requsetId);
            return future.getRpcResponse(5000);
        } finally {
            //获取成功以后，从map中移除
            futureMap.remove(requsetId);
        }
    }
}

参考文章： https://xilidou.com/2018/09/26/dourpc-remoting/

从上面实现可以看出，我们定义了一个Map,维护请求ID与响应结果的映射关系，目的是为了客户端用来验证服务端响应是否与请求相匹配，因为Netty的channel可能被多个线程使用，当结果返回时，你不知道是从哪个线程返回的，所以需要一个映射关系。

而我们的结果是封装在DefaultFuture中的，因为Netty是异步框架，所有的返回都是基于Future和Callback机制的，我们这里自定义Future来实现客户端"异步调用"

public class DefaultFuture {
    private RpcResponse rpcResponse;
    private volatile boolean isSucceed = false;
    private final Object object = new Object();

    public RpcResponse getRpcResponse(int timeout) {
        synchronized (object) {
            while (!isSucceed) {
                try {
                    object.wait(timeout);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
            return rpcResponse;
        }
    }

    public void setResponse(RpcResponse response) {
        if (isSucceed) {
            return;
        }
        synchronized (object) {
            this.rpcResponse = response;
            this.isSucceed = true;
            object.notify();
        }
    }
}

实际上用了wait和notify机制，同时使用一个boolean变量做辅助

5. Netty服务端

Netty服务端的实现跟客户端的实现差不多，只不过要注意的是，当对请求进行解码过后，需要通过代理的方式调用本地函数。下面是Server端代码：

public class NettyServer implements InitializingBean {
    private EventLoopGroup boss = null;
    private EventLoopGroup worker = null;
    @Autowired
    private ServerHandler serverHandler;
    @Override
    public void afterPropertiesSet() throws Exception {
        //此处使用了zookeeper做注册中心，本文不涉及，可忽略
        ServiceRegistry registry = new ZkServiceRegistry("127.0.0.1:2181");
        start(registry);
    }

    public void start(ServiceRegistry registry) throws Exception {
        //负责处理客户端连接的线程池
        boss = new NioEventLoopGroup();
        //负责处理读写操作的线程池
        worker = new NioEventLoopGroup();
        ServerBootstrap serverBootstrap = new ServerBootstrap();
        serverBootstrap.group(boss, worker)
                .channel(NioServerSocketChannel.class)
                .option(ChannelOption.SO_BACKLOG, 1024)
                .childHandler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    protected void initChannel(SocketChannel ch) throws Exception {
                        ChannelPipeline pipeline = ch.pipeline();
                        //添加解码器
                        pipeline.addLast(new RpcEncoder(RpcResponse.class, new JSONSerializer()));
                        //添加编码器
                        pipeline.addLast(new RpcDecoder(RpcRequest.class, new JSONSerializer()));
                        //添加请求处理器
                        pipeline.addLast(serverHandler);

                    }
                });
        bind(serverBootstrap, 8888);
    }

    /**
     * 如果端口绑定失败，端口数+1,重新绑定
     *
     * @param serverBootstrap
     * @param port
     */
    public void bind(final ServerBootstrap serverBootstrap,int port) {
        serverBootstrap.bind(port).addListener(future -> {
            if (future.isSuccess()) {
                log.info("端口[ {} ] 绑定成功",port);
            } else {
                log.error("端口[ {} ] 绑定失败", port);
                bind(serverBootstrap, port + 1);
            }
        });
    }

    @PreDestroy
    public void destory() throws InterruptedException {
        boss.shutdownGracefully().sync();
        worker.shutdownGracefully().sync();
        log.info("关闭Netty");
    }
}

下面是处理读写操作的Handler类：

@Component
@Slf4j
@ChannelHandler.Sharable
public class ServerHandler extends SimpleChannelInboundHandler<RpcRequest> implements ApplicationContextAware {
    private ApplicationContext applicationContext;
    @Override
    protected void channelRead0(ChannelHandlerContext ctx, RpcRequest msg) {
        RpcResponse rpcResponse = new RpcResponse();
        rpcResponse.setRequestId(msg.getRequestId());
        try {
            Object handler = handler(msg);
            log.info("获取返回结果: {} ", handler);
            rpcResponse.setResult(handler);
        } catch (Throwable throwable) {
            rpcResponse.setError(throwable.toString());
            throwable.printStackTrace();
        }
        ctx.writeAndFlush(rpcResponse);
    }

    /**
     * 服务端使用代理处理请求
     *
     * @param request
     * @return
     */
    private Object handler(RpcRequest request) throws ClassNotFoundException, InvocationTargetException {
        //使用Class.forName进行加载Class文件
        Class<?> clazz = Class.forName(request.getClassName());
        Object serviceBean = applicationContext.getBean(clazz);
        log.info("serviceBean: {}",serviceBean);
        Class<?> serviceClass = serviceBean.getClass();
        log.info("serverClass:{}",serviceClass);
        String methodName = request.getMethodName();

        Class<?>[] parameterTypes = request.getParameterTypes();
        Object[] parameters = request.getParameters();

        //使用CGLIB Reflect
        FastClass fastClass = FastClass.create(serviceClass);
        FastMethod fastMethod = fastClass.getMethod(methodName, parameterTypes);
        log.info("开始调用CGLIB动态代理执行服务端方法...");
        return fastMethod.invoke(serviceBean, parameters);
    }

    @Override
    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
        this.applicationContext = applicationContext;
    }
}

6. 客户端代理

客户端使用Java动态代理，在代理类中实现通信细节，众所众知，Java动态代理需要实现InvocationHandler接口

@Slf4j
public class RpcClientDynamicProxy<T> implements InvocationHandler {
    private Class<T> clazz;
    public RpcClientDynamicProxy(Class<T> clazz) throws Exception {
        this.clazz = clazz;
    }

    @Override
    public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
        RpcRequest request = new RpcRequest();
        String requestId = UUID.randomUUID().toString();

        String className = method.getDeclaringClass().getName();
        String methodName = method.getName();

        Class<?>[] parameterTypes = method.getParameterTypes();

        request.setRequestId(requestId);
        request.setClassName(className);
        request.setMethodName(methodName);
        request.setParameterTypes(parameterTypes);
        request.setParameters(args);
        log.info("请求内容: {}",request);
        
        //开启Netty 客户端，直连
        NettyClient nettyClient = new NettyClient("127.0.0.1", 8888);
        log.info("开始连接服务端：{}",new Date());
        nettyClient.connect();
        RpcResponse send = nettyClient.send(request);
        log.info("请求调用返回结果：{}", send.getResult());
        return send.getResult();
    }
}

在invoke方法中封装请求对象，构建NettyClient对象，并开启客户端，发送请求消息

代理工厂类如下：

public class ProxyFactory {
    public static <T> T create(Class<T> interfaceClass) throws Exception {
        return (T) Proxy.newProxyInstance(interfaceClass.getClassLoader(),new Class<?>[] {interfaceClass}, new RpcClientDynamicProxy<T>(interfaceClass));
    }
}

通过Proxy.newProxyInstance创建接口的代理类

7. RPC远程调用测试

API：

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

准备一个测试API接口

客户端：

@SpringBootApplication
@Slf4j
public class ClientApplication {
    public static void main(String[] args) throws Exception {
        SpringApplication.run(ClientApplication.class, args);
        HelloService helloService = ProxyFactory.create(HelloService.class);
        log.info("响应结果“: {}",helloService.hello("pjmike"));
    }
}

客户端调用接口的方法

服务端：

//服务端实现
@Service
public class HelloServiceImpl implements HelloService {
    @Override
    public String hello(String name) {
        return "hello, " + name;
    }
}

运行结果：

小结

以上我们基于Netty实现了一个非非非常简陋的RPC框架，比起成熟的RPC框架来说相差甚远，甚至说基本的注册中心都没有实现，但是通过本次实践，可以说我对于RPC的理解更深了，了解了一个RPC框架到底需要关注哪些方面，未来当我们使用成熟的RPC框架时，比如Dubbo，能够做到心中有数，能明白其底层不过也是使用Netty作为基础通讯框架。往后，如果更深入翻看开源RPC框架源码时，也相对比较容易

Name		Name	Last commit message	Last commit date
Latest commit History 9 Commits
rpc-client		rpc-client
rpc-common		rpc-common
rpc-server		rpc-server
.gitignore		.gitignore
README.md		README.md
pom.xml		pom.xml

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Repository files navigation

前言

RPC是什么？

1. 本地函数调用

2. Socket通信

3. RPC框架

RPC调用需要关注哪些方面

基于Netty实现RPC框架

1. Protocol（协议）

2. 序列化

3. 编解码器

4. Netty 客户端

5. Netty服务端

6. 客户端代理

7. RPC远程调用测试

小结

参考资料 & 鸣谢

About

Releases

Packages

Languages

pjmike/springboot-rpc-demo

Folders and files

Latest commit

History

Repository files navigation

前言

RPC是什么？

1. 本地函数调用

2. Socket通信

3. RPC框架

RPC调用需要关注哪些方面

基于Netty实现RPC框架

1. Protocol（协议）

2. 序列化

3. 编解码器

4. Netty 客户端

5. Netty服务端

6. 客户端代理

7. RPC远程调用测试

小结

参考资料 & 鸣谢

About

Topics

Resources

Stars

Watchers

Forks

Releases

Packages 0

Languages

Packages