springboot引入netty

配置类

import cn.hutool.core.thread.ThreadUtil;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.buffer.PooledByteBufAllocator;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;import javax.annotation.PostConstruct;
import javax.annotation.Resource;@Component
@Slf4j
public class NettyServerConfig {@Resourceprivate BizHandler bizHandler;@Value("${netty.port}")private int port;@PostConstructpublic void start(){ThreadUtil.newSingleExecutor().execute(this::doStart);}public void doStart(){ServerBootstrap serverBootstrap = new ServerBootstrap();EventLoopGroup bossGroup = new NioEventLoopGroup();EventLoopGroup workGroup = new NioEventLoopGroup();serverBootstrap.group(bossGroup,workGroup).channel(NioServerSocketChannel.class).option(ChannelOption.SO_BACKLOG, 1024).childOption(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT).childHandler(new ChannelInitializer<SocketChannel>(){@Overrideprotected void initChannel(SocketChannel socketChannel) {ChannelPipeline pipeline = socketChannel.pipeline();pipeline.addFirst(bizHandler);}});ChannelFuture channelFuture;try {channelFuture = serverBootstrap.bind(port).sync();} catch (InterruptedException e) {throw new RuntimeException(e);}channelFuture.addListener(future -> {if(future.isSuccess()){log.info("netty服务 启动成功");}});ChannelFuture closeFuture = channelFuture.channel().closeFuture();try {closeFuture.sync();} catch (InterruptedException e) {e.printStackTrace();}finally {bossGroup.shutdownGracefully();workGroup.shutdownGracefully();}}
}

处理器

import com.google.common.collect.Lists;
import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;import java.util.List;
import java.util.concurrent.ConcurrentHashMap;
import java.util.stream.Collectors;@Slf4j
@Service
@ChannelHandler.Sharable
public class BizHandler extends ChannelInboundHandlerAdapter {private static final String SPACE = " ";private static final ConcurrentHashMap<ChannelHandlerContext, String> channelContextMap = new ConcurrentHashMap<>();public void send(){channelContextMap.entrySet().forEach(e -> {ChannelHandlerContext context = e.getKey();ByteBuf buf = context.alloc().buffer();byte[] dataFrame = generateDataFrame();buf.writeBytes(dataFrame);context.channel().writeAndFlush(buf);});}@Overridepublic void channelRead(ChannelHandlerContext ctx, Object msg) {ByteBuf buffer = (ByteBuf) msg;byte[] bytes = new byte[buffer.readableBytes()];buffer.readBytes(bytes);String msgHexStr = this.byteArrayToHexString(bytes);//如果不是预期的连接发过来消息 主动断开连接//ctx.channel().close();ctx.fireChannelRead(msg);}@Overridepublic void channelInactive(ChannelHandlerContext ctx) {channelContextMap.remove(ctx);}private String byteArrayToHexString(byte[] bytes) {List<String> dataFrame = Lists.newArrayList();for (byte b : bytes) {String hex = Integer.toHexString(0xFF & b);if (hex.length() == 1) {// 如果是一位的话，要补0hex = "0" + hex;}dataFrame.add(hex.toUpperCase());}return dataFrame.stream().collect(Collectors.joining(SPACE));}}