网络I/O模型--07Netty基础

package testNetty;

import java.net.InetSocketAddress;
import java.nio.channels.spi.SelectorProvider;
import java.util.concurrent.ThreadFactory;

import io.netty.bootstrap.ServerBootstrap;
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.nio.NioServerSocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.bytes.ByteArrayDecoder;
import io.netty.handler.codec.bytes.ByteArrayEncoder;
import io.netty.util.concurrent.DefaultThreadFactory;

public class TestTCPNettyServer1 {
    public static void main(String[] args) throws Exception {
        // 这就是主要的服务启动器
        ServerBootstrap serverBootstrap = new ServerBootstrap();
        // ＝＝＝＝＝＝＝＝＝＝＝＝＝＝＝＝＝＝＝＝＝＝＝下面设置线程池
        // Boss 线程池
        EventLoopGroup bossLoopGroup = new NioEventLoopGroup(1);
        // Work 线程池：这样的申明方式，主要是为了向读者说明 Netty 的线程组是怎样工作的
        ThreadFactory threadFactory = new DefaultThreadFactory("work thread pool");
        // CPU 个数
        int processorsNumber = Runtime.getRuntime().availableProcessors();
        EventLoopGroup workLoogGroup = new NioEventLoopGroup(processorsNumber, threadFactory,
                SelectorProvider.provider());
        // 指定 Netty 的 Boss 线程和 Work 线程
        serverBootstrap.group(bossLoopGroup, workLoogGroup);
        // 如果是以下的申明方式，则说明 Boss 线程和 Work 线程共享一个线程池
        // serverBootstrap . group(workLoogGroup);
        // ＝＝＝＝＝＝＝＝＝＝＝＝＝＝＝＝＝＝＝＝＝＝＝＝下面我们设置服务的通道类型
        // 只能是实现了 ServerChannel 接口的 “ 服务器”通道类
        serverBootstrap.channel(NioServerSocketChannel.class);
        // 当然也可以这样创建（ SelectorProvider 是不是感觉很熟悉〉
        /*
         * serverBootstrap.channelFactory(new ChannelFactory<NioServerSocketChannel>() {
         * 
         * @Override public NioServerSocketChannel newChannel() { return new
         * NioServerSocketChannel(SelectorProvider.provider());
         * 
         * } });
         */

        // ＝＝＝＝＝＝＝＝＝＝＝＝＝＝＝＝＝＝＝＝＝＝＝＝设置处理器
        // 这里设置了一组简单的 ByteArrayDecoder 和 ByteArrayEncoder
        // Netty 的特色就在这一连串“通道水管 ” 中的“处理器 ”
        serverBootstrap.childHandler(new ChannelInitializer<NioSocketChannel>() {

            @Override
            protected void initChannel(NioSocketChannel ch) throws Exception {
                ch.pipeline().addLast(new ByteArrayEncoder());
                ch.pipeline().addLast(new TcpServerHandler());
                ch.pipeline().addLast(new ByteArrayDecoder());
            }
        });
        // ＝＝＝ ＝＝＝＝＝＝＝＝＝＝＝＝＝＝＝＝＝＝＝＝＝设置 Netty 服务器绑起的 IP 和端口
        serverBootstrap.option(ChannelOption.SO_BACKLOG, 128);
        serverBootstrap.childOption(ChannelOption.SO_KEEPALIVE, true);
        serverBootstrap.bind(new InetSocketAddress("0.0.0.0", 8888));
        // 还可以监控多个端口
        serverBootstrap.bind(new InetSocketAddress("0.0.0.0", 8080));

    }
}

package testNetty;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandler.Sharable;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.util.AttributeKey;

@Sharable
public class TcpServerHandler extends ChannelInboundHandlerAdapter {

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

    // 每一个 Channel ，都有独立的 handler 、 ChannelHandlerContext 、 ChannelPipeline 、 Attribute
    // 所以不需要担心多个 Channel 中的这些对象相互影响
    // 这里我们使用 Content 这个 Key ，记录这个 handler 中已经接收到的客户端信息
    private static AttributeKey<StringBuffer> content = AttributeKey.valueOf("content");

    @Override
    public void channelRegistered(ChannelHandlerContext ctx) throws Exception {
        TcpServerHandler.LOGGER.info("super.channelRegistered(ctx)");
    }

    @Override
    public void channelUnregistered(ChannelHandlerContext ctx) throws Exception {
        TcpServerHandler.LOGGER.info("super.channelUnregistered(ctx)");
    }

    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        TcpServerHandler.LOGGER.info("super.channelActive(ctx) = " + ctx.toString());
    }

    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        TcpServerHandler.LOGGER.info("super.channelinactive(ctx)");
    }

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        TcpServerHandler.LOGGER.info("channelRead(ChannelHandlerContext ctx, Object msg ) ");
        // 我们使用 IDE 工具模拟长连接中的数据缓慢提交
        // 由 read 方法负责接收数据，但只是进行数据累加，不进行任何处理
        ByteBuf byteBuf = (ByteBuf) msg;
        try {
            StringBuffer contextBuffer = new StringBuffer();
            while (byteBuf.isReadable()) {
                contextBuffer.append((char) byteBuf.readByte());
            }
            // 加入临时区域
            StringBuffer content = ctx.attr(TcpServerHandler.content).get();
            if (content == null) {
                content = new StringBuffer();
                ctx.attr(TcpServerHandler.content).set(content);
            }
            content.append(contextBuffer);
        } catch (Exception e) {
            throw e;
        } finally {
            byteBuf.release();
        }
    }

    @Override
    public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
        TcpServerHandler.LOGGER.info("super.channelReadComplete(ChannelHandlerContextctx)");
        // 由 readComplete 方法负责检查数据是否接收完了
        StringBuffer content = ctx.attr(TcpServerHandler.content).get();
        // 如果条件成立，则说明还没有接收到完整的客户端信息
        if (content.indexOf("ver") == -1) {
            return;
        }
        // 当接收到信息后，首先要做的是清空原来的历史信息
        ctx.attr(TcpServerHandler.content).set(new StringBuffer());
        // 准备向客户端发送响应
        ByteBuf byteBuf = ctx.alloc().buffer(1024);
        byteBuf.writeBytes("回发响应信息！".getBytes());
        ctx.writeAndFlush(byteBuf);
        // 正常终止这个通道上下文，就可以关闭通道了
        // 如果不关闭，这个通道的会话将一直存在，
        // 只要网络是稳定的，服务器就可以随时通过这个会话向客户端发送信息
        // 关闭通道意味着 TCP 将正常新开，其中所有的
        // handler 、 ChannelHandlerContext 、 ChannelPipeline 、 Attribute 等信息
        // 都将被注销
        ctx.close();
    }
}

package testNetty;

import java.io.InputStream;
import java.io.OutputStream;
import java.net.Socket;
import java.net.URLEncoder;
import java.util.concurrent.CountDownLatch;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class TestTCPNettyClient implements Runnable {
    private final static Logger LOGGER = LoggerFactory.getLogger(TestTCPNettyClient.class);
    private CountDownLatch countDownLatch;

    private Integer clientindex;

    public TestTCPNettyClient(CountDownLatch countDownLatch, Integer clientindex) {
        this.countDownLatch = countDownLatch;
        this.clientindex = clientindex;
    }

    @Override
    public void run() {
        Socket socket = null;
        OutputStream clientRequest = null;
        InputStream clientResponse = null;
        try {
            socket = new Socket("localhost", 8888);
            clientRequest = socket.getOutputStream();
            clientResponse = socket.getInputStream();
            this.countDownLatch.await();
            clientRequest.write(URLEncoder.encode("第" + this.clientindex + "个客户端请求 11 。", "UTF-8").getBytes());
            clientRequest.flush();
            clientRequest.write(URLEncoder.encode("第" + this.clientindex + "个客户端请求 22 。 over。", "UTF-8").getBytes());
            TestTCPNettyClient.LOGGER.info("第" + this.clientindex + "个客户端请求发送完成 ， 等待服务器返回 ");
            int maxLen = 1024;
            byte[] contextBytes = new byte[maxLen];
            int realLen;
            String message = "";
            while ((realLen = clientResponse.read(contextBytes, 0, maxLen)) != -1) {
                message += new String(contextBytes, 0, realLen);
            }
            TestTCPNettyClient.LOGGER.info("接收到来自服务器的信息：" + message);
        } catch (Exception e) {
            TestTCPNettyClient.LOGGER.error(e.getMessage(), e);
        } finally {
            // 试图关闭连接
            try {
                clientRequest.close();
                clientResponse.close();
                socket.close();
            } catch (Exception e) {
                TestTCPNettyClient.LOGGER.error(e.getMessage(), e);
            }
        }
    }
}