3. [netty-RPC]--服务端程序设计

亦凉 2022-07-12 04:28 155阅读 0赞

下面一步步讲解服务端程序的设计，先借用前面一节关于服务端程序设计流程图：  
![这里写图片描述][SouthEast]

# 1. 业务核心线程池的实现 #

在服务端的线程池，我自己基于JDK的ThreadPoolExecutor做了一个自我定制，实现了核心业务线程池，实现源码如下：

package com.netty.rpc.core;
    
    import java.util.concurrent.Executor;
    import java.util.concurrent.LinkedBlockingQueue;
    import java.util.concurrent.SynchronousQueue;
    import java.util.concurrent.ThreadPoolExecutor;
    import java.util.concurrent.TimeUnit;
    
    /** * 自定义的线程池 */
    public class RpcThreadPool { 
        /** * @param threads 固定数量线程的线程池 * @param queues 设置线程池工作队列, 0:同步队列 * <0 无界队列 LinkedBlockingQueue * >0 无界队列,指定了初始容量 * * @return 返回指定了线程池之后的线程池执行器 */
        public static Executor getExecutor(int threads, int queues) {
            String name = "RpcThreadPool";
            return new ThreadPoolExecutor(threads, threads, 0, TimeUnit.MILLISECONDS,
                    queues == 0 ? new SynchronousQueue<Runnable>()
                            : (queues < 0 ? new LinkedBlockingQueue<Runnable>()
                                    : new LinkedBlockingQueue<Runnable>(queues)),
                    new NamedThreadFactory(name, true), new AbortPolicyWithReport(name));
        }
    }

在这个核心业务线程池的定制策略是：创建一个固定线程数的线程池，线程池的工作队列可选无界队列或则是同步队列。此外我还自定义了线程工厂NamedThreadFactory和线程拒绝策略AbortPolicyWithReport。

在NamedThreadFactory其实就是对ThreadFactory的一个实现，在这个实现里面其实就是对Runable进行一个包装以及对线程设置一些信息和获取监控信息。下面看看实现的源码：

package com.netty.rpc.core;
    
    import java.util.concurrent.ThreadFactory;
    import java.util.concurrent.atomic.AtomicInteger;
    
    /** * @Description: 线程工厂:实际上就是对Runable进行一个包装,对线程设置一些信息和监控信息 */
    public class NamedThreadFactory implements ThreadFactory { 
    
        private static final AtomicInteger threadNumber = new AtomicInteger(1);
    
        private final AtomicInteger mThreadNum = new AtomicInteger(1);
        //前缀
        private final String prefix;
        /** 是否是后台线程, 默认false不是后台线程 */
        private final boolean daemoThread;
        /** 线程组 */
        private final ThreadGroup threadGroup;
    
        /** * 构造器 */
        public NamedThreadFactory(String prefix) {
            this(prefix, false);
        }
        /** * 构造器 */
        public NamedThreadFactory(String prefix, boolean daemo) {
            this.prefix = prefix + "-thread-";
            daemoThread = daemo;
            SecurityManager s = System.getSecurityManager();
            threadGroup = (s == null) ? Thread.currentThread().getThreadGroup() : s.getThreadGroup();
        }
    
        /** * 核心函数:新建一个线程 * @param runnable 传入一个任务task * @return */
        public Thread newThread(Runnable runnable) {
            String name = prefix + mThreadNum.getAndIncrement();
            Thread ret = new Thread(threadGroup, runnable, name, 0);
            ret.setDaemon(daemoThread);
            return ret;
        }
    }

线程拒绝策略AbortPolicyWithReport是指当线程池出现异常时的执行策略，一般情况下指工作队列满了之后又有新的线程加入队列的情况，不过对于无界队列就不存在这种情况了。线面给出实现：

package com.netty.rpc.core;
    
    import java.util.concurrent.RejectedExecutionException;
    import java.util.concurrent.ThreadPoolExecutor;
    
    /** * 服务端: 线程池异常策略 */
    public class AbortPolicyWithReport extends ThreadPoolExecutor.AbortPolicy { 
    
        private final String threadName;
    
        /** * 构造器 * @param threadName */
        public AbortPolicyWithReport(String threadName) {
            this.threadName = threadName;
        }
    
        @Override
        public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
            String msg = String.format("RpcServer["
                    + " Thread Name: %s, Pool Size: %d (active: %d, core: %d, max: %d, largest: %d), Task: %d (completed: %d),"
                    + " Executor status:(isShutdown:%s, isTerminated:%s, isTerminating:%s)]",
                    threadName, e.getPoolSize(), e.getActiveCount(), e.getCorePoolSize(), e.getMaximumPoolSize(), e.getLargestPoolSize(),
                    e.getTaskCount(), e.getCompletedTaskCount(), e.isShutdown(), e.isTerminated(), e.isTerminating());
            System.out.println(msg);
            throw new RejectedExecutionException(msg);
        }
    }

# 2. RPC服务器的启动与执行模块 #

我们的服务器是通过Spring来启动的，通过加载Spring的配置文件来配置启动netty服务器。服务器的启动程序在RpcServerJdkNativeProtocolStarter，下面看看这个类的实现：

package com.netty.rpc.core;
    
    import org.springframework.context.support.ClassPathXmlApplicationContext;
    
    /** * @Description:rpc服务器启动模块 */
    public class RpcServerJdkNativeProtocolStarter { 
    
        public static void main(String[] args) {
            /**加载Spring容器,启动服务器*/
            new ClassPathXmlApplicationContext("com/netty/rpc/config/rpc-jdknative.xml");
        }
    }

执行这个类会自动的加载配置文件并装配bean。 下面看看配置文件的配置：

<?xml version="1.0" encoding="UTF-8"?>
    <beans xmlns="http://www.springframework.org/schema/beans" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:context="http://www.springframework.org/schema/context" xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context.xsd">
    
        
        <context:component-scan base-package="com.netty.rpc.core"/>
        
        <context:property-placeholder location="classpath:com/netty/rpc/config/rpc-server-jdknative.properties"/>
    
        
        <bean id="rpcbean" class="com.netty.rpc.model.MessageKeyVal">
            
            <property name="messageKeyVal">
                
                <map>
                    <entry key="com.netty.rpc.servicebean.Calculate">
                        <ref bean="calc"/>
                    </entry>
                </map>
            </property>
        </bean>
    
    
        <bean id="calc" class="com.netty.rpc.servicebean.CalculateImpl"/>
    
        
        <bean id="rpcServer" class="com.netty.rpc.core.MessageRecvExecutor">
            
            <constructor-arg name="serverAddress" value="${rpc.server.addr}"/>
            <constructor-arg name="serializeProtocol" value="JDKSERIALIZE"/>
        </bean>
    
    </beans>

上面的配置文件中我们需要重点关注的就是

这个bean.这个bean通过构造器注入，给MessageRecvExecutor注入两个属性。

下面我们看看服务端的MessageRecvExecutor的源码

package com.netty.rpc.core;
    
    import com.google.common.util.concurrent.*;
    import io.netty.bootstrap.ServerBootstrap;
    import io.netty.channel.*;
    import io.netty.channel.nio.NioEventLoopGroup;
    import io.netty.channel.socket.nio.NioServerSocketChannel;
    import com.netty.rpc.model.*;
    import com.netty.rpc.serialize.support.RpcSerializeProtocol;
    import com.netty.rpc.utils.LogUtil;
    import org.springframework.beans.BeansException;
    import org.springframework.beans.factory.InitializingBean;
    import org.springframework.context.ApplicationContext;
    import org.springframework.context.ApplicationContextAware;
    
    import java.nio.channels.spi.SelectorProvider;
    import java.util.Iterator;
    import java.util.Map;
    import java.util.Set;
    import java.util.concurrent.Callable;
    import java.util.concurrent.ConcurrentHashMap;
    import java.util.concurrent.ThreadFactory;
    import java.util.concurrent.ThreadPoolExecutor;
    import java.util.logging.Level;
    
    /** * 服务器执行模块 */
    public class MessageRecvExecutor implements ApplicationContextAware, InitializingBean { 
    
        /** 服务端IP地址, Spring注入 */
        private String serverAddress;
        /** 服务端序列化使用的协议,默认是JDK原生序列化, Spring注入 */
        private RpcSerializeProtocol serializeProtocol = RpcSerializeProtocol.JDKSERIALIZE;
        /** 分隔符 */
        private final static String DELIMITER = ":";
        /** 服务端处理请求体映射的Map */
        private Map<String, Object> handlerMap = new ConcurrentHashMap<String, Object>();
        /** guava的线程执行器,单例 */
        private static ListeningExecutorService threadPoolExecutor;
    
        /** * 构造器 * @param serverAddress 服务器地址 * @param serializeProtocol 序列化协议 */
        public MessageRecvExecutor(String serverAddress, String serializeProtocol) {
            this.serverAddress = serverAddress;
            this.serializeProtocol = Enum.valueOf(RpcSerializeProtocol.class, serializeProtocol);
        }
    
        /** * 提交任务 * @param task * @param ctx * @param request * @param response */
        public static void submit(Callable<Boolean> task, ChannelHandlerContext ctx, MessageRequest request, MessageResponse response) {
            /** * 这里用到了 DCL 实现单例模式 */
            if (threadPoolExecutor == null) {
                synchronized (MessageRecvExecutor.class) {
                    if (threadPoolExecutor == null) {
                        //将 ExecutorService 转为 ListeningExecutorService
                        threadPoolExecutor = MoreExecutors.listeningDecorator((ThreadPoolExecutor) RpcThreadPool.getExecutor(16, -1));
                    }
                }
            }
    
            //提交任务, 异步获取结果
            ListenableFuture<Boolean> listenableFuture = threadPoolExecutor.submit(task);
    
            //注册回调函数, 在task执行完之后 异步调用回调函数
            Futures.addCallback(listenableFuture, new FutureCallback<Boolean>() {
                /** * Future成功的时候执行，根据Future结果来判断。 * * 当消息处理完成之后,再发送回客户端. * @param result */
                public void onSuccess(Boolean result) {
                    //为返回msg回客户端添加一个监听器,当消息成功发送回客户端时被异步调用.
                    ctx.writeAndFlush(response).addListener(new ChannelFutureListener() {
                        /** * 服务端回显 request已经处理完毕 * @param channelFuture * @throws Exception */
                        public void operationComplete(ChannelFuture channelFuture) throws Exception {
                            System.out.println("RPC Server Send message-id respone:" + request.getMessageId());
                        }
    
                    });
                }
    
                /** * Future失败的时候执行，根据Future结果来判断。 * @param t */
                public void onFailure(Throwable t) {
                    t.printStackTrace();
                }
            }, threadPoolExecutor);// end of Futures.addCallback()
        }
    
    /**================================================================================================================*/
    // 下面的是Spring容器启动的时候调用的函数
    /**================================================================================================================*/
        /** * 当初始化Spring容器的时候会被调用, ApplicationContextAware包装接口中的方法在这里重载, * 在afterPropertiesSet()方法之前被调用. * 主要功能: 将容器中请求与实现类的映射关系存入 ConcurrentHashMap并发容器的Map中 * @param ctx * @throws BeansException */
        @Override
        public void setApplicationContext(ApplicationContext ctx) throws BeansException {
    
            LogUtil.log_debug("setApplicationContext()");
    
            try {
                /**获取 MessageKeyVal 实例*/
                MessageKeyVal keyVal = (MessageKeyVal) ctx.getBean(Class.forName("newlandframework.netty.rpc.model.MessageKeyVal"));
                /**获取Map映射关系*/
                Map<String, Object> rpcServiceObject = keyVal.getMessageKeyVal();
    
                Set s = rpcServiceObject.entrySet();
                Iterator<Map.Entry<String, Object>> it = s.iterator();
                Map.Entry<String, Object> entry;
    
                while (it.hasNext()) {
                    entry = it.next();
                    handlerMap.put(entry.getKey(), entry.getValue());
                }
            } catch (ClassNotFoundException ex) {
                java.util.logging.Logger.getLogger(MessageRecvExecutor.class.getName()).log(Level.SEVERE, null, ex);
            }
        }
    
        /** * 重载InitializingBean里面的方法, 当他被BeanFactory调用 * 在setApplicationContext()方法之后被调用 * 主要功能: 启动netty服务端 * @throws Exception */
        @Override
        public void afterPropertiesSet() throws Exception {
            LogUtil.log_debug("执行afterPropertiesSet()");
    
            ThreadFactory threadRpcFactory = new NamedThreadFactory("NettyRPC ThreadFactory");
    
            //获取机器的CPU数量 * 2 设置并行数,我的Mac是4核的.
            int parallel = Runtime.getRuntime().availableProcessors() * 2;
            /** 主从线程池,boss线程池是一个线程, worker线程池是parallel */
            EventLoopGroup boss = new NioEventLoopGroup();
            EventLoopGroup worker = new NioEventLoopGroup(parallel, threadRpcFactory, SelectorProvider.provider());
    
            try {
                ServerBootstrap bootstrap = new ServerBootstrap();
                bootstrap.group(boss, worker).channel(NioServerSocketChannel.class)
                        .childHandler(new MessageRecvChannelInitializer(handlerMap).buildRpcSerializeProtocol(serializeProtocol))
                        .option(ChannelOption.SO_BACKLOG, 128)
                        .childOption(ChannelOption.SO_KEEPALIVE, true);
    
                String[] ipAddr = serverAddress.split(MessageRecvExecutor.DELIMITER);
    
                if (ipAddr.length == 2) {
                    //获取服务器IP地址和端口
                    String host = ipAddr[0];
                    int port = Integer.parseInt(ipAddr[1]);
                    /**异步启动服务器*/
                    ChannelFuture future = bootstrap.bind(host, port).sync();
                    System.out.printf("[author louyuting] Netty RPC Server start success!\nip:%s\nport:%d\nprotocol:%s\n\n", host, port, serializeProtocol);
                    future.channel().closeFuture().sync();
                } else {
                    System.out.printf("[author louyuting] Netty RPC Server start fail!\n");
                }
            } finally {
                worker.shutdownGracefully();
                boss.shutdownGracefully();
            }
        }
    
    }

在Spring启动时会扫描该类，因为MessageRecvExecutor 实现了 ApplicationContextAware, InitializingBean接口，并按照顺序执行setApplicationContext()函数和afterPropertiesSet()函数。

在setApplicationContext()函数中做了以下内容：将容器中请求与实现类的映射关系MessageKeyVal存入 ConcurrentHashMap并发容器的Map中，以方便后面的线程池处理消息时调用。

afterPropertiesSet()函数做了以下内容：启动RPC服务器，其实也就是netty服务端。

在MessageRecvExecutor中还有的一个static函数就是submit(）函数了，这个函数主要作用是提交业务处理任务并异步获取结果。具体的逻辑我们在后面分析。

# 3. RPC服务器的handlers #

在启动netty服务器的时候一个最重要的步骤之一就是向netty服务器的ChannelPipeline添加了各种服务端的编解码器handler。下面我们首先看看服务端的handler容器中需要有的一些handler示意图

![这里写图片描述][SouthEast 1]

我们知道在服务端主要用到了如下5个编解码器：

//入站解码器
    LengthFieldBasedFrameDecoder
    ObjectDecoder
    MessageRecvHandler
    
    //出站的编码器
    LengthFieldPrepender
    ObjectEncoder

在这些编码器中，LengthFieldBasedFrameDecoder主要是解决TCP黏包的问题。ObjectDecoder主要是对消息的反序列化，获取消息实体对象，MessageRecvHandler就是核心的对消息的处理。

LengthFieldPrepender的作用：如果协议中的第一个字段为长度字段，Netty提供了LengthFieldPrepender编码器，它可以计算当前待发送消息的二进制字节长度，将该长度添加到ByteBuf的缓冲区头中。通过LengthFieldPrepender可以将待发送消息的长度写入到ByteBuf的前2个字节，编码后的消息组成为长度字段+原消息的方式。通过设置LengthFieldPrepender构造器参数为true，消息长度将包含长度本身占用的字节数，但是默认情况下是false。

ObjectEncoder也就是对消息序列化然后发送给服务端。

# 4. 服务端核心消息处理handler：MessageRecvHandler #

在服务端接收到的消息经过解码器之后最后会到达MessageRecvHandler，这个handler也是业务处理的核心，下面看看这个handler的源码：

package com.netty.rpc.core;
    
    import com.netty.rpc.model.MessageRequest;
    import com.netty.rpc.model.MessageResponse;
    import io.netty.channel.ChannelHandlerContext;
    import io.netty.channel.ChannelInboundHandlerAdapter;
    
    import java.util.Map;
    /** * Rpc服务器消息处理 */
    public class MessageRecvHandler extends ChannelInboundHandlerAdapter { 
    
        private final Map<String, Object> handlerMap;
    
        public MessageRecvHandler(Map<String, Object> handlerMap) {
            this.handlerMap = handlerMap;
        }
    
        public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
            //这是获取到的msg已经经过JDK序列化的解码器了,只需要强转成MessageRequest
            MessageRequest request = (MessageRequest) msg;
            //new 一个 response
            MessageResponse response = new MessageResponse();
            // new 一个服务器消息处理线程
            MessageRecvInitializeTask recvTask = new MessageRecvInitializeTask(request, response, handlerMap);
            // 将服务端的处理任务提交给服务端的 消息处理线程池
            //不要阻塞nio线程，复杂的业务逻辑丢给专门的线程池
            MessageRecvExecutor.submit(recvTask, ctx, request, response);
        }
    
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
            ctx.close();
        }
    }

注意，这个类在构造的时候传入了一个参数 handlerMap，这个Map存的就是请请求和实现的映射关系。 在这个handler里面，我们可以直接获取到请求消息的实体，然后new一个响应对象。现在我们是在handler线程里面，为了避免出现请求消息的处理可能出现耗时和堵塞线程的情况，这里我们对于每一个消息的处理都创建一个线程然后丢给我自定义的线程池处理，在我自定义的线程池里面处理完成之后异步获取结果，处理完成之后异步将response返回给客户端。这里异步返回结果，只需要把每个handler中的ctx作为参数传入就行，然后用过ctx写response到netty对用的NIO通道中。

这里对于每个消息创建的线程我们用MessageRecvInitializeTask封装了，这个类其实就是实现了Callable接口，我们看看这个类的源码：

package com.netty.rpc.core;
    
    import io.netty.channel.ChannelHandlerContext;
    import com.netty.rpc.model.MessageRequest;
    import com.netty.rpc.model.MessageResponse;
    import org.apache.commons.lang3.reflect.MethodUtils;
    
    import java.util.Map;
    import java.util.concurrent.Callable;
    
    /** * Rpc服务器消息线程任务处理 */
    public class MessageRecvInitializeTask implements Callable<Boolean> { 
    
        private MessageRequest request = null;
        private MessageResponse response = null;
        private Map<String, Object> handlerMap = null;
        private ChannelHandlerContext ctx = null;
    
        public MessageResponse getResponse() {
            return response;
        }
        public MessageRequest getRequest() {
            return request;
        }
        public void setRequest(MessageRequest request) {
            this.request = request;
        }
    
        /** * 构造器 * @param request * @param response * @param handlerMap */
        MessageRecvInitializeTask(MessageRequest request, MessageResponse response, Map<String, Object> handlerMap) {
            this.request = request;
            this.response = response;
            this.handlerMap = handlerMap;
            this.ctx = ctx;
        }
    
        /** * 这个才是服务端真正的消息处理函数 * @return */
        public Boolean call() {
            response.setMessageId(request.getMessageId());
            try {
                Object result = reflect(request);
                response.setResult(result);
                return Boolean.TRUE;
            } catch (Throwable t) {
                response.setError(t.toString());
                t.printStackTrace();
                System.err.printf("RPC Server invoke error!\n");
                return Boolean.FALSE;
            }
        }
    
        /** * 反射机制, 执行请求中的方法, 并获取response中的 Object域也就是方法执行的结果 * @param request * @return * @throws Throwable */
        private Object reflect(MessageRequest request) throws Throwable {
            //获取类名
            String className = request.getClassName();
            //根据类名,获取类名在服务器映射的类信息
            Object serviceBean = handlerMap.get(className);
            //获取方法名
            String methodName = request.getMethodName();
            //获取参数值数组
            Object[] parameters = request.getParameters();
            //反射执行,获取resopnse的result
            return MethodUtils.invokeMethod(serviceBean, methodName, parameters);
        }
    
    }

类里面通过call()函数就是核心的线程执行内容，主要就是根据请求消息的信息（比如类名，方法等等），然后利用反射实例化实现对象并存如response中。

在这里每个消息的处理任务通过调用

MessageRecvExecutor.submit()

这个方法直接提交到我自己的业务处理线程池中处理。下面我们看看在这个函数中的逻辑：

/** * 提交任务,异步获取结果. * @param task * @param ctx * @param request * @param response */
    public static void submit(Callable<Boolean> task, ChannelHandlerContext ctx, MessageRequest request, MessageResponse response) {
        /** * 这里用到了 DCL 实现单例模式 */
        if (threadPoolExecutor == null) {
            synchronized (MessageRecvExecutor.class) {
                if (threadPoolExecutor == null) {
                    //将 ExecutorService 转为 ListeningExecutorService
                    threadPoolExecutor = MoreExecutors.listeningDecorator((ThreadPoolExecutor) RpcThreadPool.getExecutor(16, -1));
                }
            }
        }
    
        //提交任务, 异步获取结果
        ListenableFuture<Boolean> listenableFuture = threadPoolExecutor.submit(task);
    
        //注册回调函数, 在task执行完之后 异步调用回调函数
        Futures.addCallback(listenableFuture, new FutureCallback<Boolean>() {
            /** * Future成功的时候执行，根据Future结果来判断。 * * 当消息处理完成之后,再发送回客户端. * @param result */
            public void onSuccess(Boolean result) {
                //为返回msg回客户端添加一个监听器,当消息成功发送回客户端时被异步调用.
                ctx.writeAndFlush(response).addListener(new ChannelFutureListener() {
                    /** * 服务端回显 request已经处理完毕 * @param channelFuture * @throws Exception */
                    public void operationComplete(ChannelFuture channelFuture) throws Exception {
                        System.out.println("RPC Server Send message-id respone:" + request.getMessageId());
                    }
    
                });
            }
    
            /** * Future失败的时候执行，根据Future结果来判断。 * @param t */
            public void onFailure(Throwable t) {
                t.printStackTrace();
            }
        }, threadPoolExecutor);// end of Futures.addCallback()
    }

这里首先是一个单例模式的应用，应用DCL 实现了线程安全的threadPoolExecutor 线程池的单例。 此外这里使用了Google的guava库，guava库是对JDK的更高一层的封装，我在这里使用了guava对多线程的封装，guava的线程池执行器框架ListeningExecutorService使用更加方便，对于任务提交之后可以直接注册回调函数，在任务执行完毕之后调用回调函数，使用起来十分方便。

在这里，当消息任务处理完毕之后，直接运行回调函数，在回调函数里面将response 响应写回给客户端。

至此，整个服务端的程序的逻辑分析就是这么多。

[SouthEast]: /images/20220711/8b7244fe836f4414bbe51e330bd28d58.png
[SouthEast 1]: /images/20220711/85a6eb53618a402fa9676253af93ca47.png