<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>

    <groupId>com.hzx.testmaven15netty</groupId>
    <artifactId>testmaven15netty</artifactId>
    <version>1.0-SNAPSHOT</version>

    <parent>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-parent</artifactId>
        <version>2.3.0.RELEASE</version>
    </parent>
    <dependencies>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter</artifactId>
        </dependency>
        <dependency>
            <groupId>io.netty</groupId>
            <artifactId>netty-all</artifactId>
            <version>4.1.31.Final</version>
        </dependency>
    </dependencies>

</project>

server:
  port: 8080

# 作為客戶端請求的服務(wù)端地址
netty:
  tcp:
    server:
      # 作為客戶端請求的服務(wù)端地址
      host: 127.0.0.1
      # 作為客戶端請求的服務(wù)端端口
      port: 7000
    client:
      # 作為服務(wù)端開放給客戶端的端口
      port: 7000

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.util.concurrent.Future;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

@Component
public class NettyTcpServer {
    private static final Logger LOGGER = LoggerFactory.getLogger(NettyTcpServer.class);

    // boss事件輪詢線程組
    // 處理Accept連接事件的線程，這里線程數(shù)設(shè)置為1即可，netty處理鏈接事件默認(rèn)為單線程，過度設(shè)置反而浪費(fèi)cpu資源
    private EventLoopGroup boss = new NioEventLoopGroup(1);

    //worker事件輪詢線程組
    //處理handler的工作線程，其實也就是處理IO讀寫 。線程數(shù)據(jù)默認(rèn)為 CPU 核心數(shù)乘以2
    private EventLoopGroup worker = new NioEventLoopGroup();

    @Autowired
    ServerChannelInitializer serverChannelInitializer;

    @Value("${netty.tcp.client.port}")
    private Integer port;

    // 與客戶端建立連接后得到的通道對象
    private Channel channel;

    /**
     * 存儲client的channel
     * key:ip value:Channel
     */
    public static Map<String, Channel> map = new ConcurrentHashMap<String, Channel>();

    /**
     * 開啟Netty tcp server服務(wù)
     *
     * @return
     */
    public ChannelFuture start() {
        // 啟動類
        ServerBootstrap serverBootstrap = new ServerBootstrap();
        serverBootstrap.group(boss, worker)//組配置，初始化ServerBootstrap的線程組
                .channel(NioServerSocketChannel.class)//構(gòu)造channel通道工廠 bossGroup的通道，只是負(fù)責(zé)連接
                .childHandler(serverChannelInitializer) //設(shè)置通道處理者ChannelHandlerWorkerGroup的處理器
                .option(ChannelOption.SO_BACKLOG, 1024)//socket參數(shù)，當(dāng)服務(wù)器請求處理程全滿時，用于臨時存放已完成三次握手請求的隊列的最大長度。如果未設(shè)置或所設(shè)置的值小于1，Java將使用默認(rèn)值50。
                .childOption(ChannelOption.SO_KEEPALIVE, true);//啟用心跳?；顧C(jī)制，tcp，默認(rèn)2小時發(fā)一次心跳
        //Future：異步任務(wù)的生命周期，可用來獲取任務(wù)結(jié)果
        ChannelFuture channelFuture1 = serverBootstrap.bind(port).syncUninterruptibly(); // 綁定端口 開啟監(jiān)聽 同步等待
        if (channelFuture1 != null && channelFuture1.isSuccess()) {
            channel = channelFuture1.channel();// 獲取通道
            LOGGER.info("Netty tcp server start success,port={}",port);
        }else {
            LOGGER.error("Netty tcp server start fail");
        }
        return channelFuture1;
    }

    /**
     * 停止Netty tcp server服務(wù)
     */
    public void destroy(){
        if (channel != null) {
            channel.close();
        }
        try {
            Future<?> future = worker.shutdownGracefully().await();
            if (!future.isSuccess()) {
                LOGGER.error("netty tcp workerGroup shutdown fail,{}",future.cause());
            }
        } catch (InterruptedException e) {
            LOGGER.error(e.toString());
        }
        LOGGER.info("Netty tcp server shutdown success");
    }
}

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.util.concurrent.TimeUnit;

@Component
public class ServerChannelInitializer extends ChannelInitializer<SocketChannel> {

    @Autowired
    ServerChannelHandler serverChannelHandler;

    @Override
    protected void initChannel(SocketChannel socketChannel) throws Exception {
        ChannelPipeline pipeline = socketChannel.pipeline();
        //IdleStateHandler心跳機(jī)制,如果超時觸發(fā)Handle中userEventTrigger()方法
        pipeline.addLast("idleStateHandler",
                new IdleStateHandler(15,0,0, TimeUnit.MINUTES));
        // 字符串編解碼器
        pipeline.addLast(
                new StringDecoder(),
                new StringEncoder()
        );
        // 自定義Handler
        pipeline.addLast("serverChannelHandler",serverChannelHandler);
    }
}

import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.handler.timeout.IdleState;
import io.netty.handler.timeout.IdleStateEvent;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Component;

@Component
@ChannelHandler.Sharable
public class ServerChannelHandler extends SimpleChannelInboundHandler<Object> {

    private static final Logger LOGGER = LoggerFactory.getLogger(ServerChannelHandler.class);
    /**
     * 拿到傳過來的msg數(shù)據(jù)，開始處理
     * @param channelHandlerContext
     * @param msg
     * @throws Exception
     */
    @Override
    protected void channelRead0(ChannelHandlerContext channelHandlerContext, Object msg) throws Exception {
        LOGGER.info("Netty tcp server receive message: {}",msg);
        channelHandlerContext.writeAndFlush(" response message "+msg).syncUninterruptibly();
    }

    /**
     * 活躍的、有效的通道
     * 第一次連接成功后進(jìn)入的方法
     * @param ctx
     * @throws Exception
     */
    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        super.channelActive(ctx);
        LOGGER.info("tcp client "+getRemoteAddress(ctx)+" connect success");
        NettyTcpServer.map.put(getIPString(ctx),ctx.channel());
    }

    /**
     * 不活動的通道
     * 連接丟失后執(zhí)行的方法（client端可據(jù)此實現(xiàn)斷線重連）
     * @param ctx
     * @throws Exception
     */
    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        // 刪除Channel Map中失效的Client
        NettyTcpServer.map.remove(getIPString(ctx));
        ctx.close();
    }

    /**
     * 異常處理
     * @param ctx
     * @param cause
     * @throws Exception
     */
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        super.exceptionCaught(ctx, cause);
        // 發(fā)生異常 關(guān)閉連接
        LOGGER.error("引擎{}的通道發(fā)生異常，斷開連接",getRemoteAddress(ctx));
        ctx.close();
    }

    /**
     * 心跳機(jī)制 超時處理
     * @param ctx
     * @param evt
     * @throws Exception
     */
    @Override
    public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
        String socketString = ctx.channel().remoteAddress().toString();
        if (evt instanceof IdleStateEvent) {
            IdleStateEvent event = (IdleStateEvent) evt;
            if (event.state()== IdleState.READER_IDLE) {
                LOGGER.info("Client: "+socketString+" READER_IDLE讀超時");
                ctx.disconnect();
            }else if (event.state()==IdleState.WRITER_IDLE){
                LOGGER.info("Client: "+socketString+" WRITER_IDLE寫超時");
                ctx.disconnect();
            }else if (event.state()==IdleState.ALL_IDLE){
                LOGGER.info("Client: "+socketString+" ALL_IDLE總超時");
                ctx.disconnect();
            }
        }
    }

    /**
     * 獲取client對象：ip+port
     * @param channelHandlerContext
     * @return
     */
    public String getRemoteAddress(ChannelHandlerContext channelHandlerContext){
        String socketString = "";
        socketString = channelHandlerContext.channel().remoteAddress().toString();
        return socketString;
    }

    /**
     * 獲取client的ip
     * @param channelHandlerContext
     * @return
     */
    public String getIPString(ChannelHandlerContext channelHandlerContext){
        String ipString = "";
        String socketString = channelHandlerContext.channel().remoteAddress().toString();
        int colonAt = socketString.indexOf(":");
        ipString = socketString.substring(1,colonAt);
        return ipString;
    }
}

import io.netty.bootstrap.Bootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelOption;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import javax.xml.ws.Holder;

@Component
public class NettyTcpClient {

    private static final Logger LOGGER = LoggerFactory.getLogger(NettyTcpClient.class);

    @Value("${netty.tcp.server.host}")
    String HOST;

    @Value("${netty.tcp.server.port}")
    int PORT;

    @Autowired
    ClientChannelInitializer clientChannelInitializer;

    private Channel channel;

    /**
     * 初始化 `Bootstrap` 客戶端引導(dǎo)程序
     * @return
     */
    private final Bootstrap getBootstrap(){
        Bootstrap bootstrap = new Bootstrap();
        NioEventLoopGroup group = new NioEventLoopGroup();
        bootstrap.group(group)
                .channel(NioSocketChannel.class)//通道連接者
                .handler(clientChannelInitializer)//通道處理者
                .option(ChannelOption.SO_KEEPALIVE,true);// 心跳報活
        return bootstrap;
    }

    /**
     *  建立連接，獲取連接通道對象
     */
    public void connect(){
        ChannelFuture channelFuture = getBootstrap().connect(HOST, PORT).syncUninterruptibly();
        if (channelFuture != null&&channelFuture.isSuccess()) {
            channel = channelFuture.channel();
            LOGGER.info("connect tcp server host = {},port = {} success", HOST,PORT);
        }else {
            LOGGER.error("connect tcp server host = {},port = {} fail",HOST,PORT);
        }
    }

    /**
     * 向服務(wù)器發(fā)送消息
     */
    public void sendMessage(Object msg) throws InterruptedException {
        if (channel != null) {
            channel.writeAndFlush(msg).sync();
        }else {
            LOGGER.warn("消息發(fā)送失敗，連接尚未建立");
        }
    }
}

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.util.concurrent.TimeUnit;

@Component
public class ClientChannelInitializer extends ChannelInitializer<SocketChannel> {

    @Autowired
    ClientChannelHandler clientChannelHandler;

    @Override
    protected void initChannel(SocketChannel socketChannel) throws Exception {
        ChannelPipeline pipeline = socketChannel.pipeline();
        pipeline.addLast("idleStateHandler",
                new IdleStateHandler(15,0,0, TimeUnit.MINUTES));
        pipeline.addLast(new StringDecoder(),new StringEncoder());
        pipeline.addLast("clientChannelHandler",clientChannelHandler);
    }
}

import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Component;

@Component
@ChannelHandler.Sharable
public class ClientChannelHandler extends SimpleChannelInboundHandler<Object> {
    private static final Logger LOGGER = LoggerFactory.getLogger(ClientChannelHandler.class);
    @Override
    protected void channelRead0(ChannelHandlerContext channelHandlerContext, Object msg) throws Exception {
        LOGGER.info("Netty tcp client receive msg : " + msg);
    }

}

import com.netty.client.NettyTcpClient;
import com.netty.server.NettyTcpServer;
import io.netty.channel.ChannelFuture;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.CommandLineRunner;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;

@SpringBootApplication
public class StartApplication implements CommandLineRunner {
    public static void main(String[] args) throws Exception {
        SpringApplication.run(StartApplication.class, args);
    }

    @Autowired
    NettyTcpServer nettyTcpServer;

    @Autowired
    NettyTcpClient nettyTcpClient;

    /**
     * Callback used to run the bean.
     *
     * @param args incoming main method arguments
     * @throws Exception on error
     */
    @Override
    public void run(String... args) throws Exception {
        ChannelFuture start = nettyTcpServer.start();
        nettyTcpClient.connect();
        for (int i = 0; i < 10; i++) {
            nettyTcpClient.sendMessage("hello world "+i);
        }
        start.channel().closeFuture().syncUninterruptibly();
    }
}