SpringBoot+Netty 防止主線程阻塞

 

https://blog.csdn.net/qq_31145141/article/details/103864600

 

@SpringBootApplication
@MapperScan("com.lhb.dao") // mybatis掃描
@EnableAsync//注意這里，這個注解啟用了線程池
public class Application extends SpringApplication {
    public static void main(String[] args) {
        SpringApplication.run(Application.class);
        System.out.println("SpringBoot start");
    }
}

　　

 

 

package com.lhb.netty;
 
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.boot.CommandLineRunner;
import org.springframework.core.annotation.Order;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Component;
 
import javax.annotation.PreDestroy;
 
/**
 * 作者：Haibo.Liu
 * 描述：
 * 日期： 2019/4/23
 * QQ：836915746
 */
@Component//當成組件處理
@Order(value = 1)//這里表示啟動順序
public class Server implements CommandLineRunner {
 
    private Logger logger = LoggerFactory.getLogger(this.getClass());
 
    private EventLoopGroup bossGroup;
    private EventLoopGroup workerGroup;
    private Channel channel;
 
    public void start() {
        bossGroup = new NioEventLoopGroup(); //監聽線程組，監聽客戶請求
        workerGroup = new NioEventLoopGroup();//處理客戶端相關操作線程組，負責處理與客戶端的數據通訊
        try {
            ServerBootstrap serverBootstrap = new ServerBootstrap();
            //設置監聽組，線程組，初始化器
            serverBootstrap.group(bossGroup, workerGroup).channel(NioServerSocketChannel.class).childHandler(new ChannelInitializer() {
                @Override
                protected void initChannel(Channel channel) {
                    ChannelPipeline pipeline = channel.pipeline();
                    pipeline.addLast(new ServerHandler());
                    logger.info("Client :" + channel.remoteAddress() + "已經連接上");
                }
            });
            logger.info("Netty Server start");
            //綁定端口號
            ChannelFuture f = serverBootstrap.bind(6300).sync();
            channel = f.channel().closeFuture().sync().channel();//這里綁定端口啟動后，會阻塞線程，也就是為什么要用線程池啟動的原因
 
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            workerGroup.shutdownGracefully();
            bossGroup.shutdownGracefully();
            stop();
            logger.info("Netty Server close");
        }
    }
 
    @PreDestroy
    public void stop() {
        if (channel != null) {
            logger.info("Netty Server close");
            workerGroup.shutdownGracefully();
            bossGroup.shutdownGracefully();
        }
    }
 
    @Async//注意這里，組件啟動時會執行run，這個注解是讓線程異步執行，這樣不影響主線程
    @Override
    public void run(String... args) {
        start();
    }
}

　　

package com.lhb.netty;
 
import com.lhb.dao.WtsrEquipmentMapper;
import com.lhb.redis.RedisUtil;
import io.netty.buffer.ByteBuf;
import io.netty.channel.Channel;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.ByteToMessageDecoder;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Component;
 
import javax.annotation.Resource;
import java.math.BigInteger;
import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
 
/**
 * 作者：Haibo.Liu
 * 描述：
 * 日期： 2019/4/23
 * QQ：836915746
 */
@Component//ServerHandler 當成組件處理，這樣可以直接注入bean
public class ServerHandler extends ByteToMessageDecoder {
 
    private static Logger logger = LoggerFactory.getLogger(ServerHandler.class);
 
    //這里是你要注入的bean
    private static RedisUtil redisUtil;
    private static WtsrEquipmentMapper wtsrEquipmentMapper;
 
    @Resource
    public void setWtsrEquipmentMapper(WtsrEquipmentMapper wtsrEquipmentMapper) {
        ServerHandler.wtsrEquipmentMapper = wtsrEquipmentMapper;
    }
 
    @Resource
    public void setRedisUtil(RedisUtil redisUtil) {
        ServerHandler.redisUtil = redisUtil;
    }
 
 
    @Override
    protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) {
        byte[] bytes = new byte[in.readableBytes()];
        in.readBytes(bytes);
        
    }
 
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
        Channel incoming = ctx.channel();
        cause.printStackTrace();
        ctx.close();
    }
 
    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        logger.info("tcp斷開");
        ctx.fireChannelInactive();
        Channel channel = ctx.channel();
        if (channel != null && channel.isActive()) {
            ctx.close();
        }
    }
 
  
 
    public static void sendDeviceMsg(ChannelHandlerContext ctx, String msg) {
        if (ctx != null) {
            byte[] bytes = msg.getBytes();
            ByteBuf encoded = ctx.alloc().buffer(bytes.length);
            encoded.writeBytes(bytes);
            ctx.channel().writeAndFlush(encoded);
        }
    }
 
 
}