devbox
从前慢现在也慢
这个屌丝很懒，什么也没留下！
关注作者
热门标签
热门文章
当前位置:   article > 正文

springboot集成netty开发服务端和客户端_spring netty client

作者：从前慢现在也慢 | 2024-03-03 19:32:38

赞

踩

spring netty client

maven里面引入netty依赖

  1. <dependency>
  2.    <groupId>io.netty</groupId>
  3.    <artifactId>netty-all</artifactId>
  4.    <version>4.1.51.Final</version>
  5. </dependency>

创建NettyServer类

  1. package com.NettyServer.service;
  2.  
  3. import com.NettyServer.common.ServerChannelInitializer;
  4. import io.netty.bootstrap.ServerBootstrap;
  5. import io.netty.channel.ChannelFuture;
  6. import io.netty.channel.ChannelOption;
  7. import io.netty.channel.EventLoopGroup;
  8. import io.netty.channel.nio.NioEventLoopGroup;
  9. import io.netty.channel.socket.nio.NioServerSocketChannel;
  10. import org.springframework.beans.factory.annotation.Value;
  11. import lombok.extern.slf4j.Slf4j;
  12. import org.springframework.stereotype.Component;
  13.  
  14. import javax.annotation.PostConstruct;
  15. import javax.annotation.PreDestroy;
  16. import java.net.InetSocketAddress;
  17.  
  18. @Component
  19. @Slf4j
  20. public class NettyServer {
  21.     /**
  22.      * boss 线程组用于TCP处理连接工作
  23.      */
  24.     private final EventLoopGroup boss = new NioEventLoopGroup();
  25.     /**
  26.      * work 线程组用于IO数据处理
  27.      */
  28.     private final EventLoopGroup work = new NioEventLoopGroup();
  29.  
  30.     @Value("${netty.port}")
  31.     private Integer port;
  32.  
  33.  
  34.     /**
  35.      * 启动Netty Server
  36.      */
  37.     @PostConstruct
  38.     public void start() throws InterruptedException {
  39.         ServerBootstrap bootstrap = new ServerBootstrap();
  40.         bootstrap.group(boss, work)
  41.                 // 指定Channel
  42.                 .channel(NioServerSocketChannel.class)
  43.                 //使用指定的端口设置套接字地址
  44.                 .localAddress(new InetSocketAddress(port))
  45.                 //服务端可连接队列数,对应TCP/IP协议listen函数中backlog参数
  46.                 .option(ChannelOption.SO_BACKLOG, 1024)
  47.                 //设置TCP长连接,一般如果两个小时内没有数据的通信时,TCP会自动发送一个活动探测数据报文
  48.                 .childOption(ChannelOption.SO_KEEPALIVE, true)
  49.                 //将小的数据包包装成更大的帧进行传送，提高网络的负载
  50.                 .childOption(ChannelOption.TCP_NODELAY, true)
  51.                 .childHandler(new ServerChannelInitializer());
  52.         ChannelFuture future = bootstrap.bind().sync();
  53.         if (future.isSuccess()) {
  54.             log.info("已启动 Netty Server");
  55.         }
  56.     }
  57.  
  58.     @PreDestroy
  59.     public void destory() throws InterruptedException {
  60.         boss.shutdownGracefully().sync();
  61.         work.shutdownGracefully().sync();
  62.         log.info("关闭Netty");
  63.     }

创建ServerChannelInitializer类

  1. package com.NettyServer.common;
  2.  
  3. import com.NettyServer.service.NettyServerHandler;
  4. import io.netty.channel.ChannelInitializer;
  5. import io.netty.channel.socket.SocketChannel;
  6. import io.netty.handler.codec.serialization.ClassResolvers;
  7. import io.netty.handler.codec.serialization.ObjectDecoder;
  8. import io.netty.handler.codec.serialization.ObjectEncoder;
  9.  
  10. public class ServerChannelInitializer extends ChannelInitializer<SocketChannel> {
  11.  
  12.     @Override
  13.     protected void initChannel(SocketChannel socketChannel) {
  14.         //添加编解码
  15.         socketChannel.pipeline().addLast(new ObjectDecoder(10 * 1024 * 1024, ClassResolvers.weakCachingConcurrentResolver(this.getClass().getClassLoader())));
  16.         socketChannel.pipeline().addLast(new ObjectEncoder());
  17.         socketChannel.pipeline().addLast(new NettyServerHandler());
  18.     }
  19.  
  20.  
  21.  
  22. }

创建NettyServerHandler类

  1. package com.NettyServer.service;
  2.  
  3. import io.netty.channel.ChannelHandlerContext;
  4. import io.netty.channel.ChannelId;
  5. import io.netty.channel.ChannelInboundHandlerAdapter;
  6. import io.netty.handler.timeout.IdleState;
  7. import io.netty.handler.timeout.IdleStateEvent;
  8. import lombok.extern.slf4j.Slf4j;
  9. import org.springframework.beans.factory.annotation.Autowired;
  10. import org.springframework.stereotype.Component;
  11.  
  12. import javax.annotation.PostConstruct;
  13. import java.net.InetSocketAddress;
  14. import java.util.concurrent.ConcurrentHashMap;
  15.  
  16. @Slf4j
  17. @Component
  18. public class NettyServerHandler extends ChannelInboundHandlerAdapter {
  19.  
  20.     private static NettyServerHandler handler;
  21.  
  22.     /**
  23.      * 管理一个全局map，保存连接进服务端的通道数量
  24.      */
  25.     private static final ConcurrentHashMap<ChannelId, ChannelHandlerContext> CHANNEL_MAP = new ConcurrentHashMap<>();
  26.  
  27.     //@Autowired
  28.     //private IBaseAttachmentService attachmentService;
  29.  
  30.     @PostConstruct
  31.     public void init() {
  32.         handler = this;
  33.     }
  34.  
  35.     /**
  36.      * @param ctx
  37.      * @author xxx on 2021/4/28 16:10
  38.      * @DESCRIPTION: 有客户端连接服务器会触发此函数
  39.      * @return: void
  40.      */
  41.     @Override
  42.     public void channelActive(ChannelHandlerContext ctx) {
  43.  
  44.         InetSocketAddress insocket = (InetSocketAddress) ctx.channel().remoteAddress();
  45.  
  46.         String clientIp = insocket.getAddress().getHostAddress();
  47.         int clientPort = insocket.getPort();
  48.  
  49.         //获取连接通道唯一标识
  50.         ChannelId channelId = ctx.channel().id();
  51.  
  52.         //如果map中不包含此连接，就保存连接
  53.         if (CHANNEL_MAP.containsKey(channelId)) {
  54.             log.info("客户端【" + channelId + "】是连接状态，连接通道数量: " + CHANNEL_MAP.size());
  55.         } else {
  56.             //保存连接
  57.             CHANNEL_MAP.put(channelId, ctx);
  58.  
  59.             log.info("客户端【" + channelId + "】连接netty服务器[IP:" + clientIp + "--->PORT:" + clientPort + "]");
  60.             log.info("连接通道数量: " + CHANNEL_MAP.size());
  61.         }
  62.     }
  63.  
  64.  
  65.     /**
  66.      * @param ctx
  67.      * @author xxx on 2021/4/28 16:10
  68.      * @DESCRIPTION: 有客户端终止连接服务器会触发此函数
  69.      * @return: void
  70.      */
  71.     @Override
  72.     public void channelInactive(ChannelHandlerContext ctx) {
  73.  
  74.         InetSocketAddress insocket = (InetSocketAddress) ctx.channel().remoteAddress();
  75.  
  76.         String clientIp = insocket.getAddress().getHostAddress();
  77.  
  78.         ChannelId channelId = ctx.channel().id();
  79.  
  80.         //包含此客户端才去删除
  81.         if (CHANNEL_MAP.containsKey(channelId)) {
  82.             //删除连接
  83.             CHANNEL_MAP.remove(channelId);
  84.             log.info("客户端【" + channelId + "】退出netty服务器[IP:" + clientIp + "--->PORT:" + insocket.getPort() + "]");
  85.             log.info("连接通道数量: " + CHANNEL_MAP.size());
  86.         }
  87.     }
  88.  
  89.  
  90.     /**
  91.      * @param ctx
  92.      * @author xxx on 2021/4/28 16:10
  93.      * @DESCRIPTION: 有客户端发消息会触发此函数
  94.      * @return: void
  95.      */
  96.     @Override
  97.     public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
  98.         log.info("【" + ctx.channel().id() + "】" + " :" + msg.toString());
  99.         //可以在这个地方写业务处理逻辑入库或者什么的...................................
  100.         // 结果写到客户端
  101.         /**
  102.          *  下面可以解析数据，保存数据，生成返回报文，将需要返回报文写入write函数
  103.          *
  104.          */
  105.  
  106.         //响应客户端
  107.         this.channelWrite(ctx.channel().id(), msg);
  108.  
  109.     }
  110.  
  111.     /**
  112.      * @param msg       需要发送的消息内容
  113.      * @param channelId 连接通道唯一id
  114.      * @author xxx 2021/05/14 16:10
  115.      * @DESCRIPTION: 服务端给客户端发送消息
  116.      * @return: void
  117.      */
  118.  
  119.     public void channelWrite(ChannelId channelId, Object msg) throws Exception {
  120.  
  121.         ChannelHandlerContext ctx = CHANNEL_MAP.get(channelId);
  122.  
  123.         if (ctx == null) {
  124.             log.info("通道【" + channelId + "】不存在");
  125.             return;
  126.         }
  127.  
  128.         if (msg == null || msg == "") {
  129.             log.info("服务端响应空的消息");
  130.             return;
  131.         }
  132.  
  133.         //将客户端的信息直接返回写入ctx
  134.         ctx.write(msg);
  135.         //刷新缓存区
  136.         ctx.flush();
  137.     }
  138.  
  139.     @Override
  140.     public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
  141.  
  142.         String socketString = ctx.channel().remoteAddress().toString();
  143.  
  144.         if (evt instanceof IdleStateEvent) {
  145.             IdleStateEvent event = (IdleStateEvent) evt;
  146.             if (event.state() == IdleState.READER_IDLE) {
  147.                 log.info("Client: " + socketString + " READER_IDLE 读超时");
  148.                 ctx.disconnect();
  149.             } else if (event.state() == IdleState.WRITER_IDLE) {
  150.                 log.info("Client: " + socketString + " WRITER_IDLE 写超时");
  151.                 ctx.disconnect();
  152.             } else if (event.state() == IdleState.ALL_IDLE) {
  153.                 log.info("Client: " + socketString + " ALL_IDLE 总超时");
  154.                 ctx.disconnect();
  155.             }
  156.         }
  157.     }
  158.  
  159.     /**
  160.      * @param ctx
  161.      * @author xxx on 2021/4/28 16:10
  162.      * @DESCRIPTION: 发生异常会触发此函数
  163.      * @return: void
  164.      */
  165.     @Override
  166.     public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
  167.  
  168.         System.out.println();
  169.         ctx.close();
  170.         log.info(ctx.channel().id() + " 发生了错误,此连接被关闭" + "此时连通数量: " + CHANNEL_MAP.size());
  171.         //cause.printStackTrace();
  172.     }
  173.  
  174.  
  175. }

下面贴出来客户端代码

创建NettyClient类

  1. package com.NettyClient.service;
  2.  
  3.  
  4. import com.NettyClient.common.NettyClientInitializer;
  5. import io.netty.bootstrap.Bootstrap;
  6. import io.netty.channel.ChannelFuture;
  7. import io.netty.channel.ChannelFutureListener;
  8. import io.netty.channel.ChannelOption;
  9. import io.netty.channel.EventLoopGroup;
  10. import io.netty.channel.nio.NioEventLoopGroup;
  11. import io.netty.channel.socket.nio.NioSocketChannel;
  12. import lombok.extern.slf4j.Slf4j;
  13. import org.springframework.stereotype.Component;
  14.  
  15. import javax.annotation.PostConstruct;
  16.  
  17. @Slf4j
  18. @Component
  19. public class NettyClient {
  20.  
  21.     static final String HOST = "127.0.0.1";
  22.     static final int PORT = 8899;
  23.     private EventLoopGroup group;
  24.     private Bootstrap b;
  25.     private ChannelFuture cf;
  26.     private NettyClientInitializer nettyClientInitializer;
  27.  
  28.  
  29.     public NettyClient() {
  30.         nettyClientInitializer = new NettyClientInitializer();
  31.         group = new NioEventLoopGroup();
  32.         b = new Bootstrap();
  33.         b.group(group)
  34.                 .channel(NioSocketChannel.class)
  35.                 .handler(nettyClientInitializer);
  36.     }
  37.  
  38.     public void connect() {
  39.         try {
  40.             this.cf = b.connect(HOST, PORT).sync();
  41.         } catch (InterruptedException e) {
  42.             log.error("客户端连接服务端异常：" + e);
  43.         }
  44.     }
  45.  
  46.     public ChannelFuture getChannelFuture() {
  47.         if (this.cf == null) {
  48.             this.connect();
  49.         }
  50.         if (!this.cf.channel().isActive()) {
  51.             this.connect();
  52.         }
  53.         return this.cf;
  54.     }
  55.  
  56.     public void close() {
  57.         try {
  58.             this.cf.channel().closeFuture().sync();
  59.             this.group.shutdownGracefully();
  60.         } catch (InterruptedException e) {
  61.             e.printStackTrace();
  62.         }
  63.     }
  64.  
  65.     public void setMessage(String msg) throws InterruptedException {
  66.         ChannelFuture cf = this.getChannelFuture();
  67.         cf.channel().writeAndFlush(msg);
  68.     }
  69.  
  70.  
  71.     public static void main(String[] args) {
  72.         try {
  73.  
  74.         } catch (Exception e) {
  75.             log.error("异常：" + e);
  76.         }
  77.     }
  78.  
  79.  
  80. }

创建NettyClientHandler类

  1. package com.NettyClient.service;
  2.  
  3.  
  4. import io.netty.channel.ChannelHandlerContext;
  5. import io.netty.channel.ChannelInboundHandlerAdapter;
  6. import lombok.extern.slf4j.Slf4j;
  7. import org.springframework.stereotype.Component;
  8.  
  9.  
  10. @Slf4j
  11. public class NettyClientHandler extends ChannelInboundHandlerAdapter {
  12.  
  13.     private String result;
  14.  
  15.     @Override
  16.  
  17.     public void channelActive(ChannelHandlerContext ctx) {
  18.     }
  19.  
  20.     @Override
  21.     public void channelRead(ChannelHandlerContext ctx, Object msg) {
  22.         //if (!ObjectUtils.isEmpty(msg)) {
  23.         result = (String) msg;
  24.         //}
  25.     }
  26.  
  27.     @Override
  28.     public void channelReadComplete(ChannelHandlerContext ctx) {
  29.         ctx.flush();
  30.     }
  31.  
  32.     @Override
  33.     public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
  34.         cause.printStackTrace();
  35.         ctx.close();
  36.     }
  37.  
  38. }

创建NettyClientInitializer类

  1. package com.NettyClient.common;
  2.  
  3. import com.NettyClient.service.NettyClientHandler;
  4. import io.netty.channel.ChannelInitializer;
  5. import io.netty.channel.socket.SocketChannel;
  6. import io.netty.handler.codec.serialization.ClassResolvers;
  7. import io.netty.handler.codec.serialization.ObjectDecoder;
  8. import io.netty.handler.codec.serialization.ObjectEncoder;
  9. import io.netty.handler.timeout.ReadTimeoutHandler;
  10. import org.springframework.beans.factory.annotation.Autowired;
  11.  
  12.  
  13. public class NettyClientInitializer extends ChannelInitializer<SocketChannel> {
  14.  
  15.  
  16.     @Override
  17.     protected void initChannel(SocketChannel socketChannel) throws Exception {
  18.         socketChannel.pipeline().addLast(new ObjectDecoder(1024, ClassResolvers.cacheDisabled(this.getClass().getClassLoader())));
  19.         socketChannel.pipeline().addLast(new ObjectEncoder());
  20.         socketChannel.pipeline().addLast(new NettyClientHandler());
  21.     }
  22.  
  23. }

客户端直接可以在业务代码里面调用发送数据到服务端就行了、我简单写了一个定时器测试发送的例子

  1. package com.NettyClient;
  2.  
  3.  
  4. import com.NettyClient.service.NettyClient;
  5. import lombok.RequiredArgsConstructor;
  6. import lombok.extern.slf4j.Slf4j;
  7. import org.springframework.beans.factory.annotation.Autowired;
  8. import org.springframework.scheduling.annotation.Async;
  9. import org.springframework.scheduling.annotation.EnableAsync;
  10. import org.springframework.scheduling.annotation.Scheduled;
  11. import org.springframework.stereotype.Component;
  12.  
  13. import java.text.SimpleDateFormat;
  14. import java.util.Date;
  15.  
  16. @Component
  17. @Slf4j
  18. @EnableAsync
  19. public class TaskManager {
  20.  
  21.     @Autowired
  22.     private NettyClient nettyClient;
  23.  
  24.     @Async
  25.     @Scheduled(cron = "0 */1 * * * ?")
  26.     public void test() throws InterruptedException {
  27.         try {
  28.             SimpleDateFormat df = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
  29.             nettyClient.setMessage("客户端发送此数据" + df.format(new Date()));
  30.         } catch (Exception e) {
  31.             log.error("异常：" + e);
  32.         }
  33.  
  34.     }
  35.  
  36.  
  37.     @Async
  38.     @Scheduled(cron = "0 */1 * * * ?")
  39.     public void test2() throws InterruptedException {
  40.         try {
  41.             SimpleDateFormat df = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
  42.             nettyClient.setMessage("客户端2发送此数据" + df.format(new Date()));
  43.         } catch (Exception e) {
  44.             log.error("异常：" + e);
  45.         }
  46.  
  47.     }
  48.  
  49.  
  50.  
  51. }

 

 

 

 

 

 

 

 

 

 

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

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

  

闽ICP备14008679号