netty服務端實戰(一)

　　先分享一下自己的經歷。

　　去年7月進入新公司沒多久，部門領導就給我分配了一個任務：給公司的一個戶外設備寫一個采集數據程序，將數據入庫，然后做一個web端。因為領導是做.NET的，當時在來之前有和領導溝通過，領導的意思是希望來一個會網絡編程和多線程，部門急需一個可以來做采集程序的java，我當時有點心虛，是這樣回復領導：自己也只是有1年多java的工作，不會網絡編程，自己搭一個簡單的web項目架構還是可以應付的，我只能盡自己的最大的可能做好這件事。

　　后來進入公司后，做采集程序是一頭霧水，自己之前只是做web的，對web項目的架構以及業務還是比較熟悉，但是關於和硬件之間的通訊還真的是從來都沒有接觸過。It is the first step that is troublesome。問過一些朋友，也百度過看過一些技術博客，有用socket 連接，多線程后加while死循環來接受做采集，后來慢慢發現用netty這種NIO網絡應用框架比較合適。

　　接受就是熟悉一下具體的業務，戶外設備會主動連接服務器IP，定時向服務器的指定端口發送報文。所以當前的采集程序結合netty的情況就是需要寫一個netty的server端，來監聽服務器本機的該指定端口，接受戶外設備發送過來的報文即可，而不需要再寫一個netty的client端。

　　首先說明一下，本項目應用的框架：springboot+netty+rabbitMq+mybatis+lombok+logback，數據庫選用的是MYSQL。

　　開發環境：idea2018+jdk1.8+mysql5.6.35+maven3.5.3

　　接下來就是搭建項目：

　　1.idea快速創建springboot項目，在pom.xml文件中配置依賴包

<dependencies>
        <!--web模塊starter依賴-->
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-test</artifactId>
            <scope>test</scope>
        </dependency>

        <!--  netty依賴 springboot2.0自動導入版本-->
        <dependency>
            <groupId>io.netty</groupId>
            <artifactId>netty-all</artifactId>
        </dependency>
        <!--rabbitmq依賴-->
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-amqp</artifactId>
        </dependency>
        <!--mybatis依賴-->
        <dependency>
            <groupId>org.mybatis.spring.boot</groupId>
            <artifactId>mybatis-spring-boot-starter</artifactId>
            <version>1.3.2</version>
        </dependency>
        <!--mysql依賴-->
        <dependency>
            <groupId>mysql</groupId>
            <artifactId>mysql-connector-java</artifactId>
            <scope>runtime</scope>
        </dependency>

        <!--pagehelper 分頁插件-->
        <dependency>
            <groupId>com.github.pagehelper</groupId>
            <artifactId>pagehelper-spring-boot-starter</artifactId>
            <version>1.2.5</version>
        </dependency>
        <!--lombok 插件-->
        <dependency>
            <groupId>org.projectlombok</groupId>
            <artifactId>lombok</artifactId>
            <version>1.16.6</version>
        </dependency>
        <dependency>
            <groupId>org.junit.jupiter</groupId>
            <artifactId>junit-jupiter-api</artifactId>
            <version>RELEASE</version>
            <scope>compile</scope>
        </dependency>
        <!--編寫更少量的代碼：使用apache commons工具類庫:
        https://www.cnblogs.com/ITtangtang/p/3966955.html-->
        <!--apache.commons.lang3-->
        <dependency>
            <groupId>org.apache.commons</groupId>
            <artifactId>commons-lang3</artifactId>
        </dependency>
        <!--你可以把這個工具看成是java.util的擴展-->
        <dependency>
            <groupId>org.apache.commons</groupId>
            <artifactId>commons-collections4</artifactId>
            <version>${commons-collections4.version}</version>
        </dependency>
        <!--apache.codec:編碼方法的工具類包
        https://blog.csdn.net/u012881904/article/details/52767853-->
        <dependency>
            <groupId>commons-codec</groupId>
            <artifactId>commons-codec</artifactId>
        </dependency>
    </dependencies>

2.自定義netty的server端BootNettyServer類

/**
 * netty的server類
 */
@Component
@Slf4j
public class BootNettyServer {
    @Autowired
    BootNettyInitializer bootNettyInitializer;

    public void run(int port) throws Exception {
        /**
         * 配置服務端的NIO線程組
         * NioEventLoopGroup 是用來處理I/O操作的Reactor線程組
         * bossGroup：用來接收進來的連接，workerGroup：用來處理已經被接收的連接,進行socketChannel的網絡讀寫，
         * bossGroup接收到連接后就會把連接信息注冊到workerGroup
         * workerGroup的EventLoopGroup默認的線程數是CPU核數的二倍
         */
        EventLoopGroup bossGroup = new NioEventLoopGroup(1);
        EventLoopGroup workerGroup = new NioEventLoopGroup();
        try {
            ServerBootstrap serverBootstrap = new ServerBootstrap();//ServerBootstrap 是一個啟動NIO服務的輔助啟動類
            serverBootstrap
                    .group(bossGroup, workerGroup)//設置group，將bossGroup， workerGroup線程組傳遞到ServerBootstrap
                    .channel(NioServerSocketChannel.class)//ServerSocketChannel是以NIO的selector為基礎進行實現的，用來接收新的連接，這里告訴Channel通過NioServerSocketChannel獲取新的連接
                    /**
                     * option是設置 bossGroup，childOption是設置workerGroup
                     * netty 默認數據包傳輸大小為1024字節, 設置它可以自動調整下一次緩沖區建立時分配的空間大小，避免內存的浪費    最小  初始化  最大 (根據生產環境實際情況來定)
                     * 使用對象池，重用緩沖區
                     */
                    .option(ChannelOption.RCVBUF_ALLOCATOR, new AdaptiveRecvByteBufAllocator(64, 10496, 1048576))
                    .childOption(ChannelOption.RCVBUF_ALLOCATOR, new AdaptiveRecvByteBufAllocator(64, 10496, 1048576))
                    .childHandler(bootNettyInitializer);//設置 I/O處理類,主要用於網絡I/O事件，記錄日志，編碼、解碼消息
            System.out.println("正在監聽8234端口中....");
            ChannelFuture f = serverBootstrap.bind(port).sync();//綁定端口，同步等待成功
            f.channel().closeFuture().sync();//等待服務器監聽端口關閉
        } catch (InterruptedException e) {

        } finally {
            /**
             * 退出，釋放線程池資源
             */
            bossGroup.shutdownGracefully().sync();
            workerGroup.shutdownGracefully().sync();
        }
    }
}

3.自定義初始化類BootNettyInitializer類

　　繼承ChannelInitializer,實現initChannel方法，初始化信道時可以配置心跳、編碼器、解碼器、以及業務處理ChannelHandler類。

/**
 * Channel初始化器
 * @param <SocketChannel>
 */
@Component
public class BootNettyInitializer<SocketChannel> extends ChannelInitializer<Channel> {

    @Autowired
    BootNettyHandler bootNettyHandler;
    @Autowired
    MyDecoder myDecoder;
    @Override
    protected void initChannel(Channel ch) throws Exception {
        // ChannelOutboundHandler，依照逆序執行
        ch.pipeline().addLast("decoder", myDecoder);
      
        ch.pipeline().addLast(bootNettyHandler);//添加業務處理handler
        System.out.println("信道初始化中....");
    }
}

4.自定義業務處理類BootNettyHandler

import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.*;
import io.netty.channel.socket.SocketChannel;
import io.netty.handler.timeout.IdleStateEvent;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import java.io.IOException;
import java.net.InetSocketAddress;
import java.text.SimpleDateFormat;
import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.Date;/**
 * I/O數據讀寫處理類
 */
@Component
@ChannelHandler.Sharable
@Slf4j
public class BootNettyHandler extends ChannelInboundHandlerAdapter {
    //  將當前客戶端連接存入map，實現長連接，控制設備下發指令
    public  static Map<String, Channel> ctxMap = new LinkedHashMap<String, Channel>();
    /**
     * 從客戶端收到新的數據時，這個方法會在收到消息時被調用
     *
     * @param ctx
     * @param msg
     */
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) {
        SocketChannel channel = (SocketChannel) ctx.channel();//將字符串轉成每兩個字符加空格形式的字符串 7D 7D 21 3C C5 3B 0D 0D 7D 7D 25 3C
        String regex = "(.{2})";
        String input = msg.toString().replaceAll(regex, "$1 ");
　　　　　//將報文信息記錄到日志中
        log.info(channel.remoteAddress().getHostString() + ":  " + input);
        System.out.println("服務端接受信息為： " + new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(new Date()) + " 接收到消息：" + input);
        //具體的處理業務邏輯，解析報文幀結構，將數據入庫
        ......
    }
    /**
     * 從客戶端收到新的數據、讀取完成時調用
     * @param ctx
     */
    @Override
    public void channelReadComplete(ChannelHandlerContext ctx)
    {
        ctx.flush();
    }
    /**
     * 當出現 Throwable 對象才會被調用，即當 Netty 由於 IO 錯誤或者處理器在處理事件時拋出的異常時
     * @param ctx
     * @param cause
     */
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause)
    {
        System.out.println("程序異常，斷開客戶端連接");
        //獲取客戶端的請求地址  取到的值為客戶端的 ip+端口號
        InetSocketAddress insocket = (InetSocketAddress) ctx.channel().remoteAddress();
        String clientIp = insocket.getAddress().getHostAddress();//設備IP地址（個人將設備的IP地址當作 map 的key，channel對象做value）
        if(ctxMap.get(clientIp)!=null){//如果不為空就剔除
            ctxMap.remove(clientIp, ctx.channel());
        }else{//否則就將當前的設備ip+端口存進map  當做下發設備的標識的key
        }
        cause.printStackTrace();
        ctx.close();//拋出異常，斷開與客戶端的連接
        log.info(clientIp+"連接斷開");
    }
    /**
     * 客戶端與服務端第一次建立連接時執行
     * @param ctx
     * @throws Exception
     */
    @Override
    public void channelActive(ChannelHandlerContext ctx)
    {
        InetSocketAddress insocket = (InetSocketAddress) ctx.channel().remoteAddress();
        String clientIp = insocket.getAddress().getHostAddress();
        if(ctxMap.get(clientIp)!=null){//如果不為空就不存
        }else{//否則就將當前的設備ip+端口存進map  當做下發設備的標識的key
            ctxMap.put(clientIp, ctx.channel());
        }
        log.info(clientIp+"連接");
    }

    /**
     * 客戶端與服務端斷連時執行
     * @param ctx
     * @throws Exception
     */
    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception
    {
        super.channelInactive(ctx);
        InetSocketAddress insocket = (InetSocketAddress) ctx.channel().remoteAddress();
        String clientIp = insocket.getAddress().getHostAddress();
        if(ctxMap.get(clientIp)!=null){//如果不為空就剔除
            ctxMap.remove(clientIp, ctx.channel());
        }else{//否則就將當前的設備ip+端口存進map  當做下發設備的標識的key
        }
        ctx.close(); //斷開連接時，必須關閉，否則造成資源浪費，並發量很大情況下可能造成宕機
        log.info(clientIp+"斷開");
    }
    /**
     * 服務端當read超時, 會調用這個方法
     * @param ctx
     * @param evt
     * @throws Exception
     */
    @Override
    public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception, IOException
    {
        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"));
            log.info(date + " " + ctx.channel().remoteAddress() + " " + eventType);
        }
    }
}

回到開始位置，自定義netty的server端BootNettyServer類該如何啟動？

　　我的方案是在application啟動類中來啟動。

application類實現CommandLineRunner，來啟動nettyserver服務

import org.mybatis.spring.annotation.MapperScan;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.CommandLineRunner;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;

@MapperScan("com.rtstjkx.jkxlistener.repository")
@SpringBootApplication
public class KjxApplication implements CommandLineRunner {

    @Value("${netty.port}")
    private Integer  port;
    @Autowired
    BootNettyServer bootNettyServer;
    public static void main(String[] args) throws Exception {
        SpringApplication.run(KjxApplication.class, args);
    }

    @Override
    public void run(String... args) throws Exception {
        /**
         * 啟動netty服務端服務
         */
        bootNettyServer.run(port);
    }
}

至此，采集程序完成。

　　此程序僅個人學習理解后完成，如有不足的地方，還請進一步多多給出指導，thank you ！！！

代碼git:[e-mail]:white66/jkxLinstener.git