devbox
笔触狂放9
这个屌丝很懒，什么也没留下！
关注作者
热门标签
热门文章
当前位置:   article > 正文

springboot集成netty实战

作者：笔触狂放9 | 2024-03-03 19:29:36

赞

踩

springboot集成netty

今天我们分享springboot集成netty的过程：

1、jar包依赖，Netty服务端和客户端依赖的jar包一样：

  1. <dependency>
  2.     <groupId>io.netty</groupId>
  3.     <artifactId>netty-all</artifactId>
  4.     <version>4.1.28.Final</version>
  5. </dependency>
  6. <!--  序列化jar -->
  7. <dependency>
  8.     <groupId>de.javakaffee</groupId>
  9.     <artifactId>kryo-serializers</artifactId>
  10.     <version>0.42</version>
  11. </dependency>

2、Netty服务端核心代码(boot启动类省略):

2.1、netty功能启动入口

  1.  
  2. import io.netty.channel.ChannelInitializer;
  3. import io.netty.channel.socket.SocketChannel;
  4. import io.netty.handler.codec.LengthFieldBasedFrameDecoder;
  5. import io.netty.handler.codec.LengthFieldPrepender;
  6. import io.netty.handler.timeout.ReadTimeoutHandler;
  7. import org.springframework.beans.factory.annotation.Autowired;
  8. import org.springframework.stereotype.Service;
  9.  
  10. /**
  11.  * 类说明：服务端Handler的初始化
  12.  * 交给Spring 托管,ServerBusiHandler用注入方式实例化后加入Netty的pipeline
  13.  */
  14. @Service//启动时扫描，即交给spring管理
  15. public class ServerInit extends ChannelInitializer<SocketChannel> {
  16.  
  17.     @Autowired
  18.     private ServerBusiHandler serverBusiHandler;
  19.  
  20.     @Override
  21.     protected void initChannel(SocketChannel ch) throws Exception {
  22.         /*Netty提供的日志打印Handler，可以展示发送接收出去的字节*/
  23.         //ch.pipeline().addLast(new LoggingHandler(LogLevel.INFO));
  24.         /*剥离接收到的消息的长度字段，拿到实际的消息报文的字节数组*/
  25.         ch.pipeline().addLast("frameDecoder",
  26.                 new LengthFieldBasedFrameDecoder(65535,
  27.                         0,2,0,
  28.                         2));
  29.         /*给发送出去的消息增加长度字段*/
  30.         ch.pipeline().addLast("frameEncoder",
  31.                 new LengthFieldPrepender(2));
  32.         /*反序列化，将字节数组转换为消息实体*/
  33.         ch.pipeline().addLast(new KryoDecoder());
  34.         /*序列化，将消息实体转换为字节数组准备进行网络传输*/
  35.         ch.pipeline().addLast("MessageEncoder",
  36.                 new KryoEncoder());
  37.         /*超时检测*/
  38.         ch.pipeline().addLast("readTimeoutHandler",
  39.                 new ReadTimeoutHandler(50));
  40.         /*登录应答*/
  41.         ch.pipeline().addLast(new LoginAuthRespHandler());
  42.  
  43.         /*心跳应答*/
  44.         ch.pipeline().addLast("HeartBeatHandler",
  45.                 new HeartBeatRespHandler());
  46.  
  47.         /*服务端业务处理*/
  48.         ch.pipeline().addLast("ServerBusiHandler",
  49.                 serverBusiHandler);
  50.     }
  51. }

2.2、netty自定义配置入口 ServerInit

  1.  
  2. import io.netty.channel.ChannelInitializer;
  3. import io.netty.channel.socket.SocketChannel;
  4. import io.netty.handler.codec.LengthFieldBasedFrameDecoder;
  5. import io.netty.handler.codec.LengthFieldPrepender;
  6. import io.netty.handler.timeout.ReadTimeoutHandler;
  7. import org.springframework.beans.factory.annotation.Autowired;
  8. import org.springframework.stereotype.Service;
  9.  
  10. /**
  11.  * 类说明：服务端Handler的初始化
  12.  * 交给Spring 托管,ServerBusiHandler用注入方式实例化后加入Netty的pipeline
  13.  */
  14. @Service//启动时扫描，即交给spring管理
  15. public class ServerInit extends ChannelInitializer<SocketChannel> {
  16.  
  17.     @Autowired
  18.     private ServerBusiHandler serverBusiHandler;
  19.  
  20.     @Override
  21.     protected void initChannel(SocketChannel ch) throws Exception {
  22.         /*Netty提供的日志打印Handler，可以展示发送接收出去的字节*/
  23.         //ch.pipeline().addLast(new LoggingHandler(LogLevel.INFO));
  24.         /*剥离接收到的消息的长度字段，拿到实际的消息报文的字节数组*/
  25.         ch.pipeline().addLast("frameDecoder",
  26.                 new LengthFieldBasedFrameDecoder(65535,
  27.                         0,2,0,
  28.                         2));
  29.         /*给发送出去的消息增加长度字段*/
  30.         ch.pipeline().addLast("frameEncoder",
  31.                 new LengthFieldPrepender(2));
  32.         /*反序列化，将字节数组转换为消息实体*/
  33.         ch.pipeline().addLast(new KryoDecoder());
  34.         /*序列化，将消息实体转换为字节数组准备进行网络传输*/
  35.         ch.pipeline().addLast("MessageEncoder",
  36.                 new KryoEncoder());
  37.         /*超时检测*/
  38.         ch.pipeline().addLast("readTimeoutHandler",
  39.                 new ReadTimeoutHandler(50));
  40.         /*登录应答*/
  41.         ch.pipeline().addLast(new LoginAuthRespHandler());
  42.  
  43.         /*心跳应答*/
  44.         ch.pipeline().addLast("HeartBeatHandler",
  45.                 new HeartBeatRespHandler());
  46.  
  47.         /*服务端业务处理*/
  48.         ch.pipeline().addLast("ServerBusiHandler",
  49.                 serverBusiHandler);
  50.     }
  51. }

2.3、netty处理业务的Handler入口 ServerBusiHandler 类：

  1. import io.netty.channel.ChannelHandler;
  2. import io.netty.channel.ChannelHandlerContext;
  3. import io.netty.channel.SimpleChannelInboundHandler;
  4. import org.apache.commons.logging.Log;
  5. import org.apache.commons.logging.LogFactory;
  6. import org.springframework.beans.factory.annotation.Autowired;
  7. import org.springframework.stereotype.Service;
  8.  
  9. import java.lang.reflect.Method;
  10. import java.util.HashMap;
  11. import java.util.Map;
  12.  
  13. /**
  14.  * @author Mark老师   享学课堂 https://enjoy.ke.qq.com
  15.  * 类说明：业务处理类
  16.  * channelRead0方法中有了实际的业务处理，负责具体的业务方法的调用
  17.  *
  18.  */
  19. @Service
  20. @ChannelHandler.Sharable//共享的单例
  21. public class ServerBusiHandler
  22.         extends SimpleChannelInboundHandler<MyMessage> {
  23.     private static final Log LOG
  24.             = LogFactory.getLog(ServerBusiHandler.class);
  25.  
  26.     @Autowired
  27.     private RegisterService registerService;
  28.  
  29.     @Override
  30.     protected void channelRead0(ChannelHandlerContext ctx, MyMessage msg)
  31.             throws Exception {
  32.         LOG.info(msg);
  33.         MyMessage message = new MyMessage();
  34.         MyHeader myHeader = new MyHeader();
  35.         myHeader.setSessionID(msg.getMyHeader().getSessionID());
  36.         myHeader.setType(MessageType.SERVICE_RESP.value());
  37.         message.setMyHeader(myHeader);
  38.         Map<String,Object> content = (HashMap<String,Object>)msg.getBody();
  39.         /*方法所在类名接口名*/
  40.         String serviceName = (String) content.get("siName");
  41.         /*方法的名字*/
  42.         String methodName = (String) content.get("methodName");
  43.         /*方法的入参类型*/
  44.         Class<?>[] paramTypes = (Class<?>[]) content.get("paraTypes");
  45.         /*方法的入参的值*/
  46.         Object[] args = (Object[]) content.get("args");
  47.         /*从容器中拿到服务的Class对象*/
  48.         Class serviceClass = registerService.getLocalService(serviceName);
  49.         if(serviceClass == null){
  50.             throw new ClassNotFoundException(serviceName+ " not found");
  51.         }
  52.  
  53.         /*通过反射，执行实际的服务*/
  54.         Method method = serviceClass.getMethod(methodName, paramTypes);
  55.         boolean result  = (boolean)method.invoke(serviceClass.newInstance(),args);
  56.         message.setBody(result);
  57.         ctx.writeAndFlush(message);
  58.     }
  59.  
  60.     @Override
  61.     public void channelInactive(ChannelHandlerContext ctx)
  62.             throws Exception {
  63.         LOG.info(ctx.channel().remoteAddress()+" 主动断开了连接!");
  64.     }
  65.  
  66. }

3、客户端核心代码实现(boot启动类省略)：

3.1、netty客户端启动入口

  1. import io.netty.bootstrap.Bootstrap;
  2. import io.netty.channel.Channel;
  3. import io.netty.channel.ChannelFuture;
  4. import io.netty.channel.ChannelOption;
  5. import io.netty.channel.EventLoopGroup;
  6. import io.netty.channel.nio.NioEventLoopGroup;
  7. import io.netty.channel.socket.nio.NioSocketChannel;
  8. import org.apache.commons.logging.Log;
  9. import org.apache.commons.logging.LogFactory;
  10. import org.springframework.beans.factory.annotation.Autowired;
  11. import org.springframework.stereotype.Service;
  12.  
  13. import javax.annotation.PostConstruct;
  14. import javax.annotation.PreDestroy;
  15. import java.lang.reflect.InvocationHandler;
  16. import java.lang.reflect.Method;
  17. import java.lang.reflect.Proxy;
  18. import java.net.InetSocketAddress;
  19. import java.util.HashMap;
  20. import java.util.Map;
  21. import java.util.concurrent.Executors;
  22. import java.util.concurrent.ScheduledExecutorService;
  23. import java.util.concurrent.TimeUnit;
  24.  
  25. /**
  26.  *类说明：rpc框架的客户端代理部分,交给Spring 托管
  27.  * 1、动态代理的实现中，不再连接服务器，而是直接发送请求
  28.  * 2、客户端网络部分的主体，包括Netty组件的初始化，连接服务器等
  29.  */
  30. @Service
  31. public class RpcClientFrame implements Runnable{
  32.  
  33.     private static final Log LOG = LogFactory.getLog(RpcClientFrame.class);
  34.  
  35.     private ScheduledExecutorService executor = Executors
  36.             .newScheduledThreadPool(1);
  37.     private Channel channel;
  38.     private EventLoopGroup group = new NioEventLoopGroup();
  39.  
  40.     /*是否用户主动关闭连接的标志值*/
  41.     private volatile boolean userClose = false;
  42.     /*连接是否成功关闭的标志值*/
  43.     private volatile boolean connected = false;
  44.  
  45.     @Autowired
  46.     private ClientInit clientInit;
  47.     @Autowired
  48.     private ClientBusiHandler clientBusiHandler;
  49.  
  50.     /*远程服务的代理对象，参数为客户端要调用的的服务*/
  51.     public <T> T getRemoteProxyObject(final Class<?> serviceInterface) throws Exception {
  52.  
  53.         /*拿到一个代理对象，由这个代理对象通过网络进行实际的服务调用*/
  54.         return (T)Proxy.newProxyInstance(serviceInterface.getClassLoader(),
  55.                 new Class<?>[]{serviceInterface},
  56.                 new DynProxy(serviceInterface,clientBusiHandler));
  57.     }
  58.  
  59.     /*动态代理，实现对远程服务的访问*/
  60.     private static class DynProxy implements InvocationHandler{
  61.         private Class<?> serviceInterface;
  62.         private ClientBusiHandler clientBusiHandler;
  63.  
  64.         public DynProxy(Class<?> serviceInterface, ClientBusiHandler clientBusiHandler) {
  65.             this.serviceInterface = serviceInterface;
  66.             this.clientBusiHandler = clientBusiHandler;
  67.         }
  68.  
  69.         @Override
  70.         public Object invoke(Object proxy, Method method, Object[] args)
  71.                 throws Throwable {
  72.             Map<String,Object> content = new HashMap<>();
  73.             content.put("siName",serviceInterface.getName());
  74.             content.put("methodName",method.getName());
  75.             content.put("paraTypes",method.getParameterTypes());
  76.             content.put("args",args);
  77.             return clientBusiHandler.send(content);
  78.         }
  79.     }
  80.  
  81.     public boolean isConnected() {
  82.         return connected;
  83.     }
  84.  
  85.     /*连接服务器*/
  86.     public void connect(int port, String host) throws Exception {
  87.  
  88.         try {
  89.             Bootstrap b = new Bootstrap();
  90.             b.group(group).channel(NioSocketChannel.class)
  91.                     .option(ChannelOption.TCP_NODELAY, true)
  92.                     .handler(clientInit);
  93.             // 发起异步连接操作
  94.             ChannelFuture future = b.connect(
  95.                     new InetSocketAddress(host, port)).sync();
  96.             channel = future.sync().channel();
  97.             /*连接成功后通知等待线程，连接已经建立*/
  98.             synchronized (this){
  99.                 this.connected = true;
  100.                 this.notifyAll();
  101.             }
  102.             future.channel().closeFuture().sync();
  103.         } finally {
  104.             if(!userClose){/*非用户主动关闭，说明发生了网络问题，需要进行重连操作*/
  105.                 System.out.println("发现异常，可能发生了服务器异常或网络问题，" +
  106.                         "准备进行重连.....");
  107.                 //再次发起重连操作
  108.                 executor.execute(new Runnable() {
  109.                     @Override
  110.                     public void run() {
  111.                         try {
  112.                             TimeUnit.SECONDS.sleep(1);
  113.                             try {
  114.                                 // 发起重连操作
  115.                                 connect(NettyConstant.REMOTE_PORT,
  116.                                         NettyConstant.REMOTE_IP);
  117.                             } catch (Exception e) {
  118.                                 e.printStackTrace();
  119.                             }
  120.                         } catch (InterruptedException e) {
  121.                             e.printStackTrace();
  122.                         }
  123.                     }
  124.                 });
  125.             }else{/*用户主动关闭，释放资源*/
  126.                 channel = null;
  127.                 group.shutdownGracefully().sync();
  128.                 connected = false;
  129. //                synchronized (this){
  130. //                    this.connected = false;
  131. //                    this.notifyAll();
  132. //                }
  133.             }
  134.         }
  135.     }
  136.  
  137.     @Override
  138.     public void run() {
  139.         try {
  140.             connect(NettyConstant.REMOTE_PORT, NettyConstant.REMOTE_IP);
  141.         } catch (Exception e) {
  142.             e.printStackTrace();
  143.         }
  144.     }
  145.  
  146.     public void close() {
  147.         userClose = true;
  148.         channel.close();
  149.     }
  150.  
  151.     @PostConstruct
  152.     public void startNet() throws InterruptedException {
  153.         new Thread(this).start();
  154.         while(!this.isConnected()){
  155.             synchronized (this){
  156.                 this.wait();
  157.             }
  158.         }
  159.         LOG.info("网络通信已准备好，可以进行业务操作了........");
  160.     }
  161.  
  162.     @PreDestroy
  163.     public void stopNet(){
  164.         close();
  165.     }
  166.  
  167. }

3.2、netty客户端自定义配置入口  ClientInit

  1. import io.netty.channel.ChannelInitializer;
  2. import io.netty.channel.socket.SocketChannel;
  3. import io.netty.handler.codec.LengthFieldBasedFrameDecoder;
  4. import io.netty.handler.codec.LengthFieldPrepender;
  5. import io.netty.handler.timeout.ReadTimeoutHandler;
  6. import org.springframework.beans.factory.annotation.Autowired;
  7. import org.springframework.stereotype.Service;
  8.  
  9. /**
  10.  * @author Mark老师   享学课堂 https://enjoy.ke.qq.com
  11.  * 类说明：客户端Handler的初始化
  12.  * 交给Spring 托管,clientBusiHandler用注入方式实例化后加入Netty的pipeline
  13.  */
  14. @Service
  15. public class ClientInit extends ChannelInitializer<SocketChannel> {
  16.  
  17.     @Autowired
  18.     private ClientBusiHandler clientBusiHandler;
  19.  
  20.     @Override
  21.     protected void initChannel(SocketChannel ch) throws Exception {
  22.         /*剥离接收到的消息的长度字段，拿到实际的消息报文的字节数组*/
  23.         ch.pipeline().addLast("frameDecoder",
  24.                 new LengthFieldBasedFrameDecoder(65535,
  25.                         0,2,0,
  26.                         2));
  27.  
  28.         /*给发送出去的消息增加长度字段*/
  29.         ch.pipeline().addLast("frameEncoder",
  30.                 new LengthFieldPrepender(2));
  31.  
  32.         /*反序列化，将字节数组转换为消息实体*/
  33.         ch.pipeline().addLast(new KryoDecoder());
  34.         /*序列化，将消息实体转换为字节数组准备进行网络传输*/
  35.         ch.pipeline().addLast("MessageEncoder",
  36.                 new KryoEncoder());
  37.  
  38.         /*超时检测*/
  39.         ch.pipeline().addLast("readTimeoutHandler",
  40.                 new ReadTimeoutHandler(10));
  41.  
  42.         /*发出登录请求*/
  43.         ch.pipeline().addLast("LoginAuthHandler",
  44.                 new LoginAuthReqHandler());
  45.  
  46.         /*发出心跳请求*/
  47.         ch.pipeline().addLast("HeartBeatHandler",
  48.                 new HeartBeatReqHandler());
  49.  
  50.         /*业务处理*/
  51.         ch.pipeline().addLast("ClientBusiHandler",
  52.                 clientBusiHandler);
  53.     }
  54. }

3.3、netty通信中序列化的实现

序列化的Handler

  1. import io.netty.buffer.ByteBuf;
  2. import io.netty.channel.ChannelHandlerContext;
  3. import io.netty.handler.codec.MessageToByteEncoder;
  4.  
  5. /**
  6.  * 类说明：序列化的Handler
  7.  */
  8. public class KryoEncoder  extends MessageToByteEncoder<MyMessage> {
  9.  
  10.     @Override
  11.     protected void encode(ChannelHandlerContext ctx, MyMessage message,
  12.                           ByteBuf out) throws Exception {
  13.         KryoSerializer.serialize(message, out);
  14.         ctx.flush();
  15.     }
  16. }

反序列化的Handler

  1. import io.netty.buffer.ByteBuf;
  2. import io.netty.channel.ChannelHandlerContext;
  3. import io.netty.handler.codec.ByteToMessageDecoder;
  4.  
  5. import java.util.List;
  6.  
  7. /**
  8.  * 类说明：反序列化的Handler
  9.  */
  10. public class KryoDecoder extends ByteToMessageDecoder {
  11.  
  12.     @Override
  13.     protected void decode(ChannelHandlerContext ctx, ByteBuf in,
  14.                           List<Object> out) throws Exception {
  15.         Object obj = KryoSerializer.deserialize(in);
  16.         out.add(obj);
  17.     }
  18. }

信息类的定义：

  1. /**
  2.  * 类说明：消息实体类
  3.  */
  4. public final class MyMessage {
  5.  
  6.     private MyHeader myHeader;
  7.  
  8.     private Object body;
  9.  
  10.     public final MyHeader getMyHeader() {
  11.     	return myHeader;
  12.     }
  13.  
  14.     public final void setMyHeader(MyHeader myHeader) {
  15.     	this.myHeader = myHeader;
  16.     }
  17.  
  18.     public final Object getBody() {
  19.     	return body;
  20.     }
  21.  
  22.     public final void setBody(Object body) {
  23.     	this.body = body;
  24.     }
  25.  
  26.     @Override
  27.     public String toString() {
  28.     	return "MyMessage [myHeader=" + myHeader + "][body="+body+"]";
  29.     }
  30. }

4、启动Netty服务端：

5、启动客户端

接下来就可以通信了，比如，客户端发送消息：
 

到此，springboot集成netty的过程基本结束，后期会详细分析其前后端交互过程，敬请期待。 

声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/w/笔触狂放9/article/detail/185420
推荐阅读
相关标签

Copyright © 2003-2013 www.wpsshop.cn 版权所有，并保留所有权利。

  

闽ICP备14008679号