TSDDCS/tsddcs-datacollect-stream/src/main/java/org/ts/ddcs/net/DataChannelTcpWithNio.java

package org.ts.ddcs.net;


import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import org.ts.ddcs.log.LogHelper;


/**
 * TCP网络连接
 *
 * @author dylan
 */
public class DataChannelTcpWithNio {
    private String channelKey;
    private String bindAgreeement;

    private boolean nioChannelRunning = false;
    /*
    监听端口
     */
    private int port = 0;

    private EventLoopGroup bossGroup = null;
    private EventLoopGroup workerGroup = null;

    public DataChannelTcpWithNio(String key, String agreement) {
        this.channelKey = key;
        this.bindAgreeement = agreement;
    }

    public DataChannelTcpWithNio colseChannel() {
        this.nioChannelRunning = false;
        if (null != this.workerGroup) {
            this.workerGroup.shutdownGracefully();
            this.workerGroup = null;
        }
        if (null != this.bossGroup) {
            this.bossGroup.shutdownGracefully();
            this.bossGroup = null;
        }
        LogHelper.info("******************** 关闭 NIO ********************");
        return this;
    }

    public DataChannelTcpWithNio openChannel(int port) {
        this.nioChannelRunning = true;
        this.port = port;
        LogHelper.info("******************** 启动 NIO ********************");
        return this;
    }

    public void async() {
        if (!this.nioChannelRunning) {
            LogHelper.error("调用 Async() 发生错误:未打开通道");
            return;
        }
        new AcceptThread().start();
    }

    private class AcceptThread extends Thread {

        @Override
        public void run() {
            bossGroup = new NioEventLoopGroup();
            workerGroup = new NioEventLoopGroup();
            try {
                ServerBootstrap b = new ServerBootstrap();
                b.group(bossGroup, workerGroup)
                        .channel(NioServerSocketChannel.class)
                        .childHandler(new ChannelInitializer<SocketChannel>() {
                            @Override
                            public void initChannel(SocketChannel ch) throws Exception {
                                ch.pipeline().addLast(new DataDiscardHandler(channelKey,bindAgreeement));
                            }
                        })
                        .option(ChannelOption.SO_BACKLOG, 128)
                        .childOption(ChannelOption.SO_KEEPALIVE, true);
                ChannelFuture f = b.bind(port).sync();
                f.channel().closeFuture().sync();
            } catch (InterruptedException e) {
                LogHelper.error(e);
            } finally {
                if (null != workerGroup) {
                    workerGroup.shutdownGracefully();
                }
                if (null != bossGroup) {
                    bossGroup.shutdownGracefully();
                }
            }
        }
    }
}