Spring Boot+Netty

因工作中需要给第三方屏幕厂家下发广告，音频，图片等内容，对方提供TCP接口于是我使用Netty长链接进行数据传输

1.添加依赖

 <!--  netty依赖--><dependency><groupId>io.netty</groupId><artifactId>netty-all</artifactId></dependency>   

2.创建Netty服务

@Slf4j
@Component
public class NettyServer {public void start(InetSocketAddress address) {//配置服务端的NIO线程组EventLoopGroup bossGroup = new NioEventLoopGroup();EventLoopGroup workerGroup = new NioEventLoopGroup();try {// 绑定线程池,编码解码//服务端接受连接的队列长度，如果队列已满，客户端连接将被拒绝ServerBootstrap bootstrap = new ServerBootstrap().group(bossGroup, workerGroup)// 指定Channel.channel(NioServerSocketChannel.class)//使用指定的端口设置套接字地址.localAddress(address)//使用自定义处理类.childHandler(new NettyServerChannelInitializer())//服务端可连接队列数,对应TCP/IP协议listen函数中backlog参数.option(ChannelOption.SO_BACKLOG, 128)//保持长连接，2小时无数据激活心跳机制.childOption(ChannelOption.SO_KEEPALIVE, true)//将小的数据包包装成更大的帧进行传送，提高网络的负载.childOption(ChannelOption.TCP_NODELAY, true);// 绑定端口，开始接收进来的连接ChannelFuture future = bootstrap.bind(address).sync();if (future.isSuccess()) {log.info("netty服务器开始监听端口：{}",address.getPort());}//关闭channel和块，直到它被关闭future.channel().closeFuture().sync();} catch (Exception e) {e.printStackTrace();bossGroup.shutdownGracefully();workerGroup.shutdownGracefully();}}
}

3.创建Socket配置类（也可以直接在步骤2中写死）

1.在配置文件中

socket:# 监听端口 8090port: 8090#ip地址host: 0.0.0.0
#  host: 192.168.31.2@Setter
@Getter
@ToString
@Component
@Configuration
@PropertySource("classpath:application.yml")
@ConfigurationProperties(prefix = "socket")
public class SocketProperties {private Integer port;private String host;}

4.在springboot 启动类中启用Netty服务

@SpringBootApplication
public class Application implements CommandLineRunner {public static void main(String[] args) {SpringApplication application = new SpringApplication(Application.class);application.setApplicationStartup(new BufferingApplicationStartup(2048));application.run(args);}@Resourceprivate NettyServer nettyServer;@Resourceprivate SocketProperties socketProperties;@Overridepublic void run(String... args) {InetSocketAddress address = new InetSocketAddress(socketProperties.getHost(),socketProperties.getPort());nettyServer.start(address);}}

5.创建字符解析器，解析收到的消息

/*** 功能描述: 服务端初始化，客户端与服务器端连接一旦创建，这个类中方法就会被回调，设置出站编码器和入站解码器**/
public class NettyServerChannelInitializer extends ChannelInitializer<SocketChannel> {@Overrideprotected void initChannel(SocketChannel socketChannel) throws Exception {ChannelPipeline pipeline = socketChannel.pipeline();//接收消息格式,使用自定义解析数据格式
//        pipeline.addLast("decoder",new MyDecoder());//发送消息格式，使用自定义解析数据格式
//        pipeline.addLast("encoder",new MyEncoder());pipeline.addLast(new StringEncoder(CharsetUtil.UTF_8));//针对客户端，如果在1分钟时没有想服务端发送写心跳(ALL)，则主动断开//如果是读空闲或者写空闲，不处理,这里根据自己业务考虑使用pipeline.addLast(new IdleStateHandler(0,0,90, TimeUnit.SECONDS));//自定义的空闲检测pipeline.addLast(new NettyServerHandler());}
}

6.创建Handler 类处理消息

/*** 功能描述: netty服务端处理类*/@Slf4j
@Component
public class NettyServerHandler extends ChannelInboundHandlerAdapter {/*** 功能描述: 有客户端连接服务器会触发此函数** @param ctx 通道* @return void*/@Overridepublic void channelActive(ChannelHandlerContext ctx) {InetSocketAddress insocket = (InetSocketAddress) ctx.channel().remoteAddress();String clientIp = insocket.getAddress().getHostAddress();int clientPort = insocket.getPort();//获取连接通道唯一标识ChannelId channelId = ctx.channel().id();//如果map中不包含此连接，就保存连接if (ChannelMap.getChannelMap().containsKey(channelId)) {log.info("客户端:{},是连接状态，连接通道数量:{} ", channelId, ChannelMap.getChannelMap().size());} else {//保存连接ChannelMap.addChannel(channelId, ctx.channel());log.info("客户端:{},连接netty服务器[IP:{}-->PORT:{}]", channelId, clientIp, clientPort);log.info("连接通道数量: {}", ChannelMap.getChannelMap().size());}}/*** 功能描述: 有客户端终止连接服务器会触发此函数* @param ctx 通道处理程序上下文* @return void*/@Overridepublic void channelInactive(ChannelHandlerContext ctx) {InetSocketAddress inSocket = (InetSocketAddress) ctx.channel().remoteAddress();String clientIp = inSocket.getAddress().getHostAddress();ChannelId channelId = ctx.channel().id();//包含此客户端才去删除if (ChannelMap.getChannelMap().containsKey(channelId)) {//删除连接ChannelMap.getChannelMap().remove(channelId);log.info("客户端:{},断开netty服务器[IP:{}-->PORT:{}]", channelId, clientIp, inSocket.getPort());log.info("连接通道数量: " + ChannelMap.getChannelMap().size());}}@Transactional@Overridepublic void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {ByteBuf buf = (ByteBuf) msg;ByteBuf rebuf = Unpooled.buffer();RedisUtils.setChannelId(ctx.channel().id().toString(), ctx.channel().id());// 读取帧头标识byte frameHeader = buf.readByte();if (frameHeader != 0x7E) {byte[] data = ByteBufUtil.getBytes(buf);String hex = bytesToHex(data);buf.release();String content = ((ByteBuf) msg).toString(Charset.defaultCharset());}  // 读取消息帧类型else {byte messageType = buf.readByte();// 读取帧尾标识if (buf.isReadable()) {// 读取校验值byte checksum = buf.readByte();byte frameTail = buf.readByte();}}buf.release();}/*** 功能描述: 服务端给客户端发送消息** @param channelId 连接通道唯一id* @param msg       需要发送的消息内容* @return void*/public void channelWrite(ChannelId channelId, Object msg) throws Exception {Channel channel = ChannelMap.getChannelMap().get(channelId);if (channel == null) {log.info("通道:{},不存在", channelId);return;}if (msg == null || msg == "") {log.info("服务端响应空的消息");return;}//将客户端的信息直接返回写入ctxchannel.write(msg);//刷新缓存区channel.flush();}@Overridepublic void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {String socketString = ctx.channel().remoteAddress().toString();if (evt instanceof IdleStateEvent) {IdleStateEvent event = (IdleStateEvent) evt;if (event.state() == IdleState.READER_IDLE) {log.info("Client:{},READER_IDLE 读超时", socketString);Channel channel = ctx.channel();ChannelId id = channel.id();// 超时未收到心跳包，更新设备状态为离线// todo 更新设备状态ctx.disconnect();ChannelMap.removeChannelByName(id);} else if (event.state() == IdleState.WRITER_IDLE) {log.info("Client:{}, WRITER_IDLE 写超时", socketString);ctx.disconnect();Channel channel = ctx.channel();ChannelId id = channel.id();ChannelMap.removeChannelByName(id);} else if (event.state() == IdleState.ALL_IDLE) {log.info("Client:{},ALL_IDLE 总超时", socketString);Channel channel = ctx.channel();ChannelId id = channel.id();// 超时未收到心跳包，更新设备状态为离线// todo 更新设备状态ctx.disconnect();ChannelMap.removeChannelByName(id);}}}/*** 功能描述: 发生异常会触发此函数** @param ctx   通道处理程序上下文* @param cause 异常* @return void*/@Overridepublic void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {ctx.close();log.info("{}:发生了错误,此连接被关闭。此时连通数量:{}", ctx.channel().id(), ChannelMap.getChannelMap().size());}}

ChannelMap类

/*** 功能描述: 管理通道Map类**/
public class ChannelMap {/*** 管理一个全局map，保存连接进服务端的通道数量*/private static final ConcurrentHashMap<ChannelId, Channel> CHANNEL_MAP = new ConcurrentHashMap<>(128);public static ConcurrentHashMap<ChannelId, Channel> getChannelMap() {return CHANNEL_MAP;}/***  获取指定name的channel*/public static Channel getChannelByName(ChannelId channelId){if(CollectionUtils.isEmpty(CHANNEL_MAP)){return null;}return CHANNEL_MAP.get(channelId);}/***  将通道中的消息推送到每一个客户端*/public static boolean pushNewsToAllClient(String obj){if(CollectionUtils.isEmpty(CHANNEL_MAP)){return false;}for(ChannelId channelId: CHANNEL_MAP.keySet()) {Channel channel = CHANNEL_MAP.get(channelId);channel.writeAndFlush(new TextWebSocketFrame(obj));}return true;}/***  将channel和对应的name添加到ConcurrentHashMap*/public static void addChannel(ChannelId channelId,Channel channel){CHANNEL_MAP.put(channelId,channel);}/***  移除掉name对应的channel*/public static boolean removeChannelByName(ChannelId channelId){if(CHANNEL_MAP.containsKey(channelId)){CHANNEL_MAP.remove(channelId);return true;}return false;}}