Netty入門使用教程

原創：轉載需注明原創地址 https://www.cnblogs.com/fanerwei222/p/11827026.html

 

本文介紹Netty的使用, 結合我本人的一些理解和操作來快速的讓初學者入門Netty, 理論知識會有, 但是不會太深入, 夠用即可, 僅供入門! 需要想詳細的知識可以移步Netty官網查看官方文檔!

理論知識 : Netty提供異步的、事件驅動的網絡應用程序框架和工具，用以快速開發高性能、高可靠性的網絡服務器和客戶端程序

當然, 我們這里主要是用Netty來發送消息, 接收消息, 測試一下demo, 更厲害的功能后面再慢慢發掘, 我們先看看這玩意怎么玩, 后面再深入

需要工具和Java類:

　　netty-4.1.43

　　netty服務器類　　　　　　SayHelloServer.java

　　netty服務端處理器類　　　SayHelloServerHandler.java

　　netty客戶端類　　　　　　SayHelloClient.java

　　netty客戶端處理器類　　　SayHelloClientHandler.java

　　服務器main方法測試類　　MainNettyServer.java

　　客戶端main方法測試類　　MainNettyClient.java

首先先來一張演示圖, 最下面也會放:

我們看完以下部分就能實現這個東西了!

話不多說, 先貼代碼:

package netty.server;

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 netty.handler.SayHelloServerHandler;

/**
 * sayhello 服務器
 */
public class SayHelloServer {

    /**
     * 端口
     */
    private int port ;

    public SayHelloServer(int port){
        this.port = port;
    }

    public void run() throws Exception{
        /**
         * Netty 負責裝領導的事件處理線程池
         */
        EventLoopGroup leader = new NioEventLoopGroup();
        /**
         * Netty 負責裝碼農的事件處理線程池
         */
        EventLoopGroup coder = new NioEventLoopGroup();
        try {
            /**
             * 服務端啟動引導器
             */
            ServerBootstrap server = new ServerBootstrap();

            server
                    .group(leader, coder)//把事件處理線程池添加進啟動引導器
                    .channel(NioServerSocketChannel.class)//設置通道的建立方式,這里采用Nio的通道方式來建立請求連接
                    .childHandler(new ChannelInitializer<SocketChannel>() {
                        //構造一個由通道處理器構成的通道管道流水線
                        @Override
                        protected void initChannel(SocketChannel socketChannel) throws Exception {
                            /**
                             * 此處添加服務端的通道處理器
                             */
                            socketChannel.pipeline().addLast(new SayHelloServerHandler());
                        }
                    })
                    /**
                     * 用來配置一些channel的參數，配置的參數會被ChannelConfig使用
                     * BACKLOG用於構造服務端套接字ServerSocket對象，
                     * 標識當服務器請求處理線程全滿時，
                     * 用於臨時存放已完成三次握手的請求的隊列的最大長度。
                     * 如果未設置或所設置的值小於1，Java將使用默認值50
                     */
                    .option(ChannelOption.SO_BACKLOG, 128)
                    /**
                     * 是否啟用心跳保活機制。在雙方TCP套接字建立連接后（即都進入ESTABLISHED狀態）
                     * 並且在兩個小時左右上層沒有任何數據傳輸的情況下，這套機制才會被激活。
                     */
                    .childOption(ChannelOption.SO_KEEPALIVE, true);
            /**
             * 服務端綁定端口並且開始接收進來的連接請求
             */
            ChannelFuture channelFuture = server.bind(port).sync();
            /**
             * 查看一下操作是不是成功結束了
             */
            if (channelFuture.isSuccess()){
                //如果沒有成功結束就處理一些事情,結束了就執行關閉服務端等操作
                System.out.println("服務端啟動成功!");
            }
            /**
             * 關閉服務端
             */
            channelFuture.channel().closeFuture().sync();
            System.out.println("服務端即將關閉!");
        } finally {
            /**
             * 關閉事件處理組
             */
            leader.shutdownGracefully();
            coder.shutdownGracefully();
            System.out.println("服務端已關閉!");
        }

    }
}

package netty.handler;

import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.util.CharsetUtil;

/**
 * 服務端入站處理器適配器的繼承類
 * 用來處理服務端的一些事情
 * 根據需要來實現一些方法
 */
public class SayHelloServerHandler extends ChannelInboundHandlerAdapter {

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        ByteBuf buf = (ByteBuf) msg;
        System.out.println("直接打印服務端需要處理的信息: " + buf.toString(CharsetUtil.UTF_8));
        ByteBuf res = Unpooled.wrappedBuffer(new String("塔台收到!塔台收到!信息如下, 請確認 " + buf.toString(CharsetUtil.UTF_8)).getBytes());
        /**
         * 給客戶端回復消息
         */
        ctx.writeAndFlush(res);
    }

    /**
     * 連接成功后，自動執行該方法
     * @param ctx
     * @throws Exception
     */
    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        System.out.println("服務器首次處理!");
        /**
         * 這種發送的消息格式是錯誤的!!!!!
         * 消息格式必須是ByteBuf才行!!!!!
         */
        ctx.writeAndFlush("Hello is server !");
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        super.exceptionCaught(ctx, cause);
        /**
         * 異常捕獲
         */
        cause.printStackTrace();
        ctx.close();
    }
}

package netty.client;

import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import netty.handler.SayHelloClientHandler;

/**
 * sayhello 客戶端
 */
public class SayHelloClient {

    private int port;
    private String host = "127.0.0.1";
    private Channel channel;

    public SayHelloClient(int port){
        this.port = port;
    }

    /**
     * 客戶端運行方法
     * @throws InterruptedException
     */
    public void run() throws InterruptedException {
        /**
         * 負責裝客戶端的事件處理線程池
         */
        EventLoopGroup clientWorker = new NioEventLoopGroup();
        try {
            /**
             * netty客戶端引導啟動器
             */
            Bootstrap bootstrap = new Bootstrap();
            bootstrap
                    .group(clientWorker)//把事件處理線程池添加進啟動引導器
                    .channel(NioSocketChannel.class)//設置通道的建立方式,這里采用Nio的通道方式來建立請求連接
                    //.option(ChannelOption.SO_KEEPALIVE, true)
                    .handler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        protected void initChannel(SocketChannel socketChannel) throws Exception {
                            /**
                             * 此處添加客戶端的通道處理器
                             */
                            socketChannel.pipeline().addLast(new SayHelloClientHandler());
                        }
                    });
            /**
             * 客戶端綁定端口並且開始發起連接請求
             */
            ChannelFuture future = bootstrap.connect(host, port).sync();
            if (future.isSuccess()){
                System.out.println("客戶端連接服務器成功!");
            }
            /**
             * 將通道設置好, 以便外面獲取
             */
            this.channel = future.channel();
            /**
             * 關閉客戶端
             */
            future.channel().closeFuture().sync();
            System.out.println("客戶端即將關閉!");
        } finally {
            /**
             * 關閉事件處理組
             */
            clientWorker.shutdownGracefully();
            System.out.println("客戶端已關閉!");
        }
    }

    public Channel getChannel(){
        return this.channel;
    }
}

package netty.handler;

import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;

import java.nio.charset.Charset;
import java.util.Date;

/**
 * sayhello 客戶端處理器
 */
public class SayHelloClientHandler extends ChannelInboundHandlerAdapter {

    /**
     * 通道信息讀取處理
     * @param ctx
     * @param msg
     */
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) {
        ByteBuf m = (ByteBuf) msg; // 將消息轉化成bytebuf
        try {
            System.out.println("客戶端直接打印接收到的消息: " + m.toString(Charset.defaultCharset()));
            long currentTimeMillis = (m.readUnsignedInt() - 2208988800L) * 1000L;
            System.out.println(new Date(currentTimeMillis));
            /**
             * 給服務端回復消息
             */
            ctx.writeAndFlush("客戶端收到! 消息為: " + m.toString(Charset.defaultCharset()));
        } finally {
            m.release();
        }
    }

    /**
     * 連接成功后，自動執行該方法
     * @param ctx
     * @throws Exception
     */
    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        /**
         * 往服務端發送消息
         * 消息格式必須是ByteBuf才行!!!!!
         * 如果是其他格式服務端是接收不到的!!!!
         */
        String helo = "你好呀!";
        ByteBuf byteBuf = Unpooled.wrappedBuffer(helo.getBytes());
        ctx.channel().writeAndFlush(byteBuf);
        System.out.println("首次連接完成!");
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
        cause.printStackTrace();
        ctx.close();
    }
}

package netty;

import netty.server.SayHelloServer;

/**
 * Netty server 使用main類
 */
public class MainNettyServer {

    /**
     * 端口
     */
    private static int port = 8686;

    public static void main(String[] args) throws Exception {
        /**
         * 啟動netty服務器
         */
        SayHelloServer sayHelloServer = new SayHelloServer(port);
        sayHelloServer.run();
    }
}

package netty;

import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.Channel;
import netty.client.SayHelloClient;

import java.util.Scanner;

/**
 * 客戶端main方法類
 */
public class MainNettyClient {
    public static void main(String[] args) throws InterruptedException {
        /**
         * 創建netty客戶端
         */
        SayHelloClient client = new SayHelloClient(8686);
        /**
         * 新建一個線程讓它單獨去跑,我們可以main方法測試一下發送消息和接受消息
         */
        Thread clientThread = new Thread(new Runnable() {
            @Override
            public void run() {
                try {
                    client.run();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        });
        clientThread.start();
        /**
         * 如果不新建一個線程去跑客戶端的話, 以下的代碼就執行不到
         * 這里用while是因為客戶端的channel並不能立馬生成, 會在client啟動后一段時間才生成獲取到
         * 所以需要延遲一點獲取channel, 否則channel為null
         */
        Channel channel = null;
        boolean isStart = false;
        while (!isStart) {
            if (null != client.getChannel()) {
                channel = client.getChannel();
                isStart = true;
            }
        }
        String helo = "你好呀!我這里是客戶端, 收到請回答";
        ByteBuf byteBuf = Unpooled.wrappedBuffer(helo.getBytes());
        channel.writeAndFlush(byteBuf);
        /**
         * 我們通過控制台輸入來給服務端發送消息
         * 此處只做模擬使用
         */
        for (int i = 0; i < 10 ; i++) {
            Scanner scanner = new Scanner(System.in);
            String text = scanner.nextLine();
            channel.writeAndFlush(Unpooled.wrappedBuffer(text.getBytes()));
        }
    }
}

 接下來我們來使用一下看看, 我們用客戶端控制台輸入消息和服務端"對話": (gif錄制軟件-->ScreenToGif)

 首先我們把項目打包成jar包, 步驟如下:(eclipse的自行百度, 此處用的是IDEA)

 

 

 

 

 

 

 

 保存之后

 

 去輸出的文件目錄找到jar包即可!

 

我放在了一起, 然后開始用CMD運行!

運行命令 : java -jar client.jar

開兩個CMD演示!

左邊是服務器, 右邊是客戶端, 直接看圖!

 

 演示到此結束! over !