spring2.x集成netty4.x

正文

添加依赖

maven 添加如下依赖

<parent>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-parent</artifactId>
    <version>2.1.4.RELEASE</version>
    <relativePath/> <!-- lookup parent from repository -->
</parent>

<properties>
    <netty.version>4.1.22.Final</netty.version>
</properties>

<dependencies>
    <!-- Netty -->
    <dependency>
        <groupId>io.netty</groupId>
        <artifactId>netty-all</artifactId>
        <version>${netty.version}</version>
    </dependency>
    <!-- Netty -->
</dependencies>

添加配置

application.properties 配置文件添加如下配置

netty.host=0.0.0.0
netty.port=12268
netty.bossNum=2
netty.workerNum=4
netty.backlog=4096

添加配置类和启动服务类

package com.utstar.filemonitoring.netty;

import lombok.Data;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.stereotype.Component;

/**
 * FileName: NettyConf
 * Author:   creambing
 * Date:     2019-09-17 11:13
 * Description:
 * History:
 * <author>          <time>          <version>          <desc>
 * 作者姓名           修改时间           版本号              描述
 */
@ConfigurationProperties(prefix = "netty")
@Component
@Data
public class NettyConf {
    private String host;
    private int port;
    private int bossNum;
    private int workerNum;
    private int backlog;
    private Long idleTime;

    @Value("${netty.host:127.0.0.1}")
    public void setNettyHost(String nettyHost) {
        NETTY_HOST = nettyHost;
    }

    @Value("${netty.port:8888}")
    public void setNettyPort(Integer nettyPort) {
        NETTY_PORT = nettyPort;
    }

    @Value("${netty.bossNum:1}")
    public void setNettyBossnum(Integer nettyBossnum) {
        NETTY_BOSSNUM = nettyBossnum;
    }

    @Value("${netty.workerNum:2}")
    public void setNettyWorkernum(Integer nettyWorkernum) {
        NETTY_WORKERNUM = nettyWorkernum;
    }

    @Value("${netty.backlog:1024}")
    public void setNettyBacklog(Integer nettyBacklog) {
        NETTY_BACKLOG = nettyBacklog;
    }

    @Value("${netty.idleTime:60}")
    public void setNettyIdletime(Long nettyIdletime) {
        NETTY_IDLETIME = nettyIdletime;
    }

    public static String NETTY_HOST;
    public static Integer NETTY_PORT;
    public static Integer NETTY_BOSSNUM;
    public static Integer NETTY_WORKERNUM;
    public static Integer NETTY_BACKLOG;
    public static Long NETTY_IDLETIME;
}

package com.creambing.netty.server.thread;

import com.creambing.netty.server.handler.*;
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 io.netty.handler.logging.LogLevel;
import io.netty.handler.logging.LoggingHandler;
import lombok.extern.slf4j.Slf4j;
import org.springframework.context.annotation.DependsOn;
import org.springframework.stereotype.Component;

import javax.annotation.PreDestroy;

import java.io.PrintWriter;
import java.io.StringWriter;

import static com.creambing.netty.server.conf.NettyConf.*;
import static com.creambing.netty.server.constants.NettyConstants.*;

/**
 * FileName: NettyThread
 * Author:   creambing
 * Date:     2019-09-17 13:19
 * Description:
 * History:
 * <author>          <time>          <version>          <desc>
 * 作者姓名           修改时间           版本号              描述
 */
@DependsOn("nettyConf")
@Component
@Slf4j
public class NettyThread implements Runnable {

    private final EventLoopGroup bossGroup = new NioEventLoopGroup(NETTY_BOSSNUM);
    private final EventLoopGroup workerGroup = new NioEventLoopGroup(NETTY_WORKERNUM);

    @Override
    public void run() {
        try {
            ServerBootstrap b = new ServerBootstrap();
            b.option(ChannelOption.SO_BACKLOG, NETTY_BACKLOG)
                    .option(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT)
                    .childOption(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT)
                    .childOption(ChannelOption.SO_KEEPALIVE, true)
                    .childOption(ChannelOption.TCP_NODELAY, true)
                    .childOption(ChannelOption.SO_REUSEADDR, true) //重用地址
                    .childOption(ChannelOption.SO_RCVBUF, 65536)
                    .childOption(ChannelOption.SO_SNDBUF, 65536);
            b.group(bossGroup, workerGroup)
                    .channel(NioServerSocketChannel.class)
                    .handler(new LoggingHandler(LogLevel.INFO))
                    .childHandler(new ChannelInitializer<SocketChannel>() {//每连上一个链接调用一次
                        @Override
                        protected void initChannel(SocketChannel ch) {
                            log.info("Accept new connection:  " + ch);
                            // 设定读空闲时间，在这段时间内如果没有数据读到，就表示连接假死；
                            ch.pipeline().addLast("idle-state-handler", new NettyIdleStateHandler());
                            ch.pipeline().addLast(new AcceptorIdleStateTrigger());
                            ch.pipeline().addLast("decoder", new NewLengthFieldBasedFrameDecoder(MAX_FRAME_LENGTH, LENGTH_FIELD_LENGTH, LENGTH_FIELD_OFFSET, LENGTH_ADJUSTMENT, INITIAL_BYTES_TO_STRIP, false));
                            ch.pipeline().addLast("new-encoder", new NewEncoder());
                            ch.pipeline().addLast("heartbeat", new HeartBeatReceiveHandler());
                            ch.pipeline().addLast("process", new NettyBusinessHandler());
                        }
                    });
            // 开始真正绑定端口进行监听
            ChannelFuture future = b.bind(NETTY_HOST, NETTY_PORT).sync();
            Channel channel = future.channel();
            log.info("listening on [{}] at [{}] and ready for connection ...", NETTY_HOST, NETTY_PORT);
            if (future.isSuccess()) {
                log.info("netty server start success");
            }
            channel.closeFuture().sync();
        } catch (Exception e) {
            StringWriter stringWriter = new StringWriter();
            e.printStackTrace(new PrintWriter(stringWriter));
            log.error("start netty server has exception,caused by[{}]", stringWriter.toString());
        }
    }

    /**
     * 销毁
     */
    @PreDestroy
    public void stop() {
        bossGroup.shutdownGracefully().syncUninterruptibly();
        bossGroup.shutdownGracefully().syncUninterruptibly();
        log.info("stop netty server success");
    }
}

package com.creambing.netty.server;

import com.creambing.netty.server.thread.NettyThread;
import lombok.extern.slf4j.Slf4j;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;

import javax.annotation.PostConstruct;
import javax.annotation.Resource;

@Slf4j
@SpringBootApplication
public class ServerApplication {

    public static void main(String[] args) {
        SpringApplication.run(ServerApplication.class, args);
    }


    @Resource
    NettyThread nettyThread;

    @PostConstruct
    public void startCsServer() {
        log.info("start netty server ...");
        Thread t = new Thread(nettyThread);
        t.start();
    }

}

参考资料