package com.wanshi.netty.simple;

import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
public class NettyServer {
    public static void main(String[] args) throws Exception {
        // 創(chuàng)建BossGroup 和 WorkerGroup
        //說(shuō)明
        //1.創(chuàng)建2個(gè)線(xiàn)程組，分別是boosGroup和workerGroup
        //2.boosGroup只是處理連接請(qǐng)求，真正的與客戶(hù)端業(yè)務(wù)處理，會(huì)交給workerGroup完成
        //3.兩個(gè)都是無(wú)限循環(huán)
        //4. boosGroup 和 workerGroup 含有的子線(xiàn)程（NioEventLoop）的個(gè)數(shù)
        // 默認(rèn)實(shí)際 CPU核數(shù)*2
        EventLoopGroup boosGroup = new NioEventLoopGroup();
        EventLoopGroup workerGroup = new NioEventLoopGroup();
        try {
            //創(chuàng)建服務(wù)器端的啟動(dòng)的對(duì)象，配置參數(shù)
            ServerBootstrap bootstrap = new ServerBootstrap();
            //使用鏈?zhǔn)骄幊虂?lái)進(jìn)行設(shè)置
            bootstrap.group(boosGroup, workerGroup) // 設(shè)置兩個(gè)線(xiàn)程組
                    .channel(NioServerSocketChannel.class) //使用NioServerSocketChannel作為服務(wù)器的通道實(shí)現(xiàn)
                    .option(ChannelOption.SO_BACKLOG, 128) // 設(shè)置線(xiàn)程隊(duì)列等待連接個(gè)數(shù)
                    .childOption(ChannelOption.SO_KEEPALIVE, true) // 設(shè)置保持活動(dòng)連接狀態(tài)
                    .childHandler(new ChannelInitializer<SocketChannel>() { // 創(chuàng)建一個(gè)通道初始化對(duì)象（匿名對(duì)象）
                        //給pipeline 設(shè)置處理器
                        @Override
                        protected void initChannel(SocketChannel socketChannel) throws Exception {
                            //可以使用一個(gè)集合管理SocketChannel，再推送消息時(shí)，可以將業(yè)務(wù)加入到各個(gè)channel對(duì)應(yīng)的NioEventLoop的taskQueue
                            //或者 scheduleTaskQueue
                            System.out.println("客戶(hù) SocketChannel：" + socketChannel.hashCode());
                            socketChannel.pipeline().addLast(new NettyServerHandler());
                        }
                    }); //給我們的workerGroup的某一個(gè)EventLoop的對(duì)應(yīng)的管道設(shè)置處理器
            System.out.println("服務(wù)器 is ready...");
            //綁定一個(gè)端口并且同步，生成了一個(gè)ChannelFuture對(duì)象
            //啟動(dòng)服務(wù)器并綁定端口
            ChannelFuture channelFuture = bootstrap.bind(6668).sync();
            channelFuture.addListener(new ChannelFutureListener() {
                @Override
                public void operationComplete(ChannelFuture future) throws Exception {
                    if (channelFuture.isSuccess()) {
                        System.out.println("監(jiān)聽(tīng)端口 6668 成功");
                    } else {
                        System.out.println("監(jiān)聽(tīng)端口 6668 失敗");
                    }
                }
            });
            //對(duì)關(guān)閉通道進(jìn)行監(jiān)聽(tīng)
            channelFuture.channel().closeFuture().sync();
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            //優(yōu)雅關(guān)閉
            boosGroup.shutdownGracefully();
            workerGroup.shutdownGracefully();
        }
    }
}

package com.wanshi.netty.simple;

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;
import java.util.concurrent.TimeUnit;
/**
 * 自定義一個(gè)Handler，需要繼承netty規(guī)定好的某個(gè)HandlerAdapter
 * 這時(shí)我們自定義的handler才能稱(chēng)為一個(gè)handler
 */
public class NettyServerHandler extends ChannelInboundHandlerAdapter {
    //讀取數(shù)據(jù)事件（這里我們可以讀取客戶(hù)端發(fā)送的消息）
    /**
     * 1.ChannelHandlerContext ctx： 上下文對(duì)象，含有 管道pipeline，通道channel，地址
     * 2.Object msg：就是客戶(hù)端發(fā)送的數(shù)據(jù)，默認(rèn)Object
     * @param ctx
     * @param msg
     * @throws Exception
     */
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        System.out.println("server ctx =" + ctx);
        //將 msg 轉(zhuǎn)成一個(gè)ByteBuf
//        ByteBuf buf = (ByteBuf) msg;
//        System.out.println("客戶(hù)端發(fā)送消息是：" + buf.toString(CharsetUtil.UTF_8));
//        System.out.println("客戶(hù)端地址：" + ctx.channel().remoteAddress());
        //自定義普通任務(wù)隊(duì)列，將耗時(shí)長(zhǎng)的任務(wù)加入隊(duì)列，定義到NioEventLoop --> taskQueue
        ctx.channel().eventLoop().execute(new Runnable() {
            @Override
            public void run() {
                try {
                    Thread.currentThread().sleep(10 * 1000);
                    ctx.writeAndFlush(Unpooled.copiedBuffer("hello，客戶(hù)端：喵2~", CharsetUtil.UTF_8));
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        });
                    Thread.currentThread().sleep(20 * 1000);
                    ctx.writeAndFlush(Unpooled.copiedBuffer("hello，客戶(hù)端：喵3~", CharsetUtil.UTF_8));
        //用戶(hù)自定義定時(shí)任務(wù) --》 該任務(wù)是提交到 scheduleQueue中
        ctx.channel().eventLoop().schedule(new Runnable() {
                    Thread.currentThread().sleep(5 * 1000);
                    ctx.writeAndFlush(Unpooled.copiedBuffer("hello，客戶(hù)端：喵4~", CharsetUtil.UTF_8));
        }, 5, TimeUnit.SECONDS);
        System.out.println("go ~");
    }
     * 數(shù)據(jù)讀取完畢
    public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
        //writeAndFlush 是 write+flush
        //將數(shù)據(jù)寫(xiě)入到緩存，并刷新
        //一般講，需要對(duì)發(fā)送的數(shù)據(jù)進(jìn)行編碼
        ctx.writeAndFlush(Unpooled.copiedBuffer("hello，客戶(hù)端：喵1~", CharsetUtil.UTF_8));
    //處理異常，一般是需要關(guān)閉通道
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        ctx.close();
}

package com.wanshi.netty.simple;

import io.netty.bootstrap.Bootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
public class NettyClient {
    public static void main(String[] args) throws Exception {
        //客戶(hù)端需要一個(gè)事件循環(huán)組
        EventLoopGroup eventExecutors = new NioEventLoopGroup();
        try {
            //創(chuàng)建一個(gè)客戶(hù)端啟動(dòng)對(duì)象
            //客戶(hù)端使用的不是ServerGroup 而是Bootstrap
            Bootstrap bootstrap = new Bootstrap();
            //設(shè)置相關(guān)參數(shù)
            bootstrap.group(eventExecutors) //設(shè)置線(xiàn)程組
                    .channel(NioSocketChannel.class) //設(shè)置客戶(hù)端通道的實(shí)現(xiàn)類(lèi)（反射）
                    .handler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        protected void initChannel(SocketChannel socketChannel) throws Exception {
                            socketChannel.pipeline().addLast(new NettyClientHandler()); //加入自己的處理器
                        }
                    });
            System.out.println("客戶(hù)端 is ok...");
            //啟動(dòng)客戶(hù)端去連接服務(wù)器端， netty異步模型ChannelFuture
            ChannelFuture channelFuture = bootstrap.connect("127.0.0.1", 6668).sync();
            //給關(guān)閉通道進(jìn)行監(jiān)聽(tīng)
            channelFuture.channel().closeFuture().sync();
        } finally {
            //優(yōu)雅關(guān)閉線(xiàn)程池
            eventExecutors.shutdownGracefully();
        }
    }
}

package com.wanshi.netty.simple;

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 NettyClientHandler extends ChannelInboundHandlerAdapter {
    /**
     * 當(dāng)通道就緒就會(huì)觸發(fā)該方法
     * @param ctx
     * @throws Exception
     */
    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        System.out.println("client " + ctx);
        ctx.writeAndFlush(Unpooled.copiedBuffer("hello，服務(wù)端Server：喵~", CharsetUtil.UTF_8));
    }
     * 當(dāng)通道有讀取事件時(shí)，會(huì)觸發(fā)
     * @param msg
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        //將msg轉(zhuǎn)成buf
        ByteBuf buf = (ByteBuf) msg;
        System.out.println("服務(wù)器回復(fù)的消息：" + buf.toString(CharsetUtil.UTF_8));
        System.out.println("服務(wù)器的地址：" + ctx.channel().remoteAddress());
	// 當(dāng)通道發(fā)生異常時(shí)執(zhí)行此方法    
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        cause.printStackTrace();
        ctx.close();
}