基于Netty手动模拟聊天室和RPC

前面我们用了Java的NIO，但是非常的难用，尤其是ByteBuffer,几个指针很容易出错。于是有了Netty框架，Netty框架是对Java的NIO进行了封装，使Java的NIO更容易上手。

Netty框架中没有使用NIO原生的ByteBuffer,而是使用新封装的ByteBuf,下面我们看看的ByteBuf的用法。废话不多说，直接上代码。

import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;

import java.util.Arrays;

public class TestByteBuf {
    public static void main(String[] args) {
        // 1.创建一个非池化的ByteBuf，大小为10个字节
        ByteBuf buf = Unpooled.buffer(10);
        System.out.println("原始ByteBuf为====================>" + buf.toString());
        System.out.println("1.ByteBuf中的内容为===============>" + Arrays.toString(buf.array()) + "\n");

        // 2.写入一段内容
        byte[] bytes = {1, 2, 3, 4, 5};
        buf.writeBytes(bytes);
        System.out.println("写入的bytes为====================>" + Arrays.toString(bytes));
        System.out.println("写入一段内容后ByteBuf为===========>" + buf.toString());
        System.out.println("2.ByteBuf中的内容为===============>" + Arrays.toString(buf.array()) + "\n");

        // 3.读取一段内容
        byte b1 = buf.readByte();
        byte b2 = buf.readByte();
        System.out.println("读取的bytes为====================>" + Arrays.toString(new byte[]{b1, b2}));
        System.out.println("读取一段内容后ByteBuf为===========>" + buf.toString());
        System.out.println("3.ByteBuf中的内容为===============>" + Arrays.toString(buf.array()) + "\n");

        // 4.将读取的内容丢弃
        buf.discardReadBytes();
        System.out.println("将读取的内容丢弃后ByteBuf为========>" + buf.toString());
        System.out.println("4.ByteBuf中的内容为===============>" + Arrays.toString(buf.array()) + "\n");

        // 5.清空读写指针
        buf.clear();
        System.out.println("将读写指针清空后ByteBuf为==========>" + buf.toString());
        System.out.println("5.ByteBuf中的内容为===============>" + Arrays.toString(buf.array()) + "\n");

        // 6.再次写入一段内容，比第一段内容少
        byte[] bytes2 = {1, 2, 3};
        buf.writeBytes(bytes2);
        System.out.println("写入的bytes为====================>" + Arrays.toString(bytes2));
        System.out.println("写入一段内容后ByteBuf为===========>" + buf.toString());
        System.out.println("6.ByteBuf中的内容为===============>" + Arrays.toString(buf.array()) + "\n");

        // 7.将ByteBuf清零
        buf.setZero(0, buf.capacity());
        System.out.println("将内容清零后ByteBuf为==============>" + buf.toString());
        System.out.println("7.ByteBuf中的内容为================>" + Arrays.toString(buf.array()) + "\n");

        // 8.再次写入一段超过容量的内容
        byte[] bytes3 = {1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11};
        buf.writeBytes(bytes3);
        System.out.println("写入的bytes为====================>" + Arrays.toString(bytes3));
        System.out.println("写入一段内容后ByteBuf为===========>" + buf.toString());
        System.out.println("8.ByteBuf中的内容为===============>" + Arrays.toString(buf.array()) + "\n");

    }
}

运行的结果如下：

从上面的运行的结果我们可以知道：

• ByteBuf中有三个指针ridx，widx，cap，分别是读指针，写指针，总量指针。

  

• 写入bytes,只移动写的指针

  

• 读取，只移动读的指针

  

• 将读的内容丢弃，会将没有读取的内容赋值到已丢弃的位置

  

• 清空读写指针，只会将读写指针变成0，并没有将内容清空

  

• 将ByteBuf清零

  

• 写入一段超过容量的内容，可以自动扩容

上面的内容，让我们大概的了解了一下Netty中的ByteBuf，下面让我们来也写一个聊天室的案例，让我们对Netty有个简单的认识，首先我们要先写服务端。

import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioServerSocketChannel;
public class TestServer {
    public static void main(String[] args) {
        //监控客户端的连接的
        EventLoopGroup bossGroup = new NioEventLoopGroup();
        //监控客户端的读写的
        EventLoopGroup workGroup = new NioEventLoopGroup();
        try {
            //创建启动的类
            ServerBootstrap serverBootstrap = new ServerBootstrap();
            //配置一些启动的参数
            serverBootstrap.group(bossGroup, workGroup)
                    .channel(NioServerSocketChannel.class)
                    .childHandler(new TestServerInitializer());
            //绑定对应的端口
            ChannelFuture channelFuture = serverBootstrap.bind(8989).sync();
            channelFuture.channel().closeFuture().sync();
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            bossGroup.shutdownGracefully();
            workGroup.shutdownGracefully();
        }
    }
}

上面的代码创建了两个NioEventLoopGroup,一个用来监控客户端的连接，一个用来监控客户端的读写的。然后创建对应的启动类ServerBootstrap,并将上面创建好的两个线程组设置进去，注意：这儿第一个参数是监控客户端的连接，第二个参数是监控客户端读写的，然后再指定了NioServerSocketChannel为指定的通道，最后指定的参数是连接上来的客户端读写经过什么方式的处理。最后就是绑定8989端口启动对应的服务端并且以同步的方式启动。同时关闭通道的也是同步的，不然这儿从客户端读取内容还没有返回，服务端就终止了。

下面我们看下我们的类TestServerInitializer

import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.socket.SocketChannel;
import io.netty.handler.codec.DelimiterBasedFrameDecoder;
import io.netty.handler.codec.Delimiters;
import io.netty.handler.codec.string.StringDecoder;
import io.netty.handler.codec.string.StringEncoder;
import io.netty.util.CharsetUtil;
public class TestServerInitializer extends ChannelInitializer<SocketChannel> {
    @Override
    protected void initChannel(SocketChannel ch) throws Exception {
        ChannelPipeline pipeline = ch.pipeline();
        //基于分隔符的解码器
        pipeline.addLast(new DelimiterBasedFrameDecoder(4096, Delimiters.lineDelimiter()));
        pipeline.addLast(new StringDecoder(CharsetUtil.UTF_8));
        pipeline.addLast(new StringEncoder(CharsetUtil.UTF_8));
        pipeline.addLast(new TestServerHandler());
    }
}

这儿往pipeline添加了多个处理类，pipeline是一个队列，对客户端读写的数据进行对应的处理。这儿的处理类主要分成两种，一种是ChannelInboundHandlerAdapter一种是ChannelOutboundHandlerAdapter,ChannelInboundHandlerAdapter是处理读的数据，ChannelOutboundHandlerAdapter处理写的数据。具体的可以如下图

我们给pipeline加了基于分隔符的解码器，因为TCP有粘包和拆包的问题，所以这儿给它加了一个基于分隔符（以\r\n,\n进行拆包）的解码器。后面就是两个UTF-8的解码和编码器，最后一个我们自己书写的处理器类，具体看如下的代码

import io.netty.channel.Channel;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.channel.group.ChannelGroup;
import io.netty.channel.group.DefaultChannelGroup;
import io.netty.util.concurrent.GlobalEventExecutor;
public class TestServerHandler extends SimpleChannelInboundHandler<String> {
    //用来存储所有的客户端的连接
    private static ChannelGroup group = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);
    //channel读取数据
    @Override
    protected void channelRead0(ChannelHandlerContext ctx, String msg) throws Exception {
        Channel channel = ctx.channel();
        //遍历并发送数据
        group.forEach(ch->{
            //给不是自己的客户端发送
            if (ch != channel) {
                ch.writeAndFlush(channel.remoteAddress() + " : " + msg + "\r\n");
            }
        });
        //向下一个pipeline传播
        ctx.fireChannelRead(msg);
    }
    //channel 助手类（拦截器）的添加
    @Override
    public void handlerAdded(ChannelHandlerContext ctx) throws Exception {
        Channel channel = ctx.channel();
        group.writeAndFlush(channel.remoteAddress() + "加入\n");
        group.add(channel);
    }
    //channel助手类（拦截器） 移除
    @Override
    public void handlerRemoved(ChannelHandlerContext ctx) throws Exception {
        Channel channel = ctx.channel();
        group.writeAndFlush(channel.remoteAddress() + "离开\n");
    }
    //channel活跃 通道准备就绪事件
    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        Channel channel = ctx.channel();
        System.out.println(channel.remoteAddress() + "上线\n");
        System.out.println(group.size());
    }
}

上面的代码主要创建了一个ChannelGroup,用来存储所有的客户端的连接，最后在对应的生命周期调用对应的方法。到此处服务端所有的代码写完了。下面我们再看看客户端的代码。

import io.netty.bootstrap.Bootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;
import java.io.BufferedReader;
import java.io.InputStreamReader;
public class TestClient {
    public static void main(String[] args) {
        //对应的线程组
        EventLoopGroup bossGroup = new NioEventLoopGroup();
        try {
            //客户端的启动类
            Bootstrap bootstrap = new Bootstrap();
            bootstrap.group(bossGroup)
                    .channel(NioSocketChannel.class)
                    .handler(new TestClientInitializer());
            //绑定端口
            ChannelFuture channelFuture = bootstrap.connect("127.0.0.1", 8989).sync();
            //获取对应的通道
            Channel channel = channelFuture.channel();
            BufferedReader bufferedReader = new BufferedReader(new InputStreamReader(System.in));
            while (true) {
                //写入对应的内容
                channel.writeAndFlush(bufferedReader.readLine() + "\r\n");
            }
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            bossGroup.shutdownGracefully();
        }
    }
}

上面的代码和服务端的内容差不多，只不过将ServerBootstrap换成了Bootstrap,将NioServerSocketChannel换成了NioSocketChannel我们在看看TestClientInitializer

import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.socket.SocketChannel;
import io.netty.handler.codec.DelimiterBasedFrameDecoder;
import io.netty.handler.codec.Delimiters;
import io.netty.handler.codec.string.StringDecoder;
import io.netty.handler.codec.string.StringEncoder;
import io.netty.util.CharsetUtil;
public class TestClientInitializer extends ChannelInitializer<SocketChannel> {
    @Override
    protected void initChannel(SocketChannel ch) throws Exception {
        ChannelPipeline pipeline = ch.pipeline();
        //基于分隔符的解码器
        pipeline.addLast(new DelimiterBasedFrameDecoder(4096, Delimiters.lineDelimiter()));
        pipeline.addLast(new StringDecoder(CharsetUtil.UTF_8));
        pipeline.addLast(new StringEncoder(CharsetUtil.UTF_8));
        pipeline.addLast(new TestClientHandler());
    }
}

和服务端一模一样，因为客户端也存在读入和写出，所以要和服务端一一对应。最后我们再来看下TestClientHandler

import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
public class TestClientHandler extends SimpleChannelInboundHandler<String> {
    //读的时候执行，将读到的内容打印在控制台
    @Override
    protected void channelRead0(ChannelHandlerContext ctx, String msg) throws Exception {
        System.out.println(msg.trim() + "\n");
    }
    //发生异常的时候执行
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        cause.printStackTrace();
        ctx.close();
    }
}

以上就是用Netty写的聊天室，我们可以运行一下，查看结果如下

最后我们来看下Netty的工作流程，具体如下图

接下来开始我们的重头戏，就是手动模拟RPC(远程调用)，RPC的核心并不在于使用什么协议。RPC的目的是让你在本地调用远程的方法，而对你来说这个调用是透明的，你并不知道这个调用的方法是部署哪里。通过RPC能解耦服务，这才是使用RPC的真正目的。RPC的原理主要用到了动态代理模式，至于http协议，只是传输协议而已。

既然要实现远程调用，那么我们需要传输什么信息给服务器，让服务器知道我们要调用哪个类中的那个方法呢？

当然是需要类名，方法名，参数值，参数的类型，具体的代码如下

package rpc.entity;
import java.io.Serializable;
public class ClassInfo implements Serializable {
    //类名
    private String className;
   	//方法名
    private String methodName;
    //参数的值
    private Object[] args;
    //参数的类型
    private Class[] clazzType;
    public ClassInfo() {
    }
    public ClassInfo(String className, String methodName, Object[] args, Class[] clazzType) {
        this.className = className;
        this.methodName = methodName;
        this.args = args;
        this.clazzType = clazzType;
    }
    public void setClassName(String className) {
        ClassName = className;
    }
    public void setMethodName(String methodName) {
        this.methodName = methodName;
    }
    public void setArgs(Object[] args) {
        this.args = args;
    }
    public void setClazzType(Class[] clazzType) {
        this.clazzType = clazzType;
    }
    public String getClassName() {
        return ClassName;
    }
    public String getMethodName() {
        return methodName;
    }
    public Object[] getArgs() {
        return args;
    }
    public Class[] getClazzType() {
        return clazzType;
    }
}

我们走来先看客户端的书写，具体的代码如下

package rpc.client.netty;
import rpc.client.service.TestService;
public class ClientSocketNetty {
    public static void main(String[] args) {
        //通过代理对象来创建对应的对象
        TestService testService = (TestService) ClientRpcProxy.create(TestService.class);
        System.out.println(testService.listById(0));
    }
}

上面的代码通过代理类远程调用对应的方法。客户端只放了对应的接口，没有实现，而是通过对应的接口去远程调用实现类中的方法。

package rpc.client.service;
import java.util.List;
public interface TestService {
    List<String> listAll();
    String listById(Integer id);
}

上面就是对应的客户端的接口，我们再来看看我们代理类

package rpc.client.netty;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.serialization.ClassResolvers;
import io.netty.handler.codec.serialization.ObjectDecoder;
import io.netty.handler.codec.serialization.ObjectEncoder;
import rpc.entity.ClassInfo;
import java.lang.reflect.InvocationHandler;
import java.lang.reflect.Method;
import java.lang.reflect.Proxy;
public class ClientRpcProxy {
    public static Object create(Class clazz) {
        return Proxy.newProxyInstance(clazz.getClassLoader(), new Class[]{clazz}, (proxy, method, args) -> {
            ClassInfo classInfo = new ClassInfo();
            //类名
            classInfo.setClassName(clazz.getName());
            //方法名
            classInfo.setMethodName(method.getName());
            //方法参数的值
            classInfo.setArgs(args);
            //方法参数的类型
            classInfo.setClazzType(method.getParameterTypes());
		   //创建一个线程组
            EventLoopGroup eventExecutors = new NioEventLoopGroup();    
            //创建客户端启动助手  完成相关配置
            Bootstrap bootstrap = new Bootstrap();
            //创建业务处理类
            ClientSocketNettyHandler nettyClientHandler = new ClientSocketNettyHandler();
            try {
                bootstrap.group(eventExecutors)    //设置线程组
                        .channel(NioSocketChannel.class) //设置使用SocketChannel为管道通信的底层实现
                        .handler(new ChannelInitializer<SocketChannel>() {
                            @Override
                            protected void initChannel(SocketChannel socketChannel) throws Exception {
                                ChannelPipeline pipeline = socketChannel.pipeline();
                                //添加编码器
                                pipeline.addLast("encoder", new ObjectEncoder());
                                //添加解码器
                                //maxObjectSize:序列化的对象的最大长度，一旦接收到的对象长度大于此值，
                                //抛出StreamCorruptedException异常
                                //classResolver：这个类（ClassResolver）会去加载已序列化的对象，
                                //常用调用方式：ClassResolvers.cacheDisabled(Plan.class.getClassLoader())
                                //或者直接ClassResolvers.cacheDisabled(null)
                                pipeline.addLast("decoder", new ObjectDecoder(
                                    Integer.MAX_VALUE, ClassResolvers.cacheDisabled(null)));
                                //将自己编写的客户端业务逻辑处理类加入到pipeline链中
                                pipeline.addLast(nettyClientHandler);
                            }
                        });
                System.out.println("......client init......");
                //设置服务端的ip和端口  异步非阻塞
                //connect方法是异步的sync方法是同步的
                ChannelFuture future = bootstrap.connect("127.0.0.1", 9091).sync();  
                future.channel().writeAndFlush(classInfo).sync();
                //关闭连接  异步非阻塞
                future.channel().closeFuture().sync();
            } catch (Exception e) {
                e.printStackTrace();
            }
            return nettyClientHandler.getResponse();
        });
    }
}

上面的代码和原来的我们写的聊天室的代码的差不多，唯一不同的是我们使用pipeline不同。下面我们再来看下我们自己数据的处理类

package rpc.client.netty;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
public class ClientSocketNettyHandler extends ChannelInboundHandlerAdapter {
    private Object response;
    public Object getResponse() {
        return response;
    }
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        //远程调用的结果赋值给response
        this.response = msg;
        ctx.close();
    }
}

看完了客户端，我们再来看看服务端接受的这些该如何处理。首先我们要先创建对应的接口和实现，方便给客户端调用，具体的代码如下：

package rpc.server.service;
import java.util.List;
public interface TestService {
    List<String> listAll();
    String listById(Integer id);
}

package rpc.server.service.impl;
import rpc.server.service.TestService;
import java.util.ArrayList;
import java.util.List;
public class TestServiceImpl implements TestService {
    static ArrayList<String> list = new ArrayList<>();
    static {
        list.add("张三");
        list.add("李四");
    }
    @Override
    public List<String> listAll() {
        return list;
    }
    @Override
    public String listById(Integer id) {
        return list.get(id);
    }
}

看完了接口和实现类，我们在看看服务端是怎么写的

package rpc.server.netty;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.codec.serialization.ClassResolvers;
import io.netty.handler.codec.serialization.ObjectDecoder;
import io.netty.handler.codec.serialization.ObjectEncoder;
public class ServerSocketNetty {
    public static void main(String[] args) {
        //创建一个线程组：接受客户端连接
        EventLoopGroup bossGroup = new NioEventLoopGroup();
        //创建一个线程组：接受网络操作
        EventLoopGroup workerGroup = new NioEventLoopGroup();
        //创建服务器启动助手来配置参数
        ServerBootstrap serverBootstrap = new ServerBootstrap();
        try{
            serverBootstrap.group(bossGroup,workerGroup)//设置两个线程组
            .channel(NioServerSocketChannel.class)//设置使用NioServerSocketChannel作为服务器通道的实现
            .option(ChannelOption.SO_BACKLOG,128)//设置线程队列中等待连接的个数
            .childOption(ChannelOption.SO_KEEPALIVE,true)//保持活动连接状态
            .childHandler(new ChannelInitializer<SocketChannel>() {//创建一个初始化管道对象
                @Override
                protected void initChannel(SocketChannel ch) throws Exception {
                    ChannelPipeline pipeline = ch.pipeline();
                    //添加编码器
                    pipeline.addLast("encoder", new ObjectEncoder());
                    //添加解码器
                    pipeline.addLast("decoder",new ObjectDecoder(
                        Integer.MAX_VALUE, ClassResolvers.cacheDisabled(null)));
                    //将自己编写的服务器端的业务逻辑处理类加入pipeline链中
                    pipeline.addLast(ServerSocketNettyHandler.serverSocketNettyHandler);
                }
            });
            System.out.println(".........server  init..........");
            ChannelFuture future = serverBootstrap.bind(9091).sync();//设置端口  非阻塞
            System.out.println(".........server start..........");
            //关闭通道  关闭线程组  非阻塞
            future.channel().closeFuture().sync();
        }catch (Exception e){
            e.printStackTrace();
        }finally {
            bossGroup.shutdownGracefully();
            workerGroup.shutdownGracefully();
        }
    }
}

和客户端添加的处理器差不多，唯一的区别就是加了一个自己的处理器，也是最关键的东西，将客户端发来的信息进行相应的处理，决定调用哪个对应的类的对应方法。具体代码如下

package rpc.server.netty;
import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import org.reflections.Reflections;
import rpc.entity.ClassInfo;
import java.lang.reflect.Method;
import java.util.Set;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
@ChannelHandler.Sharable
public class ServerSocketNettyHandler extends ChannelInboundHandlerAdapter {
    public static ServerSocketNettyHandler serverSocketNettyHandler = new ServerSocketNettyHandler();
    private static ExecutorService executorService = Executors.newFixedThreadPool(1000);
    //得到某个接口下的实现类
    public String getImpClassName(ClassInfo classInfo) throws Exception {
        //服务器接口与实现类地址
        String iName = "rpc.server.service";
        int i = classInfo.getClassName().lastIndexOf(".");
        //获取客户端传过来的类名
        String className = classInfo.getClassName().substring(i);
        //拼成服务端的全类名
        Class aClass = Class.forName(iName + className);
        //通过对应的工具类找到对应的实现
        Reflections reflections = new Reflections(iName);
        Set<Class<?>> classes = reflections.getSubTypesOf(aClass);
        if (classes.size() == 0) {
            System.out.println("未找到实现类");
            return null;
        } else if (classes.size() > 1) {
            System.out.println("找到多个实现类，未明确使用哪个实现类");
            return null;
        } else {
            Class[] classes1 = classes.toArray(new Class[0]);
            //返回对应实现的类名
            return classes1[0].getName();
        }
    }

    //读取数据事件
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        executorService.submit(() -> {
            try {
                ClassInfo classInfo = (ClassInfo) msg;
                //根据上面的方法返回的类名通过反射创建对应的类
                Object o = Class.forName(getImpClassName(classInfo)).newInstance();
                //获取对应的方法
                Method method = o.getClass().getMethod(classInfo.getMethodName(), classInfo.getClazzType());
                //调用对应的方法
                Object invoke = method.invoke(o, classInfo.getArgs());
                //将返回的结果写给客户端
                ctx.channel().writeAndFlush(invoke);
            } catch (Exception e) {
                e.printStackTrace();
            }
        });
    }
}

上面的代码将客户端传过来的信息，进行一系列的处理，然后调用对应的实现类的方法，然后将结果返回给客户端，就实现了一个简单的远程调用，上面的Reflections是通过下面的maven引入的

<dependency>
	<groupId>org.reflections</groupId>
	<artifactId>reflections</artifactId>
	<version>0.9.10</version>
</dependency>

最后我们再来看下运行效果

可以看到我们远程调用结果获取到了。

本文地址：https://blog.csdn.net/qq_36434742/article/details/107692667

如对本文有疑问， 点击进行留言回复！！