当前位置: 开发笔记 > 编程语言 > 正文

Netty实现RPC的思路

作者：手机用户2502927973 | 来源：互联网 | 2023-09-14 11:21

RPCRPC（RemoteProcedureCall）远程过程调用，是一个计算机通信协议，可以实现远程调用远程接口就想调用本地接口一样的高效。

RPC

RPC（Remote Procedure Call）远程过程调用，是一个计算机通信协议，可以实现远程调用远程接口就想调用本地接口一样的高效。

image.png

分布式组件中：外部RESTful内部RPC。

RPC调用流程

image.png

服务消费方(client)以本地调用方式调用服务
client stub 接收到调用后负责将方法、参数等封装成能够进行网络传输的消息体
client stub 将消息进行编码并发送到服务端
server stub 收到消息后进行解码
server stub 根据解码结果调用本地的服务
本地服务执行并将结果返回给 server stub
server stub 将返回导入结果进行编码并发送至消费方
client stub 接收到消息并进行解码
服务消费方(client)得到结果

小结：RPC 的目标就是将 2-8 这些步骤都封装起来，用户无需关心这些细节，可以像调用本地方法一样即可完成远程服务调用。

通过客户端代理，实现通过netty通信，实现接口的远程调用。

code

RPC客户端

package com.pl.netty.rpc.client;

import io.netty.bootstrap.Bootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.string.StringDecoder;
import io.netty.handler.codec.string.StringEncoder;

import java.lang.reflect.Proxy;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * 
 *
 * @Description: TODO
 * 
 * @ClassName NettyClient
 * @Author pl
 * @Date 2021/3/5
 * @Version V1.0.0
 */
public class NettyClient {
    //创建一个线程池
    private static ExecutorService executor= Executors.newFixedThreadPool(5);

    private static NettyClientHandler client;

    //编写方法，使用代理模式，获取一个代理对象
    public Object getBean(final Class> serviceClass, final String providerName) {
        //通过代理对目标对象的方法进行增强
        return Proxy.newProxyInstance(Thread.currentThread().getContextClassLoader(), new Class>[]{serviceClass},
                (proxy, method, args) -> {
                    System.out.println("代理被调用");
                    if (client == null)
                        initClient();
                    //方法参数
                    client.setPara(providerName + args[0]);
                    return executor.submit(client).get();
                });
    }


    //初始化客户端
    private static void initClient() {
        client = new NettyClientHandler();
        //创建EventLoopGroup
        NioEventLoopGroup group = new NioEventLoopGroup();
        try {
            Bootstrap bootstrap = new Bootstrap();
            bootstrap.group(group)
                    .channel(NioSocketChannel.class)
                    .option(ChannelOption.TCP_NODELAY, true)
                    .handler(new ChannelInitializer() {
                        @Override
                        protected void initChannel(SocketChannel ch) throws Exception {
                            ChannelPipeline pipeline = ch.pipeline();
                            pipeline.addLast(new StringDecoder());
                            pipeline.addLast(new StringEncoder());
                            pipeline.addLast(client);
                        }
                    });
            ChannelFuture channelFuture = bootstrap.connect("127.0.0.1", 7000).sync();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}

NettyClientHandler

package com.pl.netty.rpc.client;

import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;

import java.util.concurrent.Callable;

/**
 * 
 *
 * @Description: TODO
 * 
 * @ClassName NettyClientHandler
 * @Author pl
 * @Date 2021/3/5
 * @Version V1.0.0
 */
public class NettyClientHandler extends ChannelInboundHandlerAdapter implements Callable {

    private ChannelHandlerContext context; //上下文
    private String result; //返回的结果
    private String para; //客户端调用方法时，传入的参数

    //与服务端创建连接后调用
    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        System.out.println("通道连接成功");
        cOntext= ctx; //因为我们在其他方法会使用到 ctx
    }

    @Override
    public synchronized void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        result = msg.toString();
        notify(); //唤醒等待的线程
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        cause.printStackTrace();
        ctx.close();
    }

    //被代理对象的调用，真正发送数据给服务器，发送完后就阻塞，等待被唤醒（channelRead）
    @Override
    public synchronized Object call() throws Exception {
        System.out.println("线程被调用-----");
        context.writeAndFlush(para);
        //进行wait
        wait(); //等待 channelRead 获取到服务器的结果后，进行唤醒。
        return result; //服务方返回的结果
    }

    public void setPara(String para){
        this.para = para;
    }
}

ClientBootStrap

package com.pl.netty.rpc.client;

import com.pl.netty.rpc.server.HelloService;

/**
 * 
 *
 * @Description: TODO
 * 
 * @ClassName ClientBootStrap
 * @Author pl
 * @Date 2021/3/5
 * @Version V1.0.0
 */
public class ClientBootStrap {

    //这里定义协议头
    public static final String providerName = "HelloService#hello#";

    public static void main(String[] args) throws InterruptedException {
        //创建一个消费者
        NettyClient customer = new NettyClient();
        //创建代理对象
        HelloService service = (HelloService) customer.getBean(HelloService.class, providerName);

        //通过代理对象调用服务提供者的方法
        String res = service.hello("你好 Dubbo");
        System.out.println("调用的结果，res = " + res);
        Thread.sleep(2000);
    }
}

RPC服务端

NettyServer

package com.pl.netty.rpc.server;

import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.codec.string.StringDecoder;
import io.netty.handler.codec.string.StringEncoder;

/**
 * 
 *
 * @Description: TODO
 * 
 * @ClassName NettyServer
 * @Author pl
 * @Date 2021/3/5
 * @Version V1.0.0
 */
public class NettyServer {
    public static  void startServer(String hostName, int port) {
        startServer0(hostName, port);
    }

    private static void startServer0(String hostname, int port) {
        NioEventLoopGroup bossGroup = new NioEventLoopGroup(1);
        NioEventLoopGroup workerGroup = new NioEventLoopGroup();
        try {
            ServerBootstrap serverBootstrap = new ServerBootstrap();
            serverBootstrap.group(bossGroup, workerGroup)
                    .channel(NioServerSocketChannel.class)
                    .childHandler(new ChannelInitializer() {
                        @Override
                        protected void initChannel(SocketChannel ch) throws Exception {
                            ChannelPipeline pipeline = ch.pipeline();
                            pipeline.addLast(new StringDecoder());
                            pipeline.addLast(new StringEncoder());
                            pipeline.addLast(new NettyServerHandler()); //业务处理器
                        }
                    });
            ChannelFuture channelFuture = serverBootstrap.bind(hostname,port).sync();
            System.out.println("服务提供方开始运行");
            channelFuture.channel().closeFuture().sync();
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            bossGroup.shutdownGracefully();
            workerGroup.shutdownGracefully();
        }
    }

    public static void main(String[] args) {
        startServer("127.0.0.1",7000);
    }
}

NettyServerHandler

package com.pl.netty.rpc.server;

import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;

/**
 * 
 *
 * @Description: TODO
 * 
 * @ClassName NettyServerHandler
 * @Author pl
 * @Date 2021/3/5
 * @Version V1.0.0
 */
public class NettyServerHandler extends ChannelInboundHandlerAdapter {
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        //获取客户端发送的消息，并调用服务
        System.out.println("msg=" + msg);
        //客户端在调用服务器的api 时，我们需要定义一个协议
        //比如要求，每次发消息时，都必须以某个字符串开头 "HelloService#hello#你好"
        if (msg.toString().startsWith("HelloService#hello#")) {
            String result = new HelloServiceImpl().hello(msg.toString().substring(msg.toString().lastIndexOf("#") + 1));
            ctx.writeAndFlush(result);
        }
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        ctx.close();
    }
}

HelloService

package com.pl.netty.rpc.server;

/**
 * 
 *
 * @Description: TODO
 * 
 * @ClassName HelloService
 * @Author pl
 * @Date 2021/3/5
 * @Version V1.0.0
 */
public interface  HelloService {
    String hello(String message);
}

HelloServiceImpl

package com.pl.netty.rpc.server;

/**
 * 
 *
 * @Description: TODO
 * 
 * @ClassName HelloServiceImpl
 * @Author pl
 * @Date 2021/3/5
 * @Version V1.0.0
 */
public class HelloServiceImpl implements HelloService {
    @Override
    public String hello(String message) {
        System.out.println("收到客户端消息=" + message);
        //根据 message 返回不同的结果
        if(message != null) {
            return "你好客户端，我已经收到你的消息【" + message + "】";
        } else {
            return "你好客户端，我已经收到你的消息。";
        }
    }
}

输出

image.png

推荐阅读

spring
缓存分布式锁 Redis

分布式锁现在Redis基本上没家公司都在使用，只是各自使用的场景不以，但Redis最出名的还是做为缓存服务器，提搞服务器的的吞吐量，下面我们来围绕这个作为缓存做一个总结今天的目标其 ... [详细]

蜡笔小新 2023-10-17 11:51:10
python
Thrift教程初级篇——RPC框架Thrift的安装环境变量配置与第一个实例

本文介绍了RPC框架Thrift的安装环境变量配置与第一个实例，讲解了RPC的概念以及如何解决跨语言、c++客户端、web服务端、远程调用等需求。Thrift开发方便上手快，性能和稳定性也不错，适合初学者学习和使用。 ... [详细]

蜡笔小新 2023-12-13 17:36:52
netty
Netty拆包粘包问题解决 —— 特殊结束符

本文介绍了解决Netty拆包粘包问题的一种方法——使用特殊结束符。在通讯过程中，客户端和服务器协商定义一个特殊的分隔符号，只要没有发送分隔符号，就代表一条数据没有结束。文章还提供了服务端的示例代码。 ... [详细]

蜡笔小新 2023-12-14 18:02:45
netty
mongodb数据迁移,程序员未来职场之路

一面自我介绍对象相等的判断，equals方法实现。可以简单描述挫折，并说明自己如何克服，最终有哪些收获。职业规划表明自己决心，首先自己不准备继续求学了，必须招工作了。希望去哪 ... [详细]

蜡笔小新 2023-10-17 13:57:50
go
mapreduce源码分析总结

这篇文章总结的非常到位,故而转之一MapReduce概述MapReduce是一个用于大规模数据处理的分布式计算模型，它最初是由Google工程师设计并实现的ÿ ... [详细]

蜡笔小新 2023-10-17 12:36:35
int
微服务下的几个难点问题及常见的解决方案

原文链接：https:cloud.tencent.comdevelopernews1362051背景介绍1.1幂等性定义数学定义在数学里，幂等有 ... [详细]

蜡笔小新 2023-10-15 14:24:29
spring
Java开发面试问题，2021网易Java高级面试题及答案，实战案例

前言大厂面试真题向来都是各大求职者的最佳练兵场，而今天小编带来的便是“HUAWEI”面经！这是一次真实的面试经历，虽然不是我自己亲身经历 ... [详细]

蜡笔小新 2023-10-15 11:38:31
spring
new无法执行@Autowired注解，多线程注意事项和如何判断子线程是否全部执行完成

前言最近一段时间在整公司项目里一个功能的优化，用到了多线程处理。期间也是踩了不少的坑，在这里想说下我遇到的问题和注意事项。以及怎样知道启动的那些多线程都 ... [详细]

蜡笔小新 2023-10-15 05:36:26
spring
java程序员_9大行为导致Java程序员薪资过低，你有几个？

Java程序员薪水有高有低，有的人一个月可能拿30K、50K，有的人可能只有2K、3K。同样有五年工作 ... [详细]

蜡笔小新 2023-10-14 19:09:17
spring
阿里首席架构师科普RPC框架

RPC概念及分类RPC全称为RemoteProcedureCall，翻译过来为“远程过程调用”。目前，主流的平台中都支持各种远程调用技术，以满足分布式系统架构中不同的系统之间的远程 ... [详细]

蜡笔小新 2023-10-14 13:22:57
object
如何使用Java获取服务器硬件信息和磁盘负载率

本文介绍了使用Java编程语言获取服务器硬件信息和磁盘负载率的方法。首先在远程服务器上搭建一个支持服务端语言的HTTP服务，并获取服务器的磁盘信息，并将结果输出。然后在本地使用JS编写一个AJAX脚本，远程请求服务端的程序，得到结果并展示给用户。其中还介绍了如何提取硬盘序列号的方法。 ... [详细]

蜡笔小新 2023-12-14 13:56:20
main
在虚拟服务器上安装oracle 10g客户端的问题及解决方法

本文讨论了在VMWARE5.1的虚拟服务器Windows Server 2008R2上安装oracle 10g客户端时出现的问题，并提供了解决方法。错误日志显示了异常访问违例，通过分析日志中的问题帧，找到了解决问题的线索。文章详细介绍了解决方法，帮助读者顺利安装oracle 10g客户端。 ... [详细]

蜡笔小新 2023-12-11 13:08:10
main
Hadoop 源码学习笔记(4)Hdfs 数据读写流程分析

Hdfs的数据模型在对读写流程进行分析之前，我们需要先对Hdfs的数据模型有一个简单的认知。数据模型如上图所示，在NameNode中有一个唯一的FSDirectory类负责维护文件 ... [详细]

蜡笔小新 2023-10-17 11:27:29
main
YANG，NETCONF，RESTCONF，XML之间的关系

本文为转载，原连接：https:www.zhihu.comquestion40822826简单说一下吧做要解释这些要从netconf说起。netconf ... [详细]

蜡笔小新 2023-10-14 19:16:18
spring
springboot dubbo框架中log4j与slf4jlog4j12日志冲突问题的解决方法

在基于springboot开发项目中，使用dubbo的RPC框架进行业务拆分，出 ... [详细]

蜡笔小新 2023-10-14 18:05:31

手机用户2502927973

这个家伙很懒，什么也没留下！

Tags | 热门标签

RankList | 热门文章