SpringBoot+Netty

Posted 2023-02-22

<!-- netty依赖 springboot2.x自动导入版本 -->
        <dependency>
            <groupId>io.netty</groupId>
            <artifactId>netty-all</artifactId>
        </dependency>

public class AllServiceApplication implements CommandLineRunner 
    private Logger logger = LoggerFactory.getLogger(this.getClass());

    public static void main(String[] args) 
        SpringApplication.run(AllServiceApplication.class, args);
    
    @Autowired
    private NettyServer nettyServer;

    /**
     * 此处启动netty服务
     *
     * @param args
     * @throws Exception
     */
    @Override
    public void run(String... args) throws Exception 
        InetSocketAddress address = new InetSocketAddress(6802);
        // 启动netty服务器
        ChannelFuture channelFuture = nettyServer.startServer(address);
        // 钩子方法，关闭服务器
        Runtime.getRuntime().addShutdownHook(new Thread(() ->
                nettyServer.closeServer()
        ));
        channelFuture.channel().closeFuture().sync();
    

NettyServer.java

import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelOption;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioserverSocketChannel;
import io.netty.handler.logging.LogLevel;
import io.netty.handler.logging.LoggingHandler;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.net.InetSocketAddress;

@Component
public class NettyServer 
    private final EventLoopGroup bossGroup = new NioEventLoopGroup();
    private final EventLoopGroup workerGroup = new NioEventLoopGroup();
    private Channel channel;
    private Logger logger = LoggerFactory.getLogger(this.getClass());
    @Autowired
    private ServerInit serverInit;

    public ChannelFuture startServer(InetSocketAddress address) 
        ChannelFuture channelFuture = null;
        try 
            ServerBootstrap serverBootstrap = new ServerBootstrap();
            serverBootstrap.group(bossGroup, workerGroup)
                    .channel(NioServerSocketChannel.class)
                    .handler(new LoggingHandler(LogLevel.INFO))
                    .childHandler(serverInit)
                    // 允许的最大连接数
                    .option(ChannelOption.SO_BACKLOG, 128)
                    .childOption(ChannelOption.SO_KEEPALIVE, Boolean.TRUE);
            channelFuture = serverBootstrap.bind(address).sync();
            channel = channelFuture.channel();
         catch (Exception e) 
            logger.error("netty启动出错！", e);
         finally 
            if (channelFuture != null && channelFuture.isSuccess()) 
                logger.info("netty正在监听" + address.getHostName() + " 于端口 " + address.getPort() + ", 等待连接");
             else 
                logger.error("netty启动失败！");
            
        
        return channelFuture;
    

    /**
     * 关闭netty服务
     */
    public void closeServer() 
        logger.info("关闭netty服务。。。");
        if (channel != null) 
            channel.close();
        
        bossGroup.shutdownGracefully();
        workerGroup.shutdownGracefully();
        logger.info("关闭netty服务成功！");
    

ServerHandler.java

package com.lanyu.cloud.service.netty.other;

import com.alibaba.fastjson.JSONObject;
import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.handler.timeout.IdleStateEvent;
import io.netty.util.AttributeKey;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Component;

import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;

@Component
@ChannelHandler.Sharable
public class ServerHandler extends SimpleChannelInboundHandler<String> 
    private Logger logger = LoggerFactory.getLogger(this.getClass());
    private AttributeKey<String> key = AttributeKey.valueOf("deviceId");

    /**
     * 心跳监测
     *
     * @param ctx
     * @param evt
     * @throws Exception
     */
    @Override
    public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception 
        if (evt instanceof IdleStateEvent) 
            IdleStateEvent event = (IdleStateEvent) evt;
            String eventType = null;
            switch (event.state()) 
                case READER_IDLE:
                    eventType = "读空闲";
                    break;
                case WRITER_IDLE:
                    eventType = "些空闲";
                    break;
                case ALL_IDLE:
                    eventType = "读写空闲";
                    break;
                default:
            
            String date = LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"));
            logger.info(date + " " + ctx.channel().remoteAddress() + " " + eventType);
        
    

    /**
     * 客户端连接成功执行方法
     *
     * @param ctx
     * @throws Exception
     */
    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception 
        String date = LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"));
        logger.info(date + " " + ctx.channel().remoteAddress() + " 客户端连接成功！");
        ctx.writeAndFlush("连接成功！");
    

    /**
     * 客户端下线
     *
     * @param ctx
     * @throws Exception
     */
    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception 
        String date = LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"));
        logger.info(date + " " + ctx.channel().remoteAddress() + " 客户端下线！");
    

    /**
     * 客户端连接成功后执行的回调方法
     *
     * @param ctx
     * @throws Exception
     */
    @Override
    public void handlerAdded(ChannelHandlerContext ctx) throws Exception 
        String date = LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"));
        logger.info(date + " " + ctx.channel().remoteAddress() + " 新的客户端加入！");
    

    /**
     * 断开连接
     *
     * @param ctx
     * @throws Exception
     */
    @Override
    public void handlerRemoved(ChannelHandlerContext ctx) throws Exception 
        String date = LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"));
        logger.info(date + " " + ctx.channel().remoteAddress() + " 客户端断开连接！");
    

    /**
     * 抛出异常
     *
     * @param ctx
     * @param cause
     * @throws Exception
     */
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) 
        logger.error("netty 服务出错！", cause);
        ctx.close();
    


    @Override
    protected void channelRead0(ChannelHandlerContext ctx, String message) 
//        Long deviceId = CtxUtils.getDeviceId(ctx);
//        CtxUtils.setDeviceId(ctx, System.currentTimeMillis());
        logger.error(message);
        JSONObject messageJson = JSONObject.parseObject(message);
        CommandExecutionMap.getInstance().handle(messageJson.getString("cmd"),
                messageJson, ctx);

    

ServerInit.java

import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.socket.SocketChannel;
import io.netty.handler.codec.string.StringDecoder;
import io.netty.handler.codec.string.StringEncoder;
import io.netty.handler.timeout.IdleStateHandler;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.nio.charset.StandardCharsets;
import java.util.concurrent.TimeUnit;

@Component
public class ServerInit extends ChannelInitializer<SocketChannel> 
    @Autowired
    private ServerHandler serverHandler;


    @Override
    protected void initChannel(SocketChannel ch) 
        ChannelPipeline pipeline = ch.pipeline();
        pipeline.addLast(new StringDecoder(StandardCharsets.UTF_8));
        pipeline.addLast(new StringEncoder(StandardCharsets.UTF_8));
        pipeline.addLast(new IdleStateHandler(30000, 0, 0, TimeUnit.SECONDS));
        pipeline.addLast(serverHandler);