Netty开发实现高性能的RPC服务器

Myth丶恋晨 2022-06-08 13:00 359阅读 0赞

　RPC（Remote Procedure Call Protocol）远程过程调用协议，它是一种通过网络，从远程计算机程序上请求服务，而不必了解底层网络技术的协议。说的再直白一点，就是客户端在不必知道调用细节的前提之下，调用远程计算机上运行的某个对象，使用起来就像调用本地的对象一样。目前典型的RPC实现框架有：Thrift（facebook开源）、Dubbo（alibaba开源）等等。RPC框架针对网络协议、网络I/O模型的封装是透明的，对于调用的客户端而言，它就认为自己在调用本地的一个对象。至于传输层上，运用的是TCP协议、UDP协议、亦或是HTTP协议，一概不关心。从网络I/O模型上来看，是基于select、poll、epoll方式、还是IOCP（I/O Completion Port）方式承载实现的，对于调用者而言也不用关心。

　　目前，主流的RPC框架都支持跨语言调用，即有所谓的IDL（接口定义语言），其实，这个并不是RPC所必须要求的。如果你的RPC框架没有跨语言的要求，IDL就可以不用包括了。

　　最后，值得一提的是，衡量一个RPC框架性能的好坏与否，RPC的网络I/O模型的选择，至关重要。在此基础上，设计出来的RPC服务器，可以考虑支持阻塞式同步IO、非阻塞式同步IO、当然还有所谓的多路复用IO模型、异步IO模型。支持不同的网络IO模型，在高并发的状态下，处理性能上会有很大的差别。还有一个衡量的标准，就是选择的传输协议。是基于TCP协议、还是HTTP协议、还是UDP协议？对性能也有一定的影响。但是从我目前了解的情况来看，大多数RPC开源实现框架都是基于TCP、或者HTTP的，目测没有采用UDP协议做为主要的传输协议的。

　　明白了RPC的使用原理和性能要求。现在，我们能不能撇开那些RPC开源框架，自己动手开发一个高性能的RPC服务器呢？我想，还是可以的。现在本人就使用Java，基于Netty，开发实现一个高性能的RPC服务器。

　　如何实现、基于什么原理？并发处理性能如何？请继续接着看下文。

　　我们有的时候，为了提高单个节点的通信吞吐量，提高通信性能。如果是基于Java后端的，一般首选的是NIO框架（No-block IO）。但是问题也来了，Java的NIO掌握起来要相当的技术功底，和足够的技术积累，使用起来才能得心应手。一般的开发人员，如果要使用NIO开发一个后端的TCP/HTTP服务器，附带考虑TCP粘包、网络通信异常、消息链接处理等等网络通信细节，开发门槛太高，所以比较明智的选择是，采用业界主流的NIO框架进行服务器后端开发。主流的NIO框架主要有Netty、Mina。它们主要都是基于TCP通信，非阻塞的IO、灵活的IO线程池而设计的，应对高并发请求也是绰绰有余。随着Netty、Mina这样优秀的NIO框架，设计上日趋完善，Java后端高性能服务器开发，在技术上提供了有力的支持保障，从而打破了C++在服务器后端，一统天下的局面。因为在此之前，Java的NIO一直受人诟病，让人敬而远之！

　　既然，这个RPC服务器是基于Netty的，那就在说说Netty吧。实际上Netty是对JAVA NIO框架的再次封装，它的开源网址是http://netty.io/，本文中使用的Netty版本是：4.0版本，可以通过http://dl.bintray.com/netty/downloads/netty-4.0.37.Final.tar.bz2，进行下载使用。那也许你会问，如何使用Netty进行RPC服务器的开发呢？实际不难，下面我就简单的说明一下技术原理：

　　1、定义RPC请求消息、应答消息结构，里面要包括RPC的接口定义模块、包括远程调用的类名、方法名称、参数结构、参数值等信息。

　　2、服务端初始化的时候通过容器加载RPC接口定义和RPC接口实现类对象的映射关系，然后等待客户端发起调用请求。

　　3、客户端发起的RPC消息里面包含，远程调用的类名、方法名称、参数结构、参数值等信息，通过网络，以字节流的方式送给RPC服务端，RPC服务端接收到字节流的请求之后，去对应的容器里面，查找客户端接口映射的具体实现对象。

　　4、RPC服务端找到实现对象的参数信息，通过反射机制创建该对象的实例，并返回调用处理结果，最后封装成RPC应答消息通知到客户端。

　　5、客户端通过网络，收到字节流形式的RPC应答消息，进行拆包、解析之后，显示远程调用结果。

　　上面说的是很简单，但是实现的时候，我们还要考虑如下的问题：

　　1、RPC服务器的传输层是基于TCP协议的，出现粘包咋办？这样客户端的请求，服务端不是会解析失败？好在Netty里面已经提供了解决TCP粘包问题的解码器：LengthFieldBasedFrameDecoder，可以靠它轻松搞定TCP粘包问题。

　　2、Netty服务端的线程模型是单线程、多线程（一个线程负责客户端连接，连接成功之后，丢给后端IO的线程池处理）、还是主从模式（客户端连接、后端IO处理都是基于线程池的实现）。当然在这里，我出于性能考虑，使用了Netty主从线程池模型。

　　3、Netty的IO处理线程池，如果遇到非常耗时的业务，出现阻塞了咋办？这样不是很容易把后端的NIO线程给挂死、阻塞？本文的处理方式是，对于复杂的后端业务，分派到专门的业务线程池里面，进行异步回调处理。

　　4、RPC消息的传输是通过字节流在NIO的通道（Channel）之间传输，那具体如何实现呢？本文，是通过基于Java原生对象序列化机制的编码、解码器（ObjectEncoder、ObjectDecoder）进行实现的。当然出于性能考虑，这个可能不是最优的方案。更优的方案是把消息的编码、解码器，搞成可以配置实现的。具体比如可以通过：protobuf、JBoss Marshalling方式进行解码和编码，以提高网络消息的传输效率。

　　5、RPC服务器要考虑多线程、高并发的使用场景，所以线程安全是必须的。此外尽量不要使用synchronized进行加锁，改用轻量级的ReentrantLock方式进行代码块的条件加锁。比如本文中的RPC消息处理回调，就有这方面的使用。

　　6、RPC服务端的服务接口对象和服务接口实现对象要能轻易的配置，轻松进行加载、卸载。在这里，本文是通过Spring容器进行统一的对象管理。

　　综上所述，本文设计的RPC服务器调用的流程图如下所示：

  1.  ![913759-20160625010800906-382334403.jpg][]

　　客户端并发发起RPC调用请求，然后RPC服务端使用Netty连接器，分派出N个NIO连接线程，这个时候Netty连接器的任务结束。然后NIO连接线程是统一放到Netty NIO处理线程池进行管理，这个线程池里面会对具体的RPC请求连接进行消息编码、消息解码、消息处理等等一系列操作。最后进行消息处理（Handler）的时候，处于性能考虑，这里的设计是，直接把复杂的消息处理过程，丢给专门的RPC业务处理线程池集中处理，然后Handler对应的NIO线程就立即返回、不会阻塞。这个时候RPC调用结束，客户端会异步等待服务端消息的处理结果，本文是通过消息回调机制实现（MessageCallBack）。

　　再来说一说Netty对于RPC消息的解码、编码、处理对应的模块和流程，具体如下图所示：

913759-20160625011424125-1152475230.jpg

　　从上图可以看出客户端、服务端对RPC消息编码、解码、处理调用的模块以及调用顺序了。Netty就是把这样一个一个的处理器串在一起，形成一个责任链，统一进行调用。

　　说了这么多，现在先简单看下，我设计实现的NettyRPC的代码目录层级结构：

  1.  ![913759-20160625010518766-1194749946.jpg][]

　　其中newlandframework.netty.rpc.core包是NettyRPC的核心实现。newlandframework.netty.rpc.model包里面，则封装了RPC消息请求、应答报文结构，以及RPC服务接口与实现绑定关系的容器定义。newlandframework.netty.rpc.config里面定义了NettyRPC的服务端文件配置属性。

　　下面先来看下newlandframework.netty.rpc.model包中定义的内容。具体是RPC消息请求、应答消息的结构定义：

　　RPC请求消息结构

复制代码

复制代码

  1. /**
  2.  * @filename:MessageRequest.java
  3.  *
  4.  * Newland Co. Ltd. All rights reserved.
  5.  *
  6.  * @Description:rpc服务请求结构
  7.  * @author tangjie
  8.  * @version 1.0
  9.  *
  10.  */
  11. package newlandframework.netty.rpc.model;
  13. import java.io.Serializable;
  14. import org.apache.commons.lang.builder.ToStringBuilder;
  15. import org.apache.commons.lang.builder.ToStringStyle;
  17. public class MessageRequest implements Serializable {
  19.     private String messageId;
  20.     private String className;
  21.     private String methodName;
  22.     private Class<?>[] typeParameters;
  23.     private Object[] parametersVal;
  25.     public String getMessageId() {
  26.         return messageId;
  27.     }
  29.     public void setMessageId(String messageId) {
  30.         this.messageId = messageId;
  31.     }
  33.     public String getClassName() {
  34.         return className;
  35.     }
  37.     public void setClassName(String className) {
  38.         this.className = className;
  39.     }
  41.     public String getMethodName() {
  42.         return methodName;
  43.     }
  45.     public void setMethodName(String methodName) {
  46.         this.methodName = methodName;
  47.     }
  49.     public Class<?>[] gettypeParameters() {
  50.         return typeParameters;
  51.     }
  53.     public void settypeParameters(Class<?>[] typeParameters) {
  54.         this.typeParameters = typeParameters;
  55.     }
  57.     public Object[] getParameters() {
  58.         return parametersVal;
  59.     }
  61.     public void setParameters(Object[] parametersVal) {
  62.         this.parametersVal = parametersVal;
  63.     }
  65.     public String toString() {
  66.         return new ToStringBuilder(this, ToStringStyle.SHORT_PREFIX_STYLE)
  67.                 .append("messageId", messageId).append("className", className)
  68.                 .append("methodName", methodName).toString();
  69.     }
  70. }

复制代码

复制代码

　　RPC应答消息结构

复制代码

复制代码

  1. /**
  2.  * @filename:MessageResponse.java
  3.  *
  4.  * Newland Co. Ltd. All rights reserved.
  5.  *
  6.  * @Description:rpc服务应答结构
  7.  * @author tangjie
  8.  * @version 1.0
  9.  *
  10.  */
  11. package newlandframework.netty.rpc.model;
  13. import java.io.Serializable;
  14. import org.apache.commons.lang.builder.ToStringBuilder;
  15. import org.apache.commons.lang.builder.ToStringStyle;
  17. public class MessageResponse implements Serializable {
  19.     private String messageId;
  20.     private String error;
  21.     private Object resultDesc;
  23.     public String getMessageId() {
  24.         return messageId;
  25.     }
  27.     public void setMessageId(String messageId) {
  28.         this.messageId = messageId;
  29.     }
  31.     public String getError() {
  32.         return error;
  33.     }
  35.     public void setError(String error) {
  36.         this.error = error;
  37.     }
  39.     public Object getResult() {
  40.         return resultDesc;
  41.     }
  43.     public void setResult(Object resultDesc) {
  44.         this.resultDesc = resultDesc;
  45.     }
  47.     public String toString() {
  48.         return new ToStringBuilder(this, ToStringStyle.SHORT_PREFIX_STYLE)
  49.                 .append("messageId", messageId).append("error", error).toString();
  50.     }
  51. }

复制代码

复制代码

　　RPC服务接口定义、服务接口实现绑定关系容器定义，提供给spring作为容器使用。

复制代码

复制代码

  1. /**
  2.  * @filename:MessageKeyVal.java
  3.  *
  4.  * Newland Co. Ltd. All rights reserved.
  5.  *
  6.  * @Description:rpc服务映射容器
  7.  * @author tangjie
  8.  * @version 1.0
  9.  *
  10.  */
  11. package newlandframework.netty.rpc.model;
  13. import java.util.Map;
  15. public class MessageKeyVal {
  17.     private Map<String, Object> messageKeyVal;
  19.     public void setMessageKeyVal(Map<String, Object> messageKeyVal) {
  20.         this.messageKeyVal = messageKeyVal;
  21.     }
  23.     public Map<String, Object> getMessageKeyVal() {
  24.         return messageKeyVal;
  25.     }
  26. }

复制代码

复制代码

　　好了，定义好核心模型结构之后，现在再向大家展示一下NettyRPC核心包：newlandframework.netty.rpc.core的关键部分实现代码，首先是业务线程池相关类的实现代码，具体如下：

　　线程工厂定义实现

复制代码

复制代码

  1. /**
  2.  * @filename:NamedThreadFactory.java
  3.  *
  4.  * Newland Co. Ltd. All rights reserved.
  5.  *
  6.  * @Description:线程工厂
  7.  * @author tangjie
  8.  * @version 1.0
  9.  *
  10.  */
  11. package newlandframework.netty.rpc.core;
  13. import java.util.concurrent.ThreadFactory;
  14. import java.util.concurrent.atomic.AtomicInteger;
  16. public class NamedThreadFactory implements ThreadFactory {
  18.     private static final AtomicInteger threadNumber = new AtomicInteger(1);
  20.     private final AtomicInteger mThreadNum = new AtomicInteger(1);
  22.     private final String prefix;
  24.     private final boolean daemoThread;
  26.     private final ThreadGroup threadGroup;
  28.     public NamedThreadFactory() {
  29.         this("rpcserver-threadpool-" + threadNumber.getAndIncrement(), false);
  30.     }
  32.     public NamedThreadFactory(String prefix) {
  33.         this(prefix, false);
  34.     }
  36.     public NamedThreadFactory(String prefix, boolean daemo) {
  37.         this.prefix = prefix + "-thread-";
  38.         daemoThread = daemo;
  39.         SecurityManager s = System.getSecurityManager();
  40.         threadGroup = (s == null) ? Thread.currentThread().getThreadGroup() : s.getThreadGroup();
  41.     }
  43.     public Thread newThread(Runnable runnable) {
  44.         String name = prefix + mThreadNum.getAndIncrement();
  45.         Thread ret = new Thread(threadGroup, runnable, name, 0);
  46.         ret.setDaemon(daemoThread);
  47.         return ret;
  48.     }
  50.     public ThreadGroup getThreadGroup() {
  51.         return threadGroup;
  52.     }
  53. }

复制代码

复制代码

　　业务线程池定义实现

复制代码

复制代码

  1. /**
  2.  * @filename:RpcThreadPool.java
  3.  *
  4.  * Newland Co. Ltd. All rights reserved.
  5.  *
  6.  * @Description:rpc线程池封装
  7.  * @author tangjie
  8.  * @version 1.0
  9.  *
  10.  */
  11. package newlandframework.netty.rpc.core;
  13. import java.util.concurrent.Executor;
  14. import java.util.concurrent.LinkedBlockingQueue;
  15. import java.util.concurrent.SynchronousQueue;
  16. import java.util.concurrent.ThreadPoolExecutor;
  17. import java.util.concurrent.TimeUnit;
  19. public class RpcThreadPool {
  21.     //独立出线程池主要是为了应对复杂耗I/O操作的业务，不阻塞netty的handler线程而引入
  22.     //当然如果业务足够简单，把处理逻辑写入netty的handler（ChannelInboundHandlerAdapter）也未尝不可
  23.     public static Executor getExecutor(int threads, int queues) {
  24.         String name = "RpcThreadPool";
  25.         return new ThreadPoolExecutor(threads, threads, 0, TimeUnit.MILLISECONDS,
  26.                 queues == 0 ? new SynchronousQueue<Runnable>()
  27.                         : (queues < 0 ? new LinkedBlockingQueue<Runnable>()
  28.                                 : new LinkedBlockingQueue<Runnable>(queues)),
  29.                 new NamedThreadFactory(name, true), new AbortPolicyWithReport(name));
  30.     }
  31. }

复制代码

复制代码

复制代码

复制代码

  1. /**
  2.  * @filename:AbortPolicyWithReport.java
  3.  *
  4.  * Newland Co. Ltd. All rights reserved.
  5.  *
  6.  * @Description:线程池异常策略
  7.  * @author tangjie
  8.  * @version 1.0
  9.  *
  10.  */
  11. package newlandframework.netty.rpc.core;
  13. import java.util.concurrent.RejectedExecutionException;
  14. import java.util.concurrent.ThreadPoolExecutor;
  16. public class AbortPolicyWithReport extends ThreadPoolExecutor.AbortPolicy {
  18.     private final String threadName;
  20.     public AbortPolicyWithReport(String threadName) {
  21.         this.threadName = threadName;
  22.     }
  24.     public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
  25.         String msg = String.format("RpcServer["
  26.                 + " Thread Name: %s, Pool Size: %d (active: %d, core: %d, max: %d, largest: %d), Task: %d (completed: %d),"
  27.                 + " Executor status:(isShutdown:%s, isTerminated:%s, isTerminating:%s)]",
  28.                 threadName, e.getPoolSize(), e.getActiveCount(), e.getCorePoolSize(), e.getMaximumPoolSize(), e.getLargestPoolSize(),
  29.                 e.getTaskCount(), e.getCompletedTaskCount(), e.isShutdown(), e.isTerminated(), e.isTerminating());
  30.         System.out.println(msg);
  31.         throw new RejectedExecutionException(msg);
  32.     }
  33. }

复制代码

复制代码

　　RPC调用客户端定义实现

复制代码

复制代码

  1. /**
  2.  * @filename:MessageSendExecutor.java
  3.  *
  4.  * Newland Co. Ltd. All rights reserved.
  5.  *
  6.  * @Description:Rpc客户端执行模块
  7.  * @author tangjie
  8.  * @version 1.0
  9.  *
  10.  */
  11. package newlandframework.netty.rpc.core;
  13. import java.lang.reflect.Proxy;
  15. public class MessageSendExecutor {
  17.     private RpcServerLoader loader = RpcServerLoader.getInstance();
  19.     public MessageSendExecutor(String serverAddress) {
  20.         loader.load(serverAddress);
  21.     }
  23.     public void stop() {
  24.         loader.unLoad();
  25.     }
  27.     public static <T> T execute(Class<T> rpcInterface) {
  28.         return (T) Proxy.newProxyInstance(
  29.                 rpcInterface.getClassLoader(),
  30.                 new Class<?>[]{rpcInterface},
  31.                 new MessageSendProxy<T>(rpcInterface)
  32.         );
  33.     }
  34. }

复制代码

复制代码

　　这里的RPC客户端实际上，是动态代理了MessageSendProxy，当然这里是应用了，JDK原生的动态代理实现，你还可以改成CGLIB（Code Generation Library）方式。不过本人测试了一下CGLIB方式，在高并发的情况下面会出现空指针异常，但是同样的情况，JDK原生的动态代理却没有问题。并发程度不高的情况下面，两种代理方式都运行正常。后续再深入研究看看吧！废话不说了，现在给出MessageSendProxy的实现方式

复制代码

复制代码

  1. /**
  2.  * @filename:MessageSendProxy.java
  3.  *
  4.  * Newland Co. Ltd. All rights reserved.
  5.  *
  6.  * @Description:Rpc客户端消息处理
  7.  * @author tangjie
  8.  * @version 1.0
  9.  *
  10.  */
  11. package newlandframework.netty.rpc.core;
  13. import java.lang.reflect.InvocationHandler;
  14. import java.lang.reflect.Method;
  15. import java.util.UUID;
  16. import newlandframework.netty.rpc.model.MessageRequest;
  18. public class MessageSendProxy<T> implements InvocationHandler {
  20.     private Class<T> cls;
  22.     public MessageSendProxy(Class<T> cls) {
  23.         this.cls = cls;
  24.     }
  26.     public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
  27.         MessageRequest request = new MessageRequest();
  28.         request.setMessageId(UUID.randomUUID().toString());
  29.         request.setClassName(method.getDeclaringClass().getName());
  30.         request.setMethodName(method.getName());
  31.         request.settypeParameters(method.getParameterTypes());
  32.         request.setParameters(args);
  34.         MessageSendHandler handler = RpcServerLoader.getInstance().getMessageSendHandler();
  35.         MessageCallBack callBack = handler.sendRequest(request);
  36.         return callBack.start();
  37.     }
  38. }

复制代码

复制代码

　　进一步发现MessageSendProxy其实是把消息发送给RpcServerLoader模块，它的代码如下：

复制代码

复制代码

  1. /**
  2.  * @filename:RpcServerLoader.java
  3.  *
  4.  * Newland Co. Ltd. All rights reserved.
  5.  *
  6.  * @Description:rpc服务器配置加载
  7.  * @author tangjie
  8.  * @version 1.0
  9.  *
  10.  */
  11. package newlandframework.netty.rpc.core;
  13. import io.netty.channel.EventLoopGroup;
  14. import io.netty.channel.nio.NioEventLoopGroup;
  15. import java.net.InetSocketAddress;
  16. import java.util.concurrent.ThreadPoolExecutor;
  18. public class RpcServerLoader {
  20.     private volatile static RpcServerLoader rpcServerLoader;
  21.     private final static String DELIMITER = ":";
  23.     //方法返回到Java虚拟机的可用的处理器数量
  24.     private final static int parallel = Runtime.getRuntime().availableProcessors() * 2;
  25.     //netty nio线程池
  26.     private EventLoopGroup eventLoopGroup = new NioEventLoopGroup(parallel);
  27.     private static ThreadPoolExecutor threadPoolExecutor = (ThreadPoolExecutor) RpcThreadPool.getExecutor(16, -1);
  28.     private MessageSendHandler messageSendHandler = null;
  30.     private RpcServerLoader() {
  31.     }
  33.     //并发双重锁定
  34.     public static RpcServerLoader getInstance() {
  35.         if (rpcServerLoader == null) {
  36.             synchronized (RpcServerLoader.class) {
  37.                 if (rpcServerLoader == null) {
  38.                     rpcServerLoader = new RpcServerLoader();
  39.                 }
  40.             }
  41.         }
  42.         return rpcServerLoader;
  43.     }
  45.     public void load(String serverAddress) {
  46.         String[] ipAddr = serverAddress.split(RpcServerLoader.DELIMITER);
  47.         if (ipAddr.length == 2) {
  48.             String host = ipAddr[0];
  49.             int port = Integer.parseInt(ipAddr[1]);
  50.             final InetSocketAddress remoteAddr = new InetSocketAddress(host, port);
  52.             threadPoolExecutor.submit(new MessageSendInitializeTask(eventLoopGroup, remoteAddr, this));
  53.         }
  54.     }
  56.     public void setMessageSendHandler(MessageSendHandler messageInHandler) {
  57.         this.messageSendHandler = messageInHandler;
  58.     }
  60.     public MessageSendHandler getMessageSendHandler() {
  61.         return messageSendHandler;
  62.     }
  64.     public void unLoad() {
  65.         messageSendHandler.close();
  66.         threadPoolExecutor.shutdown();
  67.         eventLoopGroup.shutdownGracefully();
  68.     }
  69. }

复制代码

复制代码

　　好了，现在一次性给出RPC客户端消息编码、解码、处理的模块实现代码。

复制代码

复制代码

  1. /**
  2.  * @filename:MessageSendInitializeTask.java
  3.  *
  4.  * Newland Co. Ltd. All rights reserved.
  5.  *
  6.  * @Description:Rpc客户端线程任务处理
  7.  * @author tangjie
  8.  * @version 1.0
  9.  *
  10.  */
  11. package newlandframework.netty.rpc.core;
  13. import io.netty.bootstrap.Bootstrap;
  14. import io.netty.channel.ChannelFuture;
  15. import io.netty.channel.ChannelFutureListener;
  16. import io.netty.channel.ChannelOption;
  17. import io.netty.channel.EventLoopGroup;
  18. import io.netty.channel.socket.nio.NioSocketChannel;
  19. import java.net.InetSocketAddress;
  21. public class MessageSendInitializeTask implements Runnable {
  23.     private EventLoopGroup eventLoopGroup = null;
  24.     private InetSocketAddress serverAddress = null;
  25.     private RpcServerLoader loader = null;
  27.     MessageSendInitializeTask(EventLoopGroup eventLoopGroup, InetSocketAddress serverAddress, RpcServerLoader loader) {
  28.         this.eventLoopGroup = eventLoopGroup;
  29.         this.serverAddress = serverAddress;
  30.         this.loader = loader;
  31.     }
  33.     public void run() {
  34.         Bootstrap b = new Bootstrap();
  35.         b.group(eventLoopGroup)
  36.                 .channel(NioSocketChannel.class).option(ChannelOption.SO_KEEPALIVE, true);
  37.         b.handler(new MessageSendChannelInitializer());
  39.         ChannelFuture channelFuture = b.connect(serverAddress);
  40.         channelFuture.addListener(new ChannelFutureListener() {
  41.             public void operationComplete(final ChannelFuture channelFuture) throws Exception {
  42.                 if (channelFuture.isSuccess()) {
  43.                     MessageSendHandler handler = channelFuture.channel().pipeline().get(MessageSendHandler.class);
  44.                     MessageSendInitializeTask.this.loader.setMessageSendHandler(handler);
  45.                 }
  46.             }
  47.         });
  48.     }
  49. }

复制代码

复制代码

复制代码

复制代码

  1. /**
  2.  * @filename:MessageSendChannelInitializer.java
  3.  *
  4.  * Newland Co. Ltd. All rights reserved.
  5.  *
  6.  * @Description:Rpc客户端管道初始化
  7.  * @author tangjie
  8.  * @version 1.0
  9.  *
  10.  */
  11. package newlandframework.netty.rpc.core;
  13. import io.netty.channel.ChannelInitializer;
  14. import io.netty.channel.ChannelPipeline;
  15. import io.netty.channel.socket.SocketChannel;
  16. import io.netty.handler.codec.LengthFieldBasedFrameDecoder;
  17. import io.netty.handler.codec.LengthFieldPrepender;
  18. import io.netty.handler.codec.serialization.ClassResolvers;
  19. import io.netty.handler.codec.serialization.ObjectDecoder;
  20. import io.netty.handler.codec.serialization.ObjectEncoder;
  22. public class MessageSendChannelInitializer extends ChannelInitializer<SocketChannel> {
  24.     //ObjectDecoder 底层默认继承半包解码器LengthFieldBasedFrameDecoder处理粘包问题的时候，
  25.     //消息头开始即为长度字段，占据4个字节。这里出于保持兼容的考虑
  26.     final public static int MESSAGE_LENGTH = 4;
  28.     protected void initChannel(SocketChannel socketChannel) throws Exception {
  29.         ChannelPipeline pipeline = socketChannel.pipeline();
  30.         //ObjectDecoder的基类半包解码器LengthFieldBasedFrameDecoder的报文格式保持兼容。因为底层的父类LengthFieldBasedFrameDecoder
  31.         //的初始化参数即为super(maxObjectSize, 0, 4, 0, 4);
  32.         pipeline.addLast(new LengthFieldBasedFrameDecoder(Integer.MAX_VALUE, 0, MessageSendChannelInitializer.MESSAGE_LENGTH, 0, MessageSendChannelInitializer.MESSAGE_LENGTH));
  33.         //利用LengthFieldPrepender回填补充ObjectDecoder消息报文头
  34.         pipeline.addLast(new LengthFieldPrepender(MessageSendChannelInitializer.MESSAGE_LENGTH));
  35.         pipeline.addLast(new ObjectEncoder());
  36.         //考虑到并发性能，采用weakCachingConcurrentResolver缓存策略。一般情况使用:cacheDisabled即可
  37.         pipeline.addLast(new ObjectDecoder(Integer.MAX_VALUE, ClassResolvers.weakCachingConcurrentResolver(this.getClass().getClassLoader())));
  38.         pipeline.addLast(new MessageSendHandler());
  39.     }
  40. }

复制代码

复制代码

复制代码

复制代码

  1. /**
  2.  * @filename:MessageSendHandler.java
  3.  *
  4.  * Newland Co. Ltd. All rights reserved.
  5.  *
  6.  * @Description:Rpc客户端处理模块
  7.  * @author tangjie
  8.  * @version 1.0
  9.  *
  10.  */
  11. package newlandframework.netty.rpc.core;
  13. import io.netty.buffer.Unpooled;
  14. import io.netty.channel.Channel;
  15. import io.netty.channel.ChannelFutureListener;
  16. import io.netty.channel.ChannelHandlerContext;
  17. import io.netty.channel.ChannelInboundHandlerAdapter;
  18. import java.net.SocketAddress;
  19. import java.util.concurrent.ConcurrentHashMap;
  20. import newlandframework.netty.rpc.model.MessageRequest;
  21. import newlandframework.netty.rpc.model.MessageResponse;
  23. public class MessageSendHandler extends ChannelInboundHandlerAdapter {
  25.     private ConcurrentHashMap<String, MessageCallBack> mapCallBack = new ConcurrentHashMap<String, MessageCallBack>();
  27.     private volatile Channel channel;
  28.     private SocketAddress remoteAddr;
  30.     public Channel getChannel() {
  31.         return channel;
  32.     }
  34.     public SocketAddress getRemoteAddr() {
  35.         return remoteAddr;
  36.     }
  38.     public void channelActive(ChannelHandlerContext ctx) throws Exception {
  39.         super.channelActive(ctx);
  40.         this.remoteAddr = this.channel.remoteAddress();
  41.     }
  43.     public void channelRegistered(ChannelHandlerContext ctx) throws Exception {
  44.         super.channelRegistered(ctx);
  45.         this.channel = ctx.channel();
  46.     }
  48.     public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
  49.         MessageResponse response = (MessageResponse) msg;
  50.         String messageId = response.getMessageId();
  51.         MessageCallBack callBack = mapCallBack.get(messageId);
  52.         if (callBack != null) {
  53.             mapCallBack.remove(messageId);
  54.             callBack.over(response);
  55.         }
  56.     }
  58.     public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
  59.         ctx.close();
  60.     }
  62.     public void close() {
  63.         channel.writeAndFlush(Unpooled.EMPTY_BUFFER).addListener(ChannelFutureListener.CLOSE);
  64.     }
  66.     public MessageCallBack sendRequest(MessageRequest request) {
  67.         MessageCallBack callBack = new MessageCallBack(request);
  68.         mapCallBack.put(request.getMessageId(), callBack);
  69.         channel.writeAndFlush(request);
  70.         return callBack;
  71.     }
  72. }

复制代码

复制代码

　　最后给出RPC服务端的实现。首先是通过spring自动加载RPC服务接口、接口实现容器绑定加载，初始化Netty主/从线程池等操作，具体是通过MessageRecvExecutor模块实现的，现在给出实现代码：

复制代码

复制代码

  1. /**
  2.  * @filename:MessageRecvExecutor.java
  3.  *
  4.  * Newland Co. Ltd. All rights reserved.
  5.  *
  6.  * @Description:Rpc服务器执行模块
  7.  * @author tangjie
  8.  * @version 1.0
  9.  *
  10.  */
  11. package newlandframework.netty.rpc.core;
  13. import io.netty.bootstrap.ServerBootstrap;
  14. import io.netty.channel.ChannelFuture;
  15. import io.netty.channel.ChannelOption;
  16. import io.netty.channel.EventLoopGroup;
  17. import io.netty.channel.nio.NioEventLoopGroup;
  18. import io.netty.channel.socket.nio.NioServerSocketChannel;
  19. import java.nio.channels.spi.SelectorProvider;
  20. import java.util.Iterator;
  21. import java.util.Map;
  22. import java.util.Set;
  23. import java.util.concurrent.ConcurrentHashMap;
  24. import java.util.concurrent.ThreadFactory;
  25. import java.util.concurrent.ThreadPoolExecutor;
  26. import java.util.logging.Level;
  27. import newlandframework.netty.rpc.model.MessageKeyVal;
  28. import org.springframework.beans.BeansException;
  29. import org.springframework.beans.factory.InitializingBean;
  30. import org.springframework.context.ApplicationContext;
  31. import org.springframework.context.ApplicationContextAware;
  33. public class MessageRecvExecutor implements ApplicationContextAware, InitializingBean {
  35.     private String serverAddress;
  36.     private final static String DELIMITER = ":";
  38.     private Map<String, Object> handlerMap = new ConcurrentHashMap<String, Object>();
  40.     private static ThreadPoolExecutor threadPoolExecutor;
  42.     public MessageRecvExecutor(String serverAddress) {
  43.         this.serverAddress = serverAddress;
  44.     }
  46.     public static void submit(Runnable task) {
  47.         if (threadPoolExecutor == null) {
  48.             synchronized (MessageRecvExecutor.class) {
  49.                 if (threadPoolExecutor == null) {
  50.                     threadPoolExecutor = (ThreadPoolExecutor) RpcThreadPool.getExecutor(16, -1);
  51.                 }
  52.             }
  53.         }
  54.         threadPoolExecutor.submit(task);
  55.     }
  57.     public void setApplicationContext(ApplicationContext ctx) throws BeansException {
  58.         try {
  59.             MessageKeyVal keyVal = (MessageKeyVal) ctx.getBean(Class.forName("newlandframework.netty.rpc.model.MessageKeyVal"));
  60.             Map<String, Object> rpcServiceObject = keyVal.getMessageKeyVal();
  62.             Set s = rpcServiceObject.entrySet();
  63.             Iterator<Map.Entry<String, Object>> it = s.iterator();
  64.             Map.Entry<String, Object> entry;
  66.             while (it.hasNext()) {
  67.                 entry = it.next();
  68.                 handlerMap.put(entry.getKey(), entry.getValue());
  69.             }
  70.         } catch (ClassNotFoundException ex) {
  71.             java.util.logging.Logger.getLogger(MessageRecvExecutor.class.getName()).log(Level.SEVERE, null, ex);
  72.         }
  73.     }
  75.     public void afterPropertiesSet() throws Exception {
  76.         //netty的线程池模型设置成主从线程池模式，这样可以应对高并发请求
  77.         //当然netty还支持单线程、多线程网络IO模型，可以根据业务需求灵活配置
  78.         ThreadFactory threadRpcFactory = new NamedThreadFactory("NettyRPC ThreadFactory");
  80.         //方法返回到Java虚拟机的可用的处理器数量
  81.         int parallel = Runtime.getRuntime().availableProcessors() * 2;
  83.         EventLoopGroup boss = new NioEventLoopGroup();
  84.         EventLoopGroup worker = new NioEventLoopGroup(parallel,threadRpcFactory,SelectorProvider.provider());
  86.         try {
  87.             ServerBootstrap bootstrap = new ServerBootstrap();
  88.             bootstrap.group(boss, worker).channel(NioServerSocketChannel.class)
  89.                     .childHandler(new MessageRecvChannelInitializer(handlerMap))
  90.                     .option(ChannelOption.SO_BACKLOG, 128)
  91.                     .childOption(ChannelOption.SO_KEEPALIVE, true);
  93.             String[] ipAddr = serverAddress.split(MessageRecvExecutor.DELIMITER);
  95.             if (ipAddr.length == 2) {
  96.                 String host = ipAddr[0];
  97.                 int port = Integer.parseInt(ipAddr[1]);
  98.                 ChannelFuture future = bootstrap.bind(host, port).sync();
  99.                 System.out.printf("[author tangjie] Netty RPC Server start success ip:%s port:%d\n", host, port);
  100.                 future.channel().closeFuture().sync();
  101.             } else {
  102.                 System.out.printf("[author tangjie] Netty RPC Server start fail!\n");
  103.             }
  104.         } finally {
  105.             worker.shutdownGracefully();
  106.             boss.shutdownGracefully();
  107.         }
  108.     }
  109. }

复制代码

复制代码

　　最后还是老规矩，给出RPC服务端消息编码、解码、处理的核心模块代码实现，具体如下：

复制代码

复制代码

  1. /**
  2.  * @filename:MessageRecvChannelInitializer.java
  3.  *
  4.  * Newland Co. Ltd. All rights reserved.
  5.  *
  6.  * @Description:Rpc服务端管道初始化
  7.  * @author tangjie
  8.  * @version 1.0
  9.  *
  10.  */
  11. package newlandframework.netty.rpc.core;
  13. import io.netty.channel.ChannelInitializer;
  14. import io.netty.channel.ChannelPipeline;
  15. import io.netty.channel.socket.SocketChannel;
  16. import io.netty.handler.codec.LengthFieldBasedFrameDecoder;
  17. import io.netty.handler.codec.LengthFieldPrepender;
  18. import io.netty.handler.codec.serialization.ClassResolvers;
  19. import io.netty.handler.codec.serialization.ObjectDecoder;
  20. import io.netty.handler.codec.serialization.ObjectEncoder;
  21. import java.util.Map;
  23. public class MessageRecvChannelInitializer extends ChannelInitializer<SocketChannel> {
  25.     //ObjectDecoder 底层默认继承半包解码器LengthFieldBasedFrameDecoder处理粘包问题的时候，
  26.     //消息头开始即为长度字段，占据4个字节。这里出于保持兼容的考虑
  27.     final public static int MESSAGE_LENGTH = 4;
  28.     private Map<String, Object> handlerMap = null;
  30.     MessageRecvChannelInitializer(Map<String, Object> handlerMap) {
  31.         this.handlerMap = handlerMap;
  32.     }
  34.     protected void initChannel(SocketChannel socketChannel) throws Exception {
  35.         ChannelPipeline pipeline = socketChannel.pipeline();
  36.         //ObjectDecoder的基类半包解码器LengthFieldBasedFrameDecoder的报文格式保持兼容。因为底层的父类LengthFieldBasedFrameDecoder
  37.         //的初始化参数即为super(maxObjectSize, 0, 4, 0, 4); 
  38.         pipeline.addLast(new LengthFieldBasedFrameDecoder(Integer.MAX_VALUE, 0, MessageRecvChannelInitializer.MESSAGE_LENGTH, 0, MessageRecvChannelInitializer.MESSAGE_LENGTH));
  39.         //利用LengthFieldPrepender回填补充ObjectDecoder消息报文头
  40.         pipeline.addLast(new LengthFieldPrepender(MessageRecvChannelInitializer.MESSAGE_LENGTH));
  41.         pipeline.addLast(new ObjectEncoder());
  42.         //考虑到并发性能，采用weakCachingConcurrentResolver缓存策略。一般情况使用:cacheDisabled即可
  43.         pipeline.addLast(new ObjectDecoder(Integer.MAX_VALUE, ClassResolvers.weakCachingConcurrentResolver(this.getClass().getClassLoader())));
  44.         pipeline.addLast(new MessageRecvHandler(handlerMap));
  45.     }
  46. }

复制代码

复制代码

复制代码

复制代码

  1. /**
  2.  * @filename:MessageRecvHandler.java
  3.  *
  4.  * Newland Co. Ltd. All rights reserved.
  5.  *
  6.  * @Description:Rpc服务器消息处理
  7.  * @author tangjie
  8.  * @version 1.0
  9.  *
  10.  */
  11. package newlandframework.netty.rpc.core;
  13. import io.netty.channel.ChannelHandlerContext;
  14. import io.netty.channel.ChannelInboundHandlerAdapter;
  15. import java.util.Map;
  16. import newlandframework.netty.rpc.model.MessageRequest;
  17. import newlandframework.netty.rpc.model.MessageResponse;
  19. public class MessageRecvHandler extends ChannelInboundHandlerAdapter {
  21.     private final Map<String, Object> handlerMap;
  23.     public MessageRecvHandler(Map<String, Object> handlerMap) {
  24.         this.handlerMap = handlerMap;
  25.     }
  27.     public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
  28.         MessageRequest request = (MessageRequest) msg;
  29.         MessageResponse response = new MessageResponse();
  30.         MessageRecvInitializeTask recvTask = new MessageRecvInitializeTask(request, response, handlerMap, ctx);
  31.         //不要阻塞nio线程，复杂的业务逻辑丢给专门的线程池
  32.         MessageRecvExecutor.submit(recvTask);
  33.     }
  35.     public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
  36.         //网络有异常要关闭通道
  37.         ctx.close();
  38.     }
  39. }

复制代码

复制代码

复制代码

复制代码

  1. /**
  2.  * @filename:MessageRecvInitializeTask.java
  3.  *
  4.  * Newland Co. Ltd. All rights reserved.
  5.  *
  6.  * @Description:Rpc服务器消息线程任务处理
  7.  * @author tangjie
  8.  * @version 1.0
  9.  *
  10.  */
  11. package newlandframework.netty.rpc.core;
  13. import io.netty.channel.ChannelFuture;
  14. import io.netty.channel.ChannelFutureListener;
  15. import io.netty.channel.ChannelHandlerContext;
  16. import java.util.Map;
  17. import newlandframework.netty.rpc.model.MessageRequest;
  18. import newlandframework.netty.rpc.model.MessageResponse;
  19. import org.apache.commons.beanutils.MethodUtils;
  21. public class MessageRecvInitializeTask implements Runnable {
  23.     private MessageRequest request = null;
  24.     private MessageResponse response = null;
  25.     private Map<String, Object> handlerMap = null;
  26.     private ChannelHandlerContext ctx = null;
  28.     public MessageResponse getResponse() {
  29.         return response;
  30.     }
  32.     public MessageRequest getRequest() {
  33.         return request;
  34.     }
  36.     public void setRequest(MessageRequest request) {
  37.         this.request = request;
  38.     }
  40.     MessageRecvInitializeTask(MessageRequest request, MessageResponse response, Map<String, Object> handlerMap, ChannelHandlerContext ctx) {
  41.         this.request = request;
  42.         this.response = response;
  43.         this.handlerMap = handlerMap;
  44.         this.ctx = ctx;
  45.     }
  47.     public void run() {
  48.         response.setMessageId(request.getMessageId());
  49.         try {
  50.             Object result = reflect(request);
  51.             response.setResult(result);
  52.         } catch (Throwable t) {
  53.             response.setError(t.toString());
  54.             t.printStackTrace();
  55.             System.err.printf("RPC Server invoke error!\n");
  56.         }
  58.         ctx.writeAndFlush(response).addListener(new ChannelFutureListener() {
  59.             public void operationComplete(ChannelFuture channelFuture) throws Exception {
  60.                 System.out.println("RPC Server Send message-id respone:" + request.getMessageId());
  61.             }
  62.         });
  63.     }
  65.     private Object reflect(MessageRequest request) throws Throwable {
  66.         String className = request.getClassName();
  67.         Object serviceBean = handlerMap.get(className);
  68.         String methodName = request.getMethodName();
  69.         Object[] parameters = request.getParameters();
  70.         return MethodUtils.invokeMethod(serviceBean, methodName, parameters);
  71.     }
  72. }

复制代码

复制代码

　　然后是RPC消息处理的回调实现模块代码

复制代码

复制代码

  1. /**
  2.  * @filename:MessageCallBack.java
  3.  *
  4.  * Newland Co. Ltd. All rights reserved.
  5.  *
  6.  * @Description:Rpc消息回调
  7.  * @author tangjie
  8.  * @version 1.0
  9.  *
  10.  */
  11. package newlandframework.netty.rpc.core;
  13. import java.util.concurrent.TimeUnit;
  14. import java.util.concurrent.locks.Condition;
  15. import java.util.concurrent.locks.Lock;
  16. import java.util.concurrent.locks.ReentrantLock;
  17. import newlandframework.netty.rpc.model.MessageRequest;
  18. import newlandframework.netty.rpc.model.MessageResponse;
  20. public class MessageCallBack {
  22.     private MessageRequest request;
  23.     private MessageResponse response;
  24.     private Lock lock = new ReentrantLock();
  25.     private Condition finish = lock.newCondition();
  27.     public MessageCallBack(MessageRequest request) {
  28.         this.request = request;
  29.     }
  31.     public Object start() throws InterruptedException {
  32.         try {
  33.             lock.lock();
  34.             //设定一下超时时间，rpc服务器太久没有相应的话，就默认返回空吧。
  35.             finish.await(10*1000, TimeUnit.MILLISECONDS);
  36.             if (this.response != null) {
  37.                 return this.response.getResult();
  38.             } else {
  39.                 return null;
  40.             }
  41.         } finally {
  42.             lock.unlock();
  43.         }
  44.     }
  46.     public void over(MessageResponse reponse) {
  47.         try {
  48.             lock.lock();
  49.             finish.signal();
  50.             this.response = reponse;
  51.         } finally {
  52.             lock.unlock();
  53.         }
  54.     }
  55. }

复制代码

复制代码

　　到此为止，NettyRPC的关键部分：服务端、客户端的模块已经通过Netty全部实现了。现在给出spring加载配置rpc-invoke-config.xml的内容：

复制代码

复制代码

  1. <?xml version="1.0" encoding="UTF-8"?>
  2. <beans xmlns="http://www.springframework.org/schema/beans"
  3.        xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
  4.        xmlns:context="http://www.springframework.org/schema/context"
  5.        xsi:schemaLocation="http://www.springframework.org/schema/beans
  6.        http://www.springframework.org/schema/beans/spring-beans.xsd
  7.        http://www.springframework.org/schema/context
  8.        http://www.springframework.org/schema/context/spring-context.xsd">
  9.   <context:component-scan base-package="newlandframework.netty.rpc.core"/>
  10.   <context:property-placeholder location="classpath:newlandframework/netty/rpc/config/rpc-server.properties"/>
  11.   <bean id="rpcbean" class="newlandframework.netty.rpc.model.MessageKeyVal">
  12.     <property name="messageKeyVal">
  13.       <map>
  14.         <entry key="newlandframework.netty.rpc.servicebean.Calculate">
  15.           <ref bean="calc"/>
  16.         </entry>
  17.       </map>
  18.     </property>
  19.   </bean>
  20.   <bean id="calc" class="newlandframework.netty.rpc.servicebean.CalculateImpl"/>
  21.   <bean id="rpcServer" class="newlandframework.netty.rpc.core.MessageRecvExecutor">
  22.     <constructor-arg name="serverAddress" value="${rpc.server.addr}"/>
  23.   </bean>
  24. </beans>

复制代码

复制代码

　　再贴出RPC服务绑定ip信息的配置文件：rpc-server.properties的内容。

  1. #rpc server's ip address config
  2. rpc.server.addr=127.0.0.1:18888

　　最后NettyRPC服务端启动方式参考如下：

  1. new ClassPathXmlApplicationContext("newlandframework/netty/rpc/config/rpc-invoke-config.xml");

　　如果一切顺利，没有出现意外的话，控制台上面，会出现如下截图所示的情况：

913759-20160625014824375-1728878897.jpg

　　如果出现了，说明NettyRPC服务器，已经启动成功！

　　上面基于Netty的RPC服务器，并发处理性能如何呢？实践是检验真理的唯一标准，下面我们就来实战一下。

　　下面的测试案例，是基于RPC远程调用两数相加函数，并返回计算结果。客户端同时开1W个线程，同一时刻，瞬时发起并发计算请求，然后观察Netty的RPC服务器是否有正常应答回复响应，以及客户端是否有正常返回调用计算结果。值得注意的是，测试案例是基于1W个线程瞬时并发请求而设计的，并不是1W个线程循环发起请求。这两者对于衡量RPC服务器的并发处理性能，还是有很大差别的。当然，前者对于并发性能的处理要求，要高上很多很多。

　　现在，先给出RPC计算接口、RPC计算接口实现类的代码实现：

复制代码

复制代码

  1. /**
  2.  * @filename:Calculate.java
  3.  *
  4.  * Newland Co. Ltd. All rights reserved.
  5.  *
  6.  * @Description:计算器定义接口
  7.  * @author tangjie
  8.  * @version 1.0
  9.  *
  10.  */
  11. package newlandframework.netty.rpc.servicebean;
  13. public interface Calculate {
  14.     //两数相加
  15.     int add(int a, int b);
  16. }

复制代码

复制代码

复制代码

复制代码

  1. /**
  2.  * @filename:CalculateImpl.java
  3.  *
  4.  * Newland Co. Ltd. All rights reserved.
  5.  *
  6.  * @Description:计算器定义接口实现
  7.  * @author tangjie
  8.  * @version 1.0
  9.  *
  10.  */
  11. package newlandframework.netty.rpc.servicebean;
  13. public class CalculateImpl implements Calculate {
  14.     //两数相加
  15.     public int add(int a, int b) {
  16.         return a + b;
  17.     }
  18. }

复制代码

复制代码

　　下面是瞬时并发RPC请求的测试样例：

复制代码

复制代码

  1. /**
  2.  * @filename:CalcParallelRequestThread.java
  3.  *
  4.  * Newland Co. Ltd. All rights reserved.
  5.  *
  6.  * @Description:并发线程模拟
  7.  * @author tangjie
  8.  * @version 1.0
  9.  *
  10.  */
  11. package newlandframework.netty.rpc.servicebean;
  13. import newlandframework.netty.rpc.core.MessageSendExecutor;
  14. import java.util.concurrent.CountDownLatch;
  15. import java.util.logging.Level;
  16. import java.util.logging.Logger;
  18. public class CalcParallelRequestThread implements Runnable {
  20.     private CountDownLatch signal;
  21.     private CountDownLatch finish;
  22.     private MessageSendExecutor executor;
  23.     private int taskNumber = 0;
  25.     public CalcParallelRequestThread(MessageSendExecutor executor, CountDownLatch signal, CountDownLatch finish, int taskNumber) {
  26.         this.signal = signal;
  27.         this.finish = finish;
  28.         this.taskNumber = taskNumber;
  29.         this.executor = executor;
  30.     }
  32.     public void run() {
  33.         try {
  34.             signal.await();
  36.             Calculate calc = executor.execute(Calculate.class);
  37.             int add = calc.add(taskNumber, taskNumber);
  38.             System.out.println("calc add result:[" + add + "]");
  40.             finish.countDown();
  41.         } catch (InterruptedException ex) {
  42.             Logger.getLogger(CalcRequestThread.class.getName()).log(Level.SEVERE, null, ex);
  43.         }
  44.     }
  45. }

复制代码

复制代码

复制代码

复制代码

  1. /**
  2.  * @filename:RpcParallelTest.java
  3.  *
  4.  * Newland Co. Ltd. All rights reserved.
  5.  *
  6.  * @Description:rpc并发测试代码
  7.  * @author tangjie
  8.  * @version 1.0
  9.  *
  10.  */
  11. package newlandframework.netty.rpc.servicebean;
  13. import java.util.concurrent.CountDownLatch;
  14. import newlandframework.netty.rpc.core.MessageSendExecutor;
  15. import org.apache.commons.lang.time.StopWatch;
  17. public class RpcParallelTest {
  19.     public static void main(String[] args) throws Exception {
  20.         final MessageSendExecutor executor = new MessageSendExecutor("127.0.0.1:18888");
  21.         //并行度10000
  22.         int parallel = 10000;
  24.         //开始计时
  25.         StopWatch sw = new StopWatch();
  26.         sw.start();
  28.         CountDownLatch signal = new CountDownLatch(1);
  29.         CountDownLatch finish = new CountDownLatch(parallel);
  31.         for (int index = 0; index < parallel; index++) {
  32.             CalcParallelRequestThread client = new CalcParallelRequestThread(executor, signal, finish, index);
  33.             new Thread(client).start();
  34.         }
  36.         //10000个并发线程瞬间发起请求操作
  37.         signal.countDown();
  38.         finish.await();
  40.         sw.stop();
  42.         String tip = String.format("RPC调用总共耗时: [%s] 毫秒", sw.getTime());
  43.         System.out.println(tip);
  45.         executor.stop();
  46.     }
  47. }

复制代码

复制代码

　　好了，现在先启动NettyRPC服务器，确认没有问题之后，运行并发RPC请求客户端，看下客户端打印的计算结果，以及处理耗时。

  1.  ![913759-20160625090441641-916118510.jpg][]

　　从上面来看，10000个瞬时RPC计算请求，总共耗时接近11秒。我们在来看下NettyRPC的服务端运行情况，如下所示：

  1.  ![913759-20160625090549563-465927784.jpg][]

　　可以很清楚地看到，RPC服务端都有收到客户端发起的RPC计算请求，并返回消息应答。

　　最后我们还是要分别验证一下，RPC服务端是否存在丢包、粘包、IO阻塞的情况？1W个并发计算请求，是否成功接收处理并应答了？实际情况说明一切，看下图所示：

  1.   ![913759-20160625090826953-1445487995.jpg][]

　　非常给力，RPC的服务端确实成功接收到了客户端发起的1W笔瞬时并发计算请求，并且成功应答处理了。并没有出现：丢包、粘包、IO阻塞的情况。再看下RPC客户端，是否成功得到计算结果的应答返回了呢？

　　913759-20160625091005391-1482737387.jpg

　　很好，RPC的客户端，确实收到了RPC服务端计算的1W笔加法请求的计算结果，而且耗时接近11秒。由此可见，基于Netty+业务线程池的NettyRPC服务器，应对并发多线程RPC请求，处理起来是得心应手，游刃有余！

　　最后，本文通过Netty这个NIO框架，实现了一个很简单的“高性能”的RPC服务器，代码虽然写出来了，但是还是有一些值得改进的地方，比如：

　　1、对象序列化传输可以支持目前主流的序列化框架：protobuf、JBoss Marshalling、Avro等等。

　　2、Netty的线程模型可以根据业务需求，进行定制。因为，并不是每笔业务都需要这么强大的并发处理性能。

　　3、目前RPC计算只支持一个RPC服务接口映射绑定一个对应的实现，后续要支持一对多的情况。

　　4、业务线程池的启动参数、线程池并发阻塞容器模型等等，可以配置化管理。

　　5、Netty的Handler处理部分，对于复杂的业务逻辑，现在是统一分派到特定的线程池进行后台异步处理。当然你还可以考虑JMS（消息队列）方式进行解耦，统一分派给消息队列的订阅者，统一处理。目前实现JMS的开源框架也有很多，ActiveMQ、RocketMQ等等，都可以考虑。

　　本文实现的NettyRPC，对于面前的您而言，一定还有很多地方，可以加以完善和改进，优化改进的工作就交给您自由发挥了。

　　由于本人技术能力、认知水平有限。本文中有说不对的地方，恳请园友们批评指正！不吝赐教！最后，感谢面前的您，耐心的阅读完本文，相信现在的你，对于Java开发高性能的服务端应用，又有了一个更深入的了解！本文算是对我Netty学习成果的阶段性总结，后续有时间，我还会继续推出Netty工业级开发的相关文章，敬请期待！

发表评论

表情：
评论列表 （有 0 条评论，359人围观）

还没有评论，来说两句吧...

相关阅读

    相关 netty-rpc

    一、什么是RPC框架？ > RPC，全称为Remote Procedure Call，即远程过程调用，是一种计算机通信协议。 > 比如现在有两台机器：A机器和B机...

    痛定思痛。痛定思痛。/ 0 赞/ 161 阅读

    相关 Netty高性能的RPC框架

    Netty是一个开源的、高性能的、异步事件驱动的网络通信框架，支持多种协议和编码解码器，能够帮助开发人员快速构建高性能、可扩展的网络应用程序。它的主要优势包括： 1. 异步

    谁践踏了优雅谁践踏了优雅/ 0 赞/ 190 阅读