Netty与SpringBoot的整合实现

最近有朋友向我询问一些Netty与SpringBoot整合的相关问题，这里，我就总结了一下基本整合流程，也就是说，这篇文章，默认大家是对netty与Spring，SpringMVC的整合是没有什么问题的。现在，就进入正题吧。

创新互联建站主要为客户提供服务项目涵盖了网页视觉设计、VI标志设计、营销型网站建设、网站程序开发、HTML5响应式重庆网站建设公司、手机网站制作设计、微商城、网站托管及成都网站维护、WEB系统开发、域名注册、国内外服务器租用、视频、平面设计、SEO优化排名。设计、前端、后端三个建站步骤的完善服务体系。一人跟踪测试的建站服务标准。已经为凿毛机行业客户提供了网站推广服务。

Server端：

总的来说，服务端还是比较简单的，自己一共写了三个核心类。分别是

NettyServerListener：服务启动监听器
ServerChannelHandlerAdapter：通道适配器，主要用于多线程共享
RequestDispatcher：请求分排器

下面开始集成过程：

在pom.xml中添加以下依赖


 io.netty
 netty-all
 5.0.0.Alpha2


 org.springframework.boot
 spring-boot-configuration-processor
 true

让SpringBoot的启动类实现CommandLineRunner接口并重写run方法,比如我的启动类是CloudApplication.java

@SpringBootApplication
public class CloudApplication implements CommandLineRunner {

 public static void main(String[] args) {
 SpringApplication.run(CloudApplication.class, args);
 }

 @Override
 public void run(String... strings) {
 }
}

创建类NettyServerListener.java

// 读取yml的一个配置类
import com.edu.hart.modules.constant.NettyConfig;
// Netty连接信息配置类
import com.edu.hart.modules.constant.NettyConstant;
// 
import com.edu.hart.rpc.util.ObjectCodec;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.codec.LengthFieldBasedFrameDecoder;
import io.netty.handler.codec.LengthFieldPrepender;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Component;

import javax.annotation.PreDestroy;
import javax.annotation.Resource;

/**
 * 服务启动监听器
 *
 * @author 叶云轩
 */
@Component
public class NettyServerListener {
 /**
 * NettyServerListener 日志输出器
 *
 * @author 叶云轩 create by 2017/10/31 18:05
 */
 private static final Logger LOGGER = LoggerFactory.getLogger(NettyServerListener.class);
 /**
 * 创建bootstrap
 */
 ServerBootstrap serverBootstrap = new ServerBootstrap();
 /**
 * BOSS
 */
 EventLoopGroup boss = new NioEventLoopGroup();
 /**
 * Worker
 */
 EventLoopGroup work = new NioEventLoopGroup();
 /**
 * 通道适配器
 */
 @Resource
 private ServerChannelHandlerAdapter channelHandlerAdapter;
 /**
 * NETT服务器配置类
 */
 @Resource
 private NettyConfig nettyConfig;

 /**
 * 关闭服务器方法
 */
 @PreDestroy
 public void close() {
 LOGGER.info("关闭服务器....");
 //优雅退出
 boss.shutdownGracefully();
 work.shutdownGracefully();
 }

 /**
 * 开启及服务线程
 */
 public void start() {
 // 从配置文件中(application.yml)获取服务端监听端口号
 int port = nettyConfig.getPort();
 serverBootstrap.group(boss, work)
  .channel(NioServerSocketChannel.class)
  .option(ChannelOption.SO_BACKLOG, 100)
  .handler(new LoggingHandler(LogLevel.INFO));
 try {
  //设置事件处理
  serverBootstrap.childHandler(new ChannelInitializer() {
  @Override
  protected void initChannel(SocketChannel ch) throws Exception {
   ChannelPipeline pipeline = ch.pipeline();
   pipeline.addLast(new LengthFieldBasedFrameDecoder(nettyConfig.getMaxFrameLength()
    , 0, 2, 0, 2));
   pipeline.addLast(new LengthFieldPrepender(2));
   pipeline.addLast(new ObjectCodec());

   pipeline.addLast(channelHandlerAdapter);
  }
  });
  LOGGER.info("netty服务器在[{}]端口启动监听", port);
  ChannelFuture f = serverBootstrap.bind(port).sync();
  f.channel().closeFuture().sync();
 } catch (InterruptedException e) {
  LOGGER.info("[出现异常] 释放资源");
  boss.shutdownGracefully();
  work.shutdownGracefully();
 }
 }
}

创建类ServerChannelHandlerAdapter.java - 通道适配器

// 记录调用方法的元信息的类
import com.edu.hart.rpc.entity.MethodInvokeMeta;
import io.netty.channel.ChannelHandler.Sharable;
import io.netty.channel.ChannelHandlerAdapter;
import io.netty.channel.ChannelHandlerContext;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;

/**
 * 多线程共享
 */
@Component
@Sharable
public class ServerChannelHandlerAdapter extends ChannelHandlerAdapter {
 /**
 * 日志处理
 */
 private Logger logger = LoggerFactory.getLogger(ServerChannelHandlerAdapter.class);
 /**
 * 注入请求分排器
 */
 @Resource
 private RequestDispatcher dispatcher;

 @Override
 public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
 cause.printStackTrace();
 ctx.close();
 }

 @Override
 public void channelRead(ChannelHandlerContext ctx, Object msg) {
 MethodInvokeMeta invokeMeta = (MethodInvokeMeta) msg;
 // 屏蔽toString()方法
 if (invokeMeta.getMethodName().endsWith("toString()")
  && !"class java.lang.String".equals(invokeMeta.getReturnType().toString()))
  logger.info("客户端传入参数 :{},返回值：{}",
   invokeMeta.getArgs(), invokeMeta.getReturnType());
 dispatcher.dispatcher(ctx, invokeMeta);
 }
}

RequestDispatcher.java

// 封装的返回信息枚举类
import com.edu.hart.modules.communicate.ResponseCodeEnum;
// 封装的返回信息实体类
import com.edu.hart.modules.communicate.ResponseResult;
// 封装的连接常量类
import com.edu.hart.modules.constant.NettyConstant;
// 记录元方法信息的实体类
import com.edu.hart.rpc.entity.MethodInvokeMeta;
// 对于返回值为空的一个处理
import com.edu.hart.rpc.entity.NullWritable;
// 封装的返回信息实体工具类
import com.edu.hart.rpc.util.ResponseResultUtil;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelHandlerContext;
import org.springframework.beans.BeansException;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.stereotype.Component;

import java.lang.reflect.Method;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * 请求分排器
 */
@Component
public class RequestDispatcher implements ApplicationContextAware {
 private ExecutorService executorService = Executors.newFixedThreadPool(NettyConstant.getMaxThreads());
 private ApplicationContext app;

 /**
 * 发送
 *
 * @param ctx
 * @param invokeMeta
 */
 public void dispatcher(final ChannelHandlerContext ctx, final MethodInvokeMeta invokeMeta) {
 executorService.submit(() -> {
  ChannelFuture f = null;
  try {
  Class<?> interfaceClass = invokeMeta.getInterfaceClass();
  String name = invokeMeta.getMethodName();
  Object[] args = invokeMeta.getArgs();
  Class<?>[] parameterTypes = invokeMeta.getParameterTypes();
  Object targetObject = app.getBean(interfaceClass);
  Method method = targetObject.getClass().getMethod(name, parameterTypes);
  Object obj = method.invoke(targetObject, args);
  if (obj == null) {
   f = ctx.writeAndFlush(NullWritable.nullWritable());
  } else {
   f = ctx.writeAndFlush(obj);
  }
  f.addListener(ChannelFutureListener.CLOSE);
  } catch (Exception e) {
  ResponseResult error = ResponseResultUtil.error(ResponseCodeEnum.SERVER_ERROR);
  f = ctx.writeAndFlush(error);
  } finally {
  f.addListener(ChannelFutureListener.CLOSE);
  }
 });
 }

 /**
 * 加载当前application.xml
 *
 * @param ctx
 * @throws BeansException
 */
 public void setApplicationContext(ApplicationContext ctx) throws BeansException {
 this.app = ctx;
 }
}

application.yml文件中对于netty的一个配置

netty:
 port: 11111

NettyConfig.java

import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.stereotype.Component;

/**
 * 读取yml配置文件中的信息
 * Created by 叶云轩 on 2017/10/31 - 18:38
 * Concat tdg_yyx@foxmail.com
 */
@Component
@ConfigurationProperties(prefix = "netty")
public class NettyConfig {

 private int port;

 public int getPort() {
 return port;
 }

 public void setPort(int port) {
 this.port = port;
 }
}

NettyConstanct.java

import org.springframework.stereotype.Component;

/**
 * Netty服务器常量
 * Created by 叶云轩 on 2017/10/31 - 17:47
 * Concat tdg_yyx@foxmail.com
 */
@Component
public class NettyConstant {

 /**
 * 最大线程量
 */
 private static final int MAX_THREADS = 1024;
 /**
 * 数据包最大长度
 */
 private static final int MAX_FRAME_LENGTH = 65535;

 public static int getMaxFrameLength() {
 return MAX_FRAME_LENGTH;
 }

 public static int getMaxThreads() {
 return MAX_THREADS;
 }
}

至此，netty服务端算是与SpringBoot整合成功。那么看一下启动情况吧。

Netty与Spring Boot的整合实现

Client端：

Client我感觉要比Server端要麻烦一点。这里还是先给出核心类吧。

NettyClient ： netty客户端
ClientChannelHandlerAdapter ：客户端通道适配器
CustomChannelInitalizer：自定义通道初始化工具
RPCProxyFactoryBean：RPC通信代理工厂

在Client端里。SpringBoot的启动类要继承SpringBootServletInitializer这个类，并覆盖SpringApplicationBuilder方法

import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.boot.builder.SpringApplicationBuilder;
import org.springframework.boot.web.support.SpringBootServletInitializer;

@SpringBootApplication
public class OaApplication extends SpringBootServletInitializer {

 public static void main(String[] args) {
 SpringApplication.run(OaApplication.class, args);
 }

 @Override
 protected SpringApplicationBuilder configure(SpringApplicationBuilder builder) {
 return builder.sources(OaApplication.class);
 }
}

NettyClient.java

// 记录元方法信息的实体类
import com.edu.hart.rpc.entity.MethodInvokeMeta;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import javax.management.MBeanServer;

/**
 * 客户端发送类
 * Created by 叶云轩 on 2017/6/16-16:58
 * Concat tdg_yyx@foxmail.com
 */
public class NettyClient {

 private Logger logger = LoggerFactory.getLogger(MBeanServer.class);
 private Bootstrap bootstrap;
 private EventLoopGroup worker;
 private int port;
 private String url;
 private int MAX_RETRY_TIMES = 10;

 public NettyClient(String url, int port) {
 this.url = url;
 this.port = port;
 bootstrap = new Bootstrap();
 worker = new NioEventLoopGroup();
 bootstrap.group(worker);
 bootstrap.channel(NioSocketChannel.class);
 }

 public void close() {
 logger.info("关闭资源");
 worker.shutdownGracefully();
 }

 public Object remoteCall(final MethodInvokeMeta cmd, int retry) {
 try {
  CustomChannelInitializerClient customChannelInitializer = new CustomChannelInitializerClient(cmd);
  bootstrap.handler(customChannelInitializer);
  ChannelFuture sync = bootstrap.connect(url, port).sync();
  sync.channel().closeFuture().sync();
  Object response = customChannelInitializer.getResponse();
  return response;
 } catch (InterruptedException e) {
  retry++;
  if (retry > MAX_RETRY_TIMES) {
  throw new RuntimeException("调用Wrong");
  } else {
  try {
   Thread.sleep(100);
  } catch (InterruptedException e1) {
   e1.printStackTrace();
  }
  logger.info("第{}次尝试....失败", retry);
  return remoteCall(cmd, retry);
  }
 }
 }
}

ClientChannelHandlerAdapter.java

import com.edu.hart.rpc.entity.MethodInvokeMeta;
import io.netty.channel.ChannelHandlerAdapter;
import io.netty.channel.ChannelHandlerContext;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * Created by 叶云轩 on 2017/6/16-17:03
 * Concat tdg_yyx@foxmail.com
 */
public class ClientChannelHandlerAdapter extends ChannelHandlerAdapter {
 private Logger logger = LoggerFactory.getLogger(ClientChannelHandlerAdapter.class);
 private MethodInvokeMeta methodInvokeMeta;
 private CustomChannelInitializerClient channelInitializerClient;

 public ClientChannelHandlerAdapter(MethodInvokeMeta methodInvokeMeta, CustomChannelInitializerClient channelInitializerClient) {
 this.methodInvokeMeta = methodInvokeMeta;
 this.channelInitializerClient = channelInitializerClient;
 }

 @Override
 public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
 logger.info("客户端出异常了,异常信息:{}", cause.getMessage());
 cause.printStackTrace();
 ctx.close();
 }

 @Override
 public void channelActive(ChannelHandlerContext ctx) throws Exception {
 if (methodInvokeMeta.getMethodName().endsWith("toString") && !"class java.lang.String".equals(methodInvokeMeta.getReturnType().toString()))
  logger.info("客户端发送信息参数:{},信息返回值类型：{}", methodInvokeMeta.getArgs(), methodInvokeMeta.getReturnType());
 ctx.writeAndFlush(methodInvokeMeta);

 }

 @Override
 public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
 channelInitializerClient.setResponse(msg);
 }
}

CustomChannelInitializerClient.java

import com.edu.hart.rpc.entity.MethodInvokeMeta;
import com.edu.hart.rpc.entity.NullWritable;
import com.edu.hart.rpc.util.ObjectCodec;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.socket.SocketChannel;
import io.netty.handler.codec.LengthFieldBasedFrameDecoder;
import io.netty.handler.codec.LengthFieldPrepender;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;/** 
Created by 叶云轩 on 2017/6/16-15:01 
Concat tdg_yyx@foxmail.com
*/
public class CustomChannelInitializerClient extends ChannelInitializer { 
 private Logger logger = LoggerFactory.getLogger(CustomChannelInitializerClient.class);

 private MethodInvokeMeta methodInvokeMeta;

 private Object response;

 public CustomChannelInitializerClient(MethodInvokeMeta methodInvokeMeta) {
 if (!"toString".equals(methodInvokeMeta.getMethodName())) {
  logger.info("[CustomChannelInitializerClient] 调用方法名：{}，入参：{},参数类型：{}，返回值类型{}"
   , methodInvokeMeta.getMethodName()
   , methodInvokeMeta.getArgs()
   , methodInvokeMeta.getParameterTypes()
   , methodInvokeMeta.getReturnType());
 }
 this.methodInvokeMeta = methodInvokeMeta;
 }

 public Object getResponse() {
 if (response instanceof NullWritable) {
  return null;
 }
 return response;
 }

 public void setResponse(Object response) {
 this.response = response;
 }

 @Override
 protected void initChannel(SocketChannel ch) {
 ChannelPipeline pipeline = ch.pipeline();
 pipeline.addLast(new LengthFieldPrepender(2));
 pipeline.addLast(new LengthFieldBasedFrameDecoder(1024 * 1024, 0, 2, 0, 2));
 pipeline.addLast(new ObjectCodec());
 pipeline.addLast(new ClientChannelHandlerAdapter(methodInvokeMeta, this));
 }}

4. RPCProxyFactoryBean.java

import com.edu.hart.rpc.entity.MethodInvokeMeta;
import com.edu.hart.rpc.util.WrapMethodUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.config.AbstractFactoryBean;

import java.lang.reflect.InvocationHandler;
import java.lang.reflect.Method;
import java.lang.reflect.Proxy;

/**
* Created by 叶云轩 on 2017/6/16-17:16
* Concat tdg_yyx@foxmail.com
*/
public class RPCProxyFactoryBean extends AbstractFactoryBean implements InvocationHandler {
private Logger logger = LoggerFactory.getLogger(RPCProxyFactoryBean.class);
 private Class interfaceClass;

 private NettyClient nettyClient;

 @Override
 public Class<?> getObjectType() {
 return interfaceClass;
 }

 @Override
 protected Object createInstance() throws Exception {
 logger.info("[代理工厂] 初始化代理Bean : {}", interfaceClass);
 return Proxy.newProxyInstance(interfaceClass.getClassLoader(), new Class[]{interfaceClass}, this);
 }

 @Override
 public Object invoke(Object proxy, Method method, Object[] args) {
 final MethodInvokeMeta methodInvokeMeta = WrapMethodUtils.readMethod(interfaceClass, method, args);
 if (!methodInvokeMeta.getMethodName().equals("toString")) {
  logger.info("[invoke] 调用接口{},调用方法名：{}，入参：{},参数类型：{}，返回值类型{}",
   methodInvokeMeta.getInterfaceClass(), methodInvokeMeta.getMethodName()
   , methodInvokeMeta.getArgs(), methodInvokeMeta.getParameterTypes(), methodInvokeMeta.getReturnType());
 }
 return nettyClient.remoteCall(methodInvokeMeta, 0);
 }

 public void setInterfaceClass(Class interfaceClass) {
 this.interfaceClass = interfaceClass;
 }

 public void setNettyClient(NettyClient nettyClient) {
 this.nettyClient = nettyClient;
 }
}

至此，netty-client与SpringBoot的集成了算完毕了。同样，在netty-client中也要加入相应的依赖

不过上面server与client使用了一些公共的类和工具。下面也给列举中出来。

MethodInvokeMeta.java

import org.springframework.stereotype.Component;

import java.io.Serializable;

/**
* 记录调用方法的元信息
* Created by 叶云轩 on 2017/6/7-15:41
* Concat tdg_yyx@foxmail.com
*/
@Component
public class MethodInvokeMeta implements Serializable {

 private static final long serialVersionUID = 8379109667714148890L;
 //接口
 private Class<?> interfaceClass;
 //方法名
 private String methodName;
 //参数
 private Object[] args;
 //返回值类型
 private Class<?> returnType;
 //参数类型
 private Class<?>[] parameterTypes;

 public Object[] getArgs() {
 return args;
 }

 public void setArgs(Object[] args) {
 this.args = args;
 }

 public Class<?> getInterfaceClass() {
 return interfaceClass;
 }

 public void setInterfaceClass(Class<?> interfaceClass) {
 this.interfaceClass = interfaceClass;
 }

 public String getMethodName() {
 return methodName;
 }

 public void setMethodName(String methodName) {
 this.methodName = methodName;
 }

 public Class[] getParameterTypes() {
 return parameterTypes;
 }

 public void setParameterTypes(Class<?>[] parameterTypes) {
 this.parameterTypes = parameterTypes;
 }

 public Class getReturnType() {
 return returnType;
 }

 public void setReturnType(Class returnType) {
 this.returnType = returnType;
 }
}

NullWritable.java

import java.io.Serializable;

/**
* 服务器可能返回空的处理
* Created by 叶云轩 on 2017/6/16-16:46
* Concat tdg_yyx@foxmail.com
*/
public class NullWritable implements Serializable {

 private static final long serialVersionUID = -8191640400484155111L;
 private static NullWritable instance = new NullWritable();

 private NullWritable() {
 }

 public static NullWritable nullWritable() {
 return instance;
 }
}

ObjectCodec.java

import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.MessageToMessageCodec;

import java.util.List;

public class ObjectCodec extends MessageToMessageCodec {

 @Override
 protected void encode(ChannelHandlerContext ctx, Object msg, List

重庆分公司，新征程启航

Netty与SpringBoot的整合实现

其他资讯