【手撸RPC框架】SpringBoot+Netty4实现RPC框架_springboot rpc框架支持

【手撸RPC框架】SpringBoot+Netty4实现RPC框架

线程模型1：传统阻塞 I/O 服务模型

模型特点：

采用阻塞IO模式获取输入的数据
每个链接都需要独立的线程完成数据的输入，业务处理、数据返回。
问题分析：

当并发数很大，就会创建大量的线程，占用很大系统资源
连接创建后，如果当前线程暂时没有数据可读，该线程会阻塞在read操作，造成线程资源浪费。
线程模型2：Reactor 模式
针对传统阻塞I/O服务模型的2个缺点，解决方案如下：

基于 I/O 复用模型：多个连接共用一个阻塞对象，应用程序只需要在一个阻塞对象等待，无需阻塞等待所有连接。当某个连接有新的数据可以处理时，操作系统通知应用程序，线程从阻塞状态返回，开始进行业务处理。Reactor对应的叫法: 1. 反应器模式 2. 分发者模式(Dispatcher) 3. 通知者模式(notifier)
基于线程池复用线程资源：不必再为每个连接创建线程，将连接完成后的业务处理任务分配给线程进行处理，一个线程可以处理多个连接的业务。

单 Reactor 单线程

模型分析

优点：模型简单，没有多线程、进程通信、竞争的问题，全部都在一个线程中完成
缺点：性能问题，只有一个线程，无法完全发挥多核 CPU 的性能。Handler 在处理某个连接上的业务时，整个进程无法处理其他连接事件，很容易导致性能瓶颈

缺点：可靠性问题，线程意外终止，或者进入死循环，会导致整个系统通信模块不可用，不能接收和处理外部消息，造成节点故障
使用场景：客户端的数量有限，业务处理非常快速，比如 Redis在业务处理的时间复杂度 O(1) 的情况
单 Reactor 多线程

模型分析

优点：可以充分的利用多核cpu 的处理能力
缺点：多线程数据共享和访问比较复杂， reactor 处理所有的事件的监听和响应，在单线程运行， 在高并发场景容易出现性能瓶颈.
主从 Reactor 多线程

模型分析

优点：父线程与子线程的数据交互简单职责明确，父线程只需要接收新连接，子线程完成后续的业务处理。
优点：父线程与子线程的数据交互简单，Reactor 主线程只需要把新连接传给子线程，子线程无需返回数据
缺点：编程复杂度较高
结合实例：这种模型在许多项目中广泛使用，包括 Nginx 主从 Reactor 多进程模型，Memcached 主从多线程，Netty 主从多线程模型的支持

先实现简单的Netty通信

服务端示例

public static void main(String[] args) {
   
    //创建连接线程组，线程数为1。只负责处理连接请求
    NioEventLoopGroup boss = new NioEventLoopGroup(1);
    //创建工作线程组，线程数默认为cpu核数*2。处理与客户端的业务处理
    NioEventLoopGroup worker = new NioEventLoopGroup();
    //创建Server端的启动对象
    ServerBootstrap serverBootstrap = new ServerBootstrap();
    //配置线程组
    serverBootstrap.group(boss, worker)
        //使用 NioServerSocketChannel 作为服务器的通道实现
        .channel(NioServerSocketChannel.class)
        //给worker线程组初始化处理器
        .childHandler(new ChannelInitializer<SocketChannel>() {
   
            @Override
            protected void initChannel(SocketChannel socketChannel) throws Exception {
   
                socketChannel.pipeline()
                    //添加字符串的编解码器
                    .addLast(new StringDecoder())
                    .addLast(new StringEncoder())
                    //添加对象的编解码器，ClassResolvers.weakCachingConcurrentResolver设置弱引用WeakReferenceMap缓存类加载器，防止内存溢出
                    .addLast(new ObjectDecoder(ClassResolvers.weakCachingConcurrentResolver(this.getClass().getClassLoader())))
                    .addLast(new ObjectEncoder())
                    //添加自定义的业务处理器
                    .addLast(new SimpleChannelInboundHandler<Object>() {
   
                        @Override
                        public void channelActive(ChannelHandlerContext ctx) throws Exception {
   
                            log.info("客户端连接啦。。。客户端地址：{}", ctx.channel().remoteAddress());
                        }
                        @Override
                        protected void channelRead0(ChannelHandlerContext channelHandlerContext, Object o) throws Exception {
   
                            log.info("服务端接收到的数据：{}", o.toString());
                            //价值1个亿的AI代码
                            String str = o.toString();
                            str = str.replace("吗", "");
                            str = str.replace("？", "！");
                            str = str.replace("？ ", "！ ");
                            channelHandlerContext.writeAndFlush(str);
                        }
                    });
            }
        });
    //启动并且监听
    ChannelFuture channelFuture = serverBootstrap.bind(8888).syncUninterruptibly();
    //监听关闭通道
    channelFuture.channel().closeFuture();
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52

客户端示例

public static void main(String[] args) {
   
    //设置客户端工作线程
    NioEventLoopGroup worker = new NioEventLoopGroup();
    //创建客户端启动对象
    Bootstrap bootstrap = new Bootstrap();
    bootstrap.group(worker)
        //通道连接者
        .channel(NioSocketChannel.class)
        //给worker线程组初始化处理器
        .handler(new ChannelInitializer<SocketChannel>() {
   
            @Override
            protected void initChannel(SocketChannel socketChannel) throws Exception {
   
                socketChannel.pipeline()
                    //添加字符串的编解码器
                    .addLast(new StringDecoder())
                    .addLast(new StringEncoder())
                    //添加对象的编解码器，ClassResolvers.weakCachingConcurrentResolver设置弱引用WeakReferenceMap缓存类加载器，防止内存溢出
                    .addLast(new ObjectDecoder(ClassResolvers.weakCachingConcurrentResolver(this.getClass().getClassLoader())))
                    .addLast(new ObjectEncoder())
                    //添加自定义的业务处理器
                    .addLast(new SimpleChannelInboundHandler<Object>() {
   

                        @Override
                        public void channelActive(ChannelHandlerContext ctx) throws Exception {
   
                            ctx.writeAndFlush("哈哈哈");
                        }

                        @Override
                        protected void channelRead0(ChannelHandlerContext channelHandlerContext, Object o) throws Exception {
   
                            log.info("客户端接收到的数据：{}", o.toString());
                        }
                    });
            }
        });

    ChannelFuture channelFuture = bootstrap.connect("127.0.0.1", 8888).syncUninterruptibly();
    //客户端需要输入信息，创建一个扫描器
    Scanner scanner = new Scanner(System.in);
    while (scanner.hasNextLine()) {
   
        String msg = scanner.nextLine();
        //通过channel发送到服务器端
        channel.writeAndFlush(msg + "\r\n"<
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48