<repo_name>paranoidq/simple-rpc<file_sep>/rpc-v1/src/main/java/me/test/RpcTest.java
package me.test;

import me.rpc.core.MessageSendExecutor;
import me.rpc.core.RpcSerializeProtocol;
import org.apache.commons.lang3.time.StopWatch;

import java.util.concurrent.CountDownLatch;

/**
 * @author paranoidq
 * @since 1.0.0
 */
public class RpcTest {

    public static void main(String[] args) throws InterruptedException {
        final MessageSendExecutor executor = new MessageSendExecutor("127.0.0.1:18888", RpcSerializeProtocol.JDK_SERIALIZE);
        int parallel = 1000;

        StopWatch sw = new StopWatch();
        sw.start();

        CountDownLatch signal = new CountDownLatch(1);
        CountDownLatch finish = new CountDownLatch(parallel);

        for (int index = 0; index < parallel; index++) {
            CalcParallelRequestThread client = new CalcParallelRequestThread(
                signal, finish, executor, index
            );
            new Thread(client).start();
        }

        signal.countDown();
        finish.await();

        sw.stop();

        String tip  = String.format("RPC调用总耗时：[%s] 毫秒", sw.getTime());
        System.out.println(tip);

        executor.stop();
    }
}
<file_sep>/rpc-v1/src/main/java/me/rpc/core/MessageSendExecutor.java
package me.rpc.core;

import com.google.common.reflect.Reflection;

/**
 * @author paranoidq
 * @since 1.0.0
 */
public class MessageSendExecutor {

    private RpcServerLoader loader = RpcServerLoader.getInstance();

    public MessageSendExecutor(String serverAddress, RpcSerializeProtocol serializeProtocol) {
        loader.load(serverAddress, serializeProtocol);
    }

    public void stop() {
        loader.unload();
    }

    public static <T> T execute(Class<T> rpcInterface) {
        return Reflection.newProxy(rpcInterface, new MessageSendProxy(rpcInterface));
    }
}
<file_sep>/rpc-v2/src/main/java/me/framework/RpcServerBoot.java
package me.framework;

import org.springframework.context.support.ClassPathXmlApplicationContext;

/**
 * @author paranoidq
 * @since 1.0.0
 */
public class RpcServerBoot {

    public static void main(String[] args) {
        new ClassPathXmlApplicationContext("classpath:rpc-invoke-config-server.xml");
    }
}
<file_sep>/rpc-v1/src/main/java/me/rpc/pool/policy/BlockingPolicy.java
package me.rpc.pool.policy;

import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.RejectedExecutionHandler;
import java.util.concurrent.ThreadPoolExecutor;

/**
 * @author paranoidq
 * @since 1.0.0
 */
public class BlockingPolicy implements RejectedExecutionHandler {
    private static final Logger logger = LoggerFactory.getLogger(BlockingPolicy.class);

    private final String threadName;

    public BlockingPolicy(String threadName) {
        this.threadName = threadName;
    }

    public BlockingPolicy() {
        this(null);
    }

    /**
     * Method that may be invoked by a {@link ThreadPoolExecutor} when
     * {@link ThreadPoolExecutor#execute execute} cannot accept a
     * task.  This may occur when no more threads or queue slots are
     * available because their bounds would be exceeded, or upon
     * shutdown of the Executor.
     * <p>
     * <p>In the absence of other alternatives, the method may throw
     * an unchecked {@link RejectedExecutionException}, which will be
     * propagated to the caller of {@code execute}.
     *
     * @param r        the runnable task requested to be executed
     * @param executor the executor attempting to execute this task
     * @throws RejectedExecutionException if there is no remedy
     */
    @Override
    public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
        if (!StringUtils.isEmpty(threadName)) {
            logger.error("RPC thread pool [{}] is exhausted, executor={}", threadName, executor.toString());
        }

        if (!executor.isShutdown()) {
            try {
                executor.getQueue().put(r);
            } catch (InterruptedException e) {

            }
        }
    }
}
<file_sep>/rpc-v3/src/main/java/me/framework/rpc/message/kryo/KryoEncoder.java
package me.framework.rpc.message.kryo;

import me.framework.rpc.message.MessageDecoder;
import me.framework.rpc.message.MessageEncoder;
import me.framework.rpc.serialize.support.MessageCodec;

/**
 * @author paranoidq
 * @since 1.0.0
 */
public class KryoEncoder extends MessageEncoder{

    public KryoEncoder(MessageCodec codec) {
        super(codec);
    }
}
<file_sep>/README.md
# NettyRPC Project
high performance java rpc server base on netty framework,using kryo,hessian,protostuff support rpc message serialization.

注：代码基于唐杰的博客和Github进行修改优化，感谢原作者！
http://www.cnblogs.com/jietang/

----------
## NettyRPC 3.0 Build 2017/09 by Qianwei<file_sep>/rpc-v2/src/main/java/me/framework/rpc/core/client/RpcSendSerializeBinder.java
package me.framework.rpc.core.client;

import io.netty.channel.ChannelPipeline;
import io.netty.handler.codec.LengthFieldBasedFrameDecoder;
import io.netty.handler.codec.LengthFieldPrepender;
import io.netty.handler.codec.serialization.ClassResolvers;
import io.netty.handler.codec.serialization.ObjectDecoder;
import io.netty.handler.codec.serialization.ObjectEncoder;
import io.netty.handler.logging.LoggingHandler;
import me.framework.rpc.message.MessageSerializeBinder;
import me.framework.rpc.message.kryo.KryoDecoder;
import me.framework.rpc.message.kryo.KryoEncoder;
import me.framework.rpc.serialize.support.MessageCodec;
import me.framework.rpc.serialize.support.RpcSerializeProtocol;
import me.framework.rpc.serialize.support.kryo.KryoMessageCodec;
import me.framework.rpc.serialize.support.kryo.KryoPoolFactory;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * @author paranoidq
 * @since 1.0.0
 */
public class RpcSendSerializeBinder implements MessageSerializeBinder {
    private static final Logger logger = LoggerFactory.getLogger(RpcSendSerializeBinder.class);

    public RpcSendSerializeBinder() {}

    @Override
    public void bind(RpcSerializeProtocol protocol, ChannelPipeline pipeline) {
        switch (protocol) {
            case JDK_SERIALIZE: {
                logger.info("Use JDK_SERIALIZE");
                pipeline.addLast(new LengthFieldBasedFrameDecoder(Integer.MAX_VALUE, 0, MessageCodec.MESSAGE_LENGTH_BYTES, 0, MessageCodec.MESSAGE_LENGTH_BYTES));
                pipeline.addLast(new LengthFieldPrepender(MessageCodec.MESSAGE_LENGTH_BYTES));
                pipeline.addLast(new ObjectEncoder());
                pipeline.addLast(new ObjectDecoder(Integer.MAX_VALUE, ClassResolvers.weakCachingConcurrentResolver(this.getClass().getClassLoader())));
                pipeline.addLast(new MessageSendHandler());
                break;
            }
            case KRYO_SERIAZLIZE: {
                logger.info("Use KRYO_SERIAZLIZE");
                KryoMessageCodec util = new KryoMessageCodec();
                pipeline.addLast(new LoggingHandler());
                pipeline.addLast(new KryoEncoder(util));
                pipeline.addLast(new KryoDecoder(util));
                pipeline.addLast(new MessageSendHandler());
                break;
            }
            case HESSIAN_SERIALIZE: {
                logger.info("Use HESSIAN_SERIALIZE");
//                HessianCodecUtil util = new HessianCodecUtil();
//                pipeline.addLast(new HessianEncoder(util));
//                pipeline.addLast(new HessianDecoder(util));
//                pipeline.addLast(new MessageRecvHandler(handlerMap));
                break;
            }
        }
    }
}
<file_sep>/rpc-v3/src/main/java/me/framework/rpc/util/pool/NamedThreadFactory.java
package me.framework.rpc.util.pool;

import java.util.concurrent.ThreadFactory;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * @author paranoidq
 * @since 1.0.0
 */
public class NamedThreadFactory implements ThreadFactory {

    private static final AtomicInteger threadNumber = new AtomicInteger(1);

    private final AtomicInteger mThreadNum = new AtomicInteger(1);

    private final String prefix;

    private final boolean daemonThread;

    private final ThreadGroup threadGroup;


    public NamedThreadFactory() {
        this("srpc-thread-pool-" + threadNumber.getAndIncrement(), false);
    }

    public NamedThreadFactory(String prefix) {
        this(prefix, false);
    }

    public NamedThreadFactory(String prefix, boolean daemonThread) {
        this.prefix = prefix + "-thread-";
        this.daemonThread = daemonThread;
        SecurityManager s = System.getSecurityManager();
        threadGroup = (s == null) ? Thread.currentThread().getThreadGroup() : s.getThreadGroup();
    }


    /**
     * Constructs a new {@code Thread}.  Implementations may also initialize
     * priority, name, daemon status, {@code ThreadGroup}, etc.
     *
     * @param r a runnable to be executed by new thread instance
     * @return constructed thread, or {@code null} if the request to
     * create a thread is rejected
     */
    public Thread newThread(Runnable r) {
        String name = prefix + mThreadNum.getAndIncrement();
        Thread thread = new Thread(threadGroup, r, name, 0); // ?? what is stack size
        thread.setDaemon(daemonThread);
        return thread;
    }

    public ThreadGroup getThreadGroup() {
        return this.threadGroup;
    }
}
<file_sep>/rpc-v3/src/main/java/me/framework/rpc/core/server/MessageRecvExecutor.java
package me.framework.rpc.core.server;

import com.google.common.util.concurrent.*;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import me.framework.rpc.config.RpcSystemConfig;
import me.framework.rpc.model.MessageRequest;
import me.framework.rpc.model.MessageResponse;
import me.framework.rpc.model.ServiceHolder;
import me.framework.rpc.serialize.support.RpcSerializeProtocol;
import me.framework.rpc.spring.NettyRpcRegistry;
import me.framework.rpc.util.nettybuilder.NettyServerBootstrapBuilder;
import me.framework.rpc.util.pool.NamedThreadFactory;
import me.framework.rpc.util.pool.RpcThreadPool;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.BeansException;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;

import java.nio.channels.spi.SelectorProvider;
import java.util.Map;
import java.util.concurrent.Callable;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;

/**
 *
 * v3 版本引入了spring之后，不在需要通过手动调用start进行启动了
 *
 * 启动在{@link NettyRpcRegistry#afterPropertiesSet()}中进行
 * 该函数会首先从配置文件中获取对应的参数，并注入到{@link MessageRecvExecutor}实例中，
 * 然后调用{@link MessageRecvExecutor#start()}启动线程池，坚挺即将到来的请求
 *
 * {@link me.framework.rpc.core.client.MessageSendExecutor}同理
 *
 * @author paranoidq
 * @since 1.0.0
 */
public class MessageRecvExecutor implements ApplicationContextAware {
    private static final Logger logger = LoggerFactory.getLogger(MessageRecvExecutor.class);
    private final static String DELIMITER = ":";

    /**
     * 存储RPC服务映射关系
     * <ClassName --> Object Instance>
     */
    private Map<String, Object> handlerMap = new ConcurrentHashMap<>();

    /**
     * 通过spring注入方式设置具体值
     * 参见：{@link me.framework.rpc.spring.NettyRpcRegistry}
     */
    private String serverAddress;
    /**
     * 通过spring注入方式设置具体值
     * 参见：{@link me.framework.rpc.spring.NettyRpcRegistry}
     */
    private RpcSerializeProtocol serializeProtocol = RpcSerializeProtocol.JDK_SERIALIZE;

    /**
     * 通过spring注入方式设置具体值
     * 参见：{@link me.framework.rpc.spring.NettyRpcRegistry}
     */
    private String echoApiPort;

    /**
     * 并发执行executor
     */
    private volatile ListeningExecutorService threadPoolExecutor;

    /**
     * 线程池线程数
     */
    private static int threadNums = RpcSystemConfig.SYSTEM_PROPERTY_THREADPOOL_THREAD_NUMS;

    /**
     * 线程池队列长度
     */
    private static int queueNums = RpcSystemConfig.SYSTEM_PROPERTY_THREADPOOL_QUEUE_NUMS;

    private ThreadFactory threadFactory = new NamedThreadFactory("Netty RPC Factory");
    private int parallel = Runtime.getRuntime().availableProcessors() * 2;
    private EventLoopGroup boss = new NioEventLoopGroup();
    private EventLoopGroup worker = new NioEventLoopGroup(parallel,threadFactory, SelectorProvider.provider());


    public MessageRecvExecutor() {
        handlerMap.clear();
    }

    private static class Holder {
        private static final MessageRecvExecutor instance = new MessageRecvExecutor();
    }
    public static MessageRecvExecutor getInstance() {
        return Holder.instance;
    }

    /**
     * 提交请求去进行业务处理
     * @param task
     * @param ctx
     * @param request
     * @param response
     */
    public void submit(Callable<Boolean> task, ChannelHandlerContext ctx, MessageRequest request, MessageResponse response) {
        if (threadPoolExecutor == null) {
            synchronized (MessageRecvExecutor.class) {
                if (threadPoolExecutor == null) {
                   threadPoolExecutor =
                       MoreExecutors.listeningDecorator((ThreadPoolExecutor)
                           (RpcSystemConfig.isMonitorServerSupport()
                               ? RpcThreadPool.getExecutorWithJmx(threadNums, queueNums)
                               : RpcThreadPool.getExecutor(threadNums, queueNums)
                           )
                       );
                }
            }
        }
        ListenableFuture<Boolean> listenableFuture = threadPoolExecutor.submit(task);

        // rpc-v2通过listeningFuture移除了v1中callback，更加简洁、高效
        Futures.addCallback(listenableFuture, new FutureCallback<Boolean>() {
            @Override
            public void onSuccess(Boolean result) {
                ctx.writeAndFlush(response).addListener(new ChannelFutureListener() {
                    @Override
                    public void operationComplete(ChannelFuture future) throws Exception {
                        logger.info("RPC Server send response. messageId=[{}]", request.getMessageId());
                    }
                });
            }
            @Override
            public void onFailure(Throwable t) {
                logger.error("", t);
            }
        }, threadPoolExecutor);
    }



    @Override
    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
        try {
            ServiceHolder holder = (ServiceHolder) applicationContext.getBean(Class.forName(
                "me.framework.rpc.model.ServiceHolder"
            ));
            Map<String, Object> services = holder.getServices();
            for (Map.Entry<String, Object> service : services.entrySet()) {
                handlerMap.put(service.getKey(), service.getValue());
            }
        } catch (ClassNotFoundException e) {
            logger.error("未找到ServiceHolder类", e);
        }
    }

    public void start() {
        try {
            ServerBootstrap bootstrap = NettyServerBootstrapBuilder.getInstance(boss, worker)
                .setSoBacklog(128)
                .build();
            bootstrap.childHandler(new MessageRecvChannelInitializer(handlerMap)
                .setSerializeProtocol(serializeProtocol));

            String[] ipAddr = serverAddress.split(MessageRecvExecutor.DELIMITER);
            if (ipAddr.length == 2) {
                String host = ipAddr[0];
                int port = Integer.parseInt(ipAddr[1]);
                ChannelFuture future = bootstrap.bind(host, port).sync();

                future.addListener(new ChannelFutureListener() {
                    @Override
                    public void operationComplete(final ChannelFuture future) throws Exception {
                        if (future.isSuccess()) {
                            System.out.printf("Netty RPC Server started success ip:%s port:%d\n", host, port);
//                            future.channel().closeFuture().sync().addListener(new ChannelFutureListener() {
//                                @Override
//                                public void operationComplete(ChannelFuture future) throws Exception {
//                                    threadPoolExecutor.shutdown();
//                                }
//                            });
                        }
                    }
                });
            } else {
                logger.error("Netty RPC Server started fail!\n");
            }
        } catch (InterruptedException e) {
            logger.error("Netty RPC Server started fail!\n", e);
        }
    }


    public void stop() {
        worker.shutdownGracefully();
        boss.shutdownGracefully();
    }

    public Map<String, Object> getHandlerMap() {
        return handlerMap;
    }

    public void setHandlerMap(Map<String, Object> handlerMap) {
        this.handlerMap = handlerMap;
    }

    public String getServerAddress() {
        return serverAddress;
    }

    public void setServerAddress(String serverAddress) {
        this.serverAddress = serverAddress;
    }

    public RpcSerializeProtocol getSerializeProtocol() {
        return serializeProtocol;
    }

    public void setSerializeProtocol(RpcSerializeProtocol serializeProtocol) {
        this.serializeProtocol = serializeProtocol;
    }

    public String getEchoApiPort() {
        return echoApiPort;
    }

    public void setEchoApiPort(String echoApiPort) {
        this.echoApiPort = echoApiPort;
    }
}
<file_sep>/rpc-v1/src/main/java/me/rpc/model/ServiceHolder.java
package me.rpc.model;

import java.util.Map;

/**
 * @author paranoidq
 * @since 1.0.0
 */
public class ServiceHolder {

    private Map<String, Object> services;

    public Map<String, Object> getServices() {
        return services;
    }

    public void setServices(Map<String, Object> services) {
        this.services = services;
    }
}
<file_sep>/rpc-v2/src/main/java/me/framework/rpc/serialize/support/kryo/KryoMessageCodec.java
package me.framework.rpc.serialize.support.kryo;

import com.esotericsoftware.kryo.pool.KryoPool;
import io.netty.buffer.ByteBuf;
import me.framework.rpc.serialize.support.MessageCodec;

import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.IOException;

/**
 * @author paranoidq
 * @since 1.0.0
 */
public class KryoMessageCodec implements MessageCodec {

    public KryoMessageCodec() {}

    @Override
    public void encode(ByteBuf writeBuffer, Object message) throws IOException {
        try (ByteArrayOutputStream byteArrayOutputStream = new ByteArrayOutputStream()) {
            KryoSerializer serializer = new KryoSerializer();
            serializer.serialize(byteArrayOutputStream, message);
            byte[] body = byteArrayOutputStream.toByteArray();
            int dataLength = body.length;
            writeBuffer.writeInt(dataLength);
            writeBuffer.writeBytes(body);
        }
    }

    @Override
    public Object decode(byte[] body) throws IOException {
        try (ByteArrayInputStream byteArrayInputStream = new ByteArrayInputStream(body)) {
            KryoSerializer serializer = new KryoSerializer();
            return serializer.deserialize(byteArrayInputStream);
        }
    }
}
<file_sep>/rpc-v3/src/main/java/me/framework/rpc/filter/support/SimpleFilter.java
package me.framework.rpc.filter.support;

import me.framework.rpc.filter.Filter;
import me.framework.rpc.logger.AppLoggerInject;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;

import java.lang.reflect.Method;

/**
 * @author paranoidq
 * @since 1.0.0
 */
public class SimpleFilter implements Filter {
    @AppLoggerInject
    private static Logger logger;

    @Override
    public boolean before(Method method, Object processor, Object[] requestObjects) {
        logger.info(StringUtils.center("[SimpleFilter##before]", 48, "*"));
        return true;
    }

    @Override
    public void after(Method method, Object processor, Object[] requestObjects) {
        logger.info(StringUtils.center("[SimpleFilter##after]", 48, "*"));
    }
}
<file_sep>/rpc-v3/src/main/java/me/framework/rpc/core/handler/NettyRpcRecvHandler.java
package me.framework.rpc.core.handler;

import io.netty.channel.ChannelPipeline;

import java.util.Map;

/**
 * @author paranoidq
 * @since 1.0.0
 */
public interface NettyRpcRecvHandler {

    void handle(Map<String, Object> handlerMap, ChannelPipeline pipeline);
}
<file_sep>/rpc-v3/src/main/resources/rpc-server.properties
rpc.server.addr=127.0.0.1:18888
rpc.server.echo.api.port=18889<file_sep>/rpc-v2/src/main/java/me/framework/rpc/core/server/MessageRecvExecutor.java
package me.framework.rpc.core.server;

import com.google.common.util.concurrent.*;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import me.framework.rpc.model.MessageRequest;
import me.framework.rpc.model.MessageResponse;
import me.framework.rpc.model.ServiceHolder;
import me.framework.rpc.serialize.support.RpcSerializeProtocol;
import me.framework.rpc.util.nettybuilder.NettyServerBootstrapBuilder;
import me.framework.rpc.util.pool.NamedThreadFactory;
import me.framework.rpc.util.pool.RpcThreadPool;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
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.Map;
import java.util.concurrent.Callable;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;

/**
 * @author paranoidq
 * @since 1.0.0
 */
public class MessageRecvExecutor implements ApplicationContextAware, InitializingBean {
    private static final Logger logger = LoggerFactory.getLogger(MessageRecvExecutor.class);

    private final static String DELIMITER = ":";

    private static Map<String, Object> handlerMap = new ConcurrentHashMap<>();
    private static volatile ListeningExecutorService threadPoolExecutor;

    private String serverAddress;
    // 默认JDK序列化方式
    private RpcSerializeProtocol protocol = RpcSerializeProtocol.JDK_SERIALIZE;


    public MessageRecvExecutor(String serverAddress, String protocol) {
        this.serverAddress = serverAddress;
        this.protocol = Enum.valueOf(RpcSerializeProtocol.class, protocol);
    }


    /**
     * 提交请求去进行业务处理
     * @param task
     * @param ctx
     * @param request
     * @param response
     */
    public static void submit(Callable<Boolean> task, ChannelHandlerContext ctx, MessageRequest request, MessageResponse response) {
        if (threadPoolExecutor == null) {
            synchronized (MessageRecvExecutor.class) {
                if (threadPoolExecutor == null) {
                   threadPoolExecutor = MoreExecutors.listeningDecorator((ThreadPoolExecutor) RpcThreadPool.getExecutor(16, -1));
                }
            }
        }
        ListenableFuture<Boolean> listenableFuture = threadPoolExecutor.submit(task);

        // rpc-v2通过listeningFuture移除了v1中callback，更加简洁、高效
        Futures.addCallback(listenableFuture, new FutureCallback<Boolean>() {
            @Override
            public void onSuccess(Boolean result) {
                ctx.writeAndFlush(response).addListener(new ChannelFutureListener() {
                    @Override
                    public void operationComplete(ChannelFuture future) throws Exception {
                        logger.info("RPC Server send response. messageId=[{}]", request.getMessageId());
                    }
                });
            }

            @Override
            public void onFailure(Throwable t) {
                logger.error("", t);
            }
        }, threadPoolExecutor);
    }

    
    @Override
    public void afterPropertiesSet() throws Exception {
        ThreadFactory threadFactory = new NamedThreadFactory("Netty RPC Factory");
        int parallel = Runtime.getRuntime().availableProcessors() * 2;
        EventLoopGroup boss = new NioEventLoopGroup();
        EventLoopGroup worker = new NioEventLoopGroup(parallel,threadFactory, SelectorProvider.provider());

        try {
            ServerBootstrap bootstrap = NettyServerBootstrapBuilder.getInstance(boss, worker)
                .setAcceptSocketsMax(128)
                .build();
            bootstrap.childHandler(new MessageRecvChannelInitializer(handlerMap)
                .setSerializeProtocol(protocol));

            String[] ipAddr = serverAddress.split(MessageRecvExecutor.DELIMITER);
            if (ipAddr.length == 2) {
                String host = ipAddr[0];
                int port = Integer.parseInt(ipAddr[1]);
                ChannelFuture future = bootstrap.bind(host, port);
                System.out.printf("Netty RPC Server started success ip:%s port:%d\n", host, port);
                future.channel().closeFuture().sync();
            } else {
                System.out.printf("Netty RPC Server started fail!\n");
            }
        } finally {
            worker.shutdownGracefully();
            boss.shutdownGracefully();
        }
    }


    @Override
    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
        try {
            ServiceHolder holder = (ServiceHolder) applicationContext.getBean(Class.forName(
                "me.framework.rpc.model.ServiceHolder"
            ));
            Map<String, Object> services = holder.getServices();
            for (Map.Entry<String, Object> service : services.entrySet()) {
                handlerMap.put(service.getKey(), service.getValue());
            }
        } catch (ClassNotFoundException e) {
            logger.error("", e);
        }
    }
}
<file_sep>/rpc-v1/src/main/java/me/rpc/core/RpcServerLoader.java
package me.rpc.core;

import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import me.rpc.pool.RpcThreadPool;

import java.net.InetSocketAddress;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

/**
 * @author paranoidq
 * @since 1.0.0
 */
public class RpcServerLoader {

    private static volatile RpcServerLoader instance;

    private final static String DELIMITER = ":";

    private RpcSerializeProtocol serializeProtocol = RpcSerializeProtocol.JDK_SERIALIZE;

    private final static int parallel = Runtime.getRuntime().availableProcessors() * 2;
    private EventLoopGroup eventLoopGroup = new NioEventLoopGroup(parallel);
    private static ThreadPoolExecutor threadPoolExecutor = (ThreadPoolExecutor) RpcThreadPool.getExecutor(16, -1);
    private MessageSendHandler sendHandler;

    private Lock lock = new ReentrantLock();
    private Condition signal = lock.newCondition();

    public static RpcServerLoader getInstance() {
        if (instance == null) {
            synchronized (RpcServerLoader.class) {
                if (instance == null) {
                    instance = new RpcServerLoader();
                }
            }
        }
        return instance;
    }

    public void load(String serverAddress, RpcSerializeProtocol serializeProtocol) {
        String[] ipAddr = serverAddress.split(DELIMITER);
        if (ipAddr.length == 2) {
             String host = ipAddr[0];
            int port = Integer.parseInt(ipAddr[1]);
            final InetSocketAddress remoteAddr = new InetSocketAddress(host, port);
            threadPoolExecutor.submit(
                new MessageSendInitializeTask(eventLoopGroup, remoteAddr, this)
            );
        }
    }

    public void unload() {
        sendHandler.close();
        threadPoolExecutor.shutdown();
        eventLoopGroup.shutdownGracefully();
    }


    public void setMessageSendHandler(MessageSendHandler sendHandler) {
        try {
            lock.lock();
            this.sendHandler = sendHandler;
            signal.signalAll();
        } finally {
            lock.unlock();
        }
    }


    public MessageSendHandler getSendHandler() throws InterruptedException {
        try {
            lock.lock();
            if (sendHandler == null) {
                signal.await();
            }
            return sendHandler;
        } finally {
            lock.unlock();
        }
    }

    public void setSerializeProtocol(RpcSerializeProtocol serializeProtocol) {
        this.serializeProtocol = serializeProtocol;
    }
}
<file_sep>/rpc-v3/src/main/java/me/framework/rpc/util/pool/policy/AbortPolicy.java
package me.framework.rpc.util.pool.policy;

import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.ThreadPoolExecutor;

/**
 * @author paranoidq
 * @since 1.0.0
 */
public class AbortPolicy extends ThreadPoolExecutor.AbortPolicy {
    private static final Logger logger = LoggerFactory.getLogger(AbortPolicy.class);
    private final String threadName;

    public AbortPolicy(String threadName) {
        this.threadName = threadName;
    }

    public AbortPolicy() {
        this(null);
    }

    /**
     * Always throws RejectedExecutionException.
     *
     * @param r the runnable task requested to be executed
     * @param e the executor attempting to execute this task
     * @throws RejectedExecutionException always
     */
    @Override
    public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
        if (!StringUtils.isEmpty(threadName)) {
            logger.error("RPC thread pool [{}] is exhausted, executor={}", threadName, executor.toString());
        }

        String msg = String.format("Rpc Server[" +
            "Thread name: %sm Pool Size: %d (active: %d, netty: %d, max: %d, largest: %d), " +
            "Task: %d (completed %d), " +
            "Executor status: (isShutdown:%s, isTerminated:%s, isTerminating:%s)]",
            threadName, executor.getPoolSize(), executor.getActiveCount(), executor.getMaximumPoolSize(), executor.getLargestPoolSize(),
            executor.isShutdown(), executor.isTerminated(), executor.isTerminating());
        logger.error(msg);
        super.rejectedExecution(r, executor);
    }
}
<file_sep>/rpc-v1/src/main/java/me/rpc/pool/RpcThreadPool.java
package me.rpc.pool;

import me.rpc.pool.policy.AbortPolicy;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.concurrent.*;

/**
 * SRpc Worker thread pool
 *
 * 独立出工作线程池，主要考虑对应对复杂的业务操作，不阻塞netty的handler IO线程
 * 当然如果业务足够简单，可以将处理逻辑写入netty的handler中也是可以的
 *
 * @author paranoidq
 * @since 1.0.0
 */
public class RpcThreadPool {
    private static Logger logger = LoggerFactory.getLogger(RpcThreadPool.class);

    public static Executor getExecutor(int threads, int queues) {
        String name = "RpcThreadPool";
        return new ThreadPoolExecutor(threads, threads, 0, TimeUnit.MILLISECONDS,
            queues == 0 ? new SynchronousQueue<>() :
                (queues < 0 ? new LinkedBlockingQueue<>() : new LinkedBlockingQueue<>(queues)),
            new NamedThreadFactory(name, true),
            new AbortPolicy(name)
            );
    }
}
<file_sep>/rpc-v3/src/main/java/me/framework/rpc/core/client/RpcServerLoader.java
package me.framework.rpc.core.client;

import com.google.common.util.concurrent.*;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import me.framework.rpc.serialize.support.RpcSerializeProtocol;
import me.framework.rpc.util.pool.RpcThreadPool;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.net.InetSocketAddress;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

/**
 * @author paranoidq
 * @since 1.0.0
 */
public class RpcServerLoader {

    private static final Logger logger = LoggerFactory.getLogger(RpcServerLoader.class);

    private static volatile RpcServerLoader instance;

    private final static String DELIMITER = ":";

    private final static int parallel = Runtime.getRuntime().availableProcessors() * 2;
    private EventLoopGroup eventLoopGroup = new NioEventLoopGroup(parallel);

    private static ListeningExecutorService threadPoolExecutor = MoreExecutors.listeningDecorator(
        (ThreadPoolExecutor) RpcThreadPool.getExecutor(16, -1));

    private volatile MessageSendHandler sendHandler;

    private Lock lock = new ReentrantLock();
    private Condition connectStatus = lock.newCondition();
    private Condition handlerStatus = lock.newCondition();


    public static RpcServerLoader getInstance() {
        if (instance == null) {
            synchronized (RpcServerLoader.class) {
                if (instance == null) {
                    instance = new RpcServerLoader();
                }
            }
        }
        return instance;
    }

    public void load(String serverAddress, RpcSerializeProtocol serializeProtocol) {
        String[] ipAddr = serverAddress.split(DELIMITER);
        if (ipAddr.length == 2) {
            String host = ipAddr[0];
            int port = Integer.parseInt(ipAddr[1]);
            final InetSocketAddress remoteAddr = new InetSocketAddress(host, port);

            ListenableFuture<Boolean> listenableFuture = threadPoolExecutor.submit(
                new MessageSendInitializeTask(eventLoopGroup, remoteAddr, serializeProtocol)
            );
            Futures.addCallback(listenableFuture, new FutureCallback<Boolean>() {
                @Override
                public void onSuccess(Boolean result) {
                    logger.info("连接服务端成功");
                    try {
                        lock.lock();
                         if (sendHandler == null) {
                             handlerStatus.await();
                         }

                         if (result == Boolean.TRUE && sendHandler != null) {
                            connectStatus.signalAll();
                         }
                    } catch (InterruptedException e) {

                    } finally {
                        lock.unlock();
                    }
                }

                @Override
                public void onFailure(Throwable t) {
                    logger.error("连接失败", t);
                }
            }, threadPoolExecutor);
        }
    }

    public void unload() {
        sendHandler.close();
        threadPoolExecutor.shutdown();
        eventLoopGroup.shutdownGracefully();
    }


    public void setMessageSendHandler(MessageSendHandler sendHandler) {
        try {
            lock.lock();
            this.sendHandler = sendHandler;
            handlerStatus.signalAll();
        } finally {
            lock.unlock();
        }
    }


    public MessageSendHandler getSendHandler() throws InterruptedException {
        try {
            lock.lock();
            if (sendHandler == null) {
                connectStatus.await();
            }
            return sendHandler;
        } finally {
            lock.unlock();
        }
    }

}
<file_sep>/rpc-v2/src/main/java/me/framework/rpc/message/MessageEncoder.java
package me.framework.rpc.message;

import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.MessageToByteEncoder;
import me.framework.rpc.serialize.support.MessageCodec;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * @author paranoidq
 * @since 1.0.0
 */
public class MessageEncoder extends MessageToByteEncoder<Object> {

    private static final Logger logger = LoggerFactory.getLogger(MessageEncoder.class);

    private MessageCodec codec;

    public MessageEncoder(MessageCodec codec) {
        this.codec = codec;
    }

    @Override
    protected void encode(ChannelHandlerContext ctx, Object msg, ByteBuf out) throws Exception {
        codec.encode(out, msg);
    }
}
<file_sep>/rpc-v3/src/main/resources/log4j.properties
log4j.rootLogger=DEBUG

log4j.additivity.me.framework.rpc = false
log4j.logger.me.framework.rpc=DEBUG,console
log4j.logger.io.netty=DEBUG,console
log4j.logger.org.springframework=DEBUG,console

log4j.appender.console=org.apache.log4j.ConsoleAppender
log4j.appender.console.layout=org.apache.log4j.PatternLayout
log4j.appender.console.layout.ConversionPattern=%d [%t] %-5p %c - %m%n
log4j.appender.console.encoding=UTF-8<file_sep>/rpc-v3/src/main/java/me/framework/rpc/core/server/MethodInvoker.java
package me.framework.rpc.core.server;

import me.framework.rpc.model.MessageRequest;
import org.apache.commons.lang3.reflect.MethodUtils;

/**
 * @author paranoidq
 * @since 1.0.0
 */
public class MethodInvoker {
    private Object serviceBean;

    public Object getServiceBean() {
        return serviceBean;
    }

    public void setServiceBean(Object serviceBean) {
        this.serviceBean = serviceBean;
    }

    public Object invoke(MessageRequest request) throws Throwable {
        String methodName = request.getMethodName();
        Object[] parameters = request.getParameterValues();
        return MethodUtils.invokeMethod(serviceBean, methodName, parameters);
    }
}
<file_sep>/rpc-v3/src/main/java/me/framework/rpc/event/ServerStartEvent.java
package me.framework.rpc.event;

import org.springframework.context.ApplicationEvent;

/**
 * @author paranoidq
 * @since 1.0.0
 */
public class ServerStartEvent extends ApplicationEvent {

    /**
     * Create a new ApplicationEvent.
     *
     * @param source the object on which the event initially occurred (never {@code null})
     */
    public ServerStartEvent(Object source) {
        super(source);
    }
}
<file_sep>/rpc-v3/src/main/java/me/framework/rpc/core/server/RpcRecvSerializeFrame.java
package me.framework.rpc.core.server;

import com.google.common.collect.ClassToInstanceMap;
import com.google.common.collect.MutableClassToInstanceMap;
import io.netty.channel.ChannelPipeline;
import me.framework.rpc.core.handler.JdkNativeRecvHandler;
import me.framework.rpc.core.handler.KryoRecvHandler;
import me.framework.rpc.core.handler.NettyRpcRecvHandler;
import me.framework.rpc.message.RpcSerializeFrame;
import me.framework.rpc.serialize.support.RpcSerializeProtocol;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Map;

/**
 * @author paranoidq
 * @since 1.0.0
 */
public class RpcRecvSerializeFrame implements RpcSerializeFrame {
    private static final Logger logger = LoggerFactory.getLogger(RpcRecvSerializeFrame.class);

    private Map<String, Object> handlerMap;

    public RpcRecvSerializeFrame(Map<String, Object> handlerMap) {
        this.handlerMap = handlerMap;
    }

    private static ClassToInstanceMap<NettyRpcRecvHandler> handler = MutableClassToInstanceMap.create();
    /**
     * 可以用注解代替
     */
    static {
        handler.putInstance(JdkNativeRecvHandler.class, new JdkNativeRecvHandler());
        handler.putInstance(KryoRecvHandler.class, new KryoRecvHandler());
        // ...
    }

    /**
     * 将组装channelHandler的部分利用策略模式独立出去
     * 不同的序列化方式组装的handler不同
     * @param protocol
     * @param pipeline
     */
    @Override
    public void select(RpcSerializeProtocol protocol, ChannelPipeline pipeline) {
        switch (protocol) {
            case JDK_SERIALIZE: {
                logger.info("Use KRYO_SERIALIZE");
                handler.getInstance(JdkNativeRecvHandler.class).handle(handlerMap, pipeline);
                break;
            }
            case KRYO_SERIALIZE: {
                logger.info("Use KRYO_SERIALIZE");
                handler.getInstance(KryoRecvHandler.class).handle(handlerMap, pipeline);
                break;
            }
            case HESSIAN_SERIALIZE: {
                logger.info("Use HESSIAN_SERIALIZE");
                break;
            }
        }
    }
}
<file_sep>/rpc-v3/src/main/java/me/framework/rpc/serialize/support/kryo/KryoSerializer.java
package me.framework.rpc.serialize.support.kryo;

import com.esotericsoftware.kryo.Kryo;
import com.esotericsoftware.kryo.io.Input;
import com.esotericsoftware.kryo.io.Output;
import com.esotericsoftware.kryo.pool.KryoFactory;
import com.esotericsoftware.kryo.pool.KryoPool;
import me.framework.rpc.serialize.support.RpcSerializer;

import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;

/**
 * 是否可以公用Serializer实例 ???
 *
 * @author paranoidq
 * @since 1.0.0
 */
public class KryoSerializer implements RpcSerializer {

    private KryoPool pool;

    public KryoSerializer() {
        pool = KryoPoolFactory.getKryoPoolInstance();
    }

    @Override
    public void serialize(OutputStream outputStream, Object object) throws IOException {
        Kryo instance = pool.borrow();
        Output output = new Output(outputStream);
        try {
            instance.writeClassAndObject(output, object);
        } finally {
            output.close();
            pool.release(instance);
        }


    }

    @Override
    public Object deserialize(InputStream inputStream) throws IOException {
        Kryo instance = pool.borrow();
        Input input = new Input(inputStream);
        try {
            return instance.readClassAndObject(input);
        } finally {
            input.close();
            pool.release(instance);
        }
    }
}
