sql复制代码LOAD DATA LOCAL INPATH "path" [OVERWRITE] INTO TABLE tablename;

lua复制代码load data local inpath '/home/hdfs/a.txt' into table temp

sql复制代码CREATE EXTERNAL TABLE if not exists xxx_temp(
    id string comment "",
    name int COMMENT ""
)
ROW FORMAT DELIMITED FIELDS TERMINATED BY ','
STORED AS TEXTFILE;

复制代码1,2
3,4
5,6

shell复制代码# hadoop fs -put a.txt /tempData/

sql复制代码hive> load data inpath '/tempData/a.txt' into table xxx_temp;
hive> select * from xxx_temp;

bash复制代码# hadoop fs -ls /warehouse/tablespace/external/hive/xx.db/xxx_temp
-rw-rw-rw-+  3 hdfs hadoop    7023975  /warehouse/tablespace/external/hive/xxx.db/xxx_temp/000000_0
-rw-rw-rw-+  3 hdfs hadoop    7013810  /warehouse/tablespace/external/hive/xxx.db/xxx_temp/000001_0
-rw-rw-rw-+  3 hdfs hadoop    7029668  /warehouse/tablespace/external/hive/xxx.db/xxx_temp/000002_0
-rw-rw-rw-+  3 hdfs hadoop    7021533  /warehouse/tablespace/external/hive/xxx.db/xxx_temp/000003_0
-rw-rw-rw-+  3 hdfs hadoop    7035739  /warehouse/tablespace/external/hive/xxx.db/xxx_temp/000004_0
-rw-rw-rw-+  3 hdfs hadoop    7033479  /warehouse/tablespace/external/hive/xxx.db/xxx_temp/000005_0
# hadoop fs -get /warehouse/tablespace/external/hive/xx.db/xxx_temp/*
# hadoop fs -put /warehouse/tablespace/external/hive/xx.db/xxx_temp_new/*

css复制代码hive> msck repair table xxx_temp_new;

markdown复制代码支付限额规则

1.如果支付金额大于500

  1).如果支付密码正确，支付成功

  2).否则，支付失败

2.如果支付金额不超过500

  1).如果开启了小额免密验证，支付成功

  2).如果没有开启小额免密验证

     1).如果支付密码正确，支付成功

     2).如果支付密码错误，支付失败

python复制代码# 正确密码为123456
# 密码赋值为123456
password='123456'

# 支付金额赋值为265

paymoney=265
# 免密支付默认开启

check=false

# 条件:如果支付金额大于500
if paymoney > 500:
    # 条件:如果输入密码为123456
    if password=='123456':
        print('大金额支付成功')
    # 条件:密码不是123456
    else:
        print('密码错误，大金额支付失败')

# 条件:支付金额小于500
else:
    # 条件:如果开启了免密支付
    if check:
        print('小金额免密支付成功')
    # 条件:如果没有开启免密支付
    else:
        # 条件:如果密码正确
        if password == '123456':
            print('小金额支付成功')
        # 条件:如果密码错误
        else:
            print('密码错误，小金额支付失败')

python复制代码address='北京'
if address=='北京':
    print('北京中心')
elif address=='上海':
    print('上海中心')
elif address=='广州':
    print('广州中心')
else:
    print('地址无效')

python复制代码if address=='北京':
    if district=='昌平区':
        print('昌平区中心')
    elif district=='顺义区':
        print('顺义区中心')
    elif district=='通州区':
        print('通州区中心')
    else:
        print('其他区')

lua复制代码read    从主内丰读取
load    将主内存读取到的值写入工作内存
use        从工作内存读取数据来计算
assign    将计算好的值重新赋值到工作内存中
store    将工作内存数据写入主内存
write    将store过去的变量值赋值给主内存中的变量

ini复制代码LoadLoad屏障：Load1;LoadLoad;Load2,确保Load1数据的装载先于Load2后所有装载指令；

StoreStore屏障：Store1;StoreStore;Store2,确保Store1的数据一定刷回主存，对其他CPU可见，先于Store2以及后续指令；

LoadStore屏障：Load1;LoadStore;Store2,确保Load1指令的数据装载，先于Store2以及后续指令；

StoreLoad屏障：Store1;StoreLoad;Load2,确保Store1指令的数据一定刷回主存，对其他CPU可见，先于Load2以及后续指令的数据装载；

xml复制代码  <br />

arduino复制代码mkdir mysql
tar -xvf mysql-5.7.27-1.e17.x86_64.rpm-bundle.tar -C mysql/

css复制代码cd mysql/
rpm -ivh mysql-community-server-5.7.27-1.e17.x86_64.rmp --force --nodeps

ini复制代码vi /etc/my.cnf
添加如下内容：
[mysqld]
character-set-server=utf8
collation-server=utf8_general_ci

-- 在文件最下方添加
[client]
default-character-set=utf8

sql复制代码service mysqld start

css复制代码mysql -u root -p 敲回车，输入密码
初始密码查看：cat /var/log/mysqld.log
在root@localhost： 后面的就是初始密码

ini复制代码set global validate_password_policy=0;
set global validate_password_length=1;
set password=password('密码');

csharp复制代码// 授权
grant all privileges on *.* to 'root' @'%' identified by '密码';
// 刷新
flush privileges;

arduino复制代码systemctl stop firewalld

复制代码service mysqld restart

arduino复制代码service mysqld stop

kotlin复制代码    "https://baidu.com".httpGet().use {
        println(it)
    }

kotlin复制代码   "https://baidu.com".asyncGet()
       .await()
       .use {
           println(it)
       }

kotlin复制代码const val DEFAULT_CONN_TIMEOUT = 30

val loggingInterceptor by lazy {
    LogManager.logProxy(object : LogProxy {  // 必须要实现 LogProxy ，否则无法打印网络请求的 request 、response
        override fun e(tag: String, msg: String) {
        }

        override fun w(tag: String, msg: String) {
        }

        override fun i(tag: String, msg: String) {
            println("$tag:$msg")
        }

        override fun d(tag: String, msg: String) {
            println("$tag:$msg")
        }
    })

    LoggingInterceptor.Builder()
        .loggable(true) // TODO: 发布到生产环境需要改成false
        .request()
        .requestTag("Request")
        .response()
        .responseTag("Response")
//        .hideVerticalLine()// 隐藏竖线边框
        .build()
}

val httpClient: HttpClient by lazy {
    HttpClientBuilder()
        .baseUrl("http://localhost:8080")
        .allTimeouts(DEFAULT_CONN_TIMEOUT.toLong(), TimeUnit.SECONDS)
        .addInterceptor(loggingInterceptor)
        .addInterceptor(CurlLoggingInterceptor())
        .serializer(GsonSerializer())
        .jsonConverter(GlobalRequestJSONConverter::class)
        .build()
}

kotlin复制代码    httpClient.get{

        url {
            url = "/response-headers-queries"

            "param1" to "value1"
            "param2" to "value2"
        }

        header {
            "key1" to "value1"
            "key2" to "value2"
        }
    }.use {
        println(it)
    }

kotlin复制代码val httpClientWithAOP by lazy {
    HttpClientBuilder()
        .baseUrl("http://localhost:8080")
        .allTimeouts(DEFAULT_CONN_TIMEOUT.toLong(), TimeUnit.SECONDS)
        .addInterceptor(loggingInterceptor)
        .serializer(GsonSerializer())
        .jsonConverter(GlobalRequestJSONConverter::class)
        .addRequestProcessor { _, builder ->
            println("request start")
            builder
        }
        .addResponseProcessor {
            println("response start")
        }
        .build()
}

kotlin复制代码// a request interceptor
typealias RequestProcessor = (HttpClient, Request.Builder) -> Request.Builder

// a response interceptor
typealias ResponseProcessor = (Response) -> Unit

kotlin复制代码    httpClient.get{

        url {
            url = "/response-headers-queries"

            "param1" to "value1"
            "param2" to "value2"
        }

        header {
            "key1" to "value1"
            "key2" to "value2"
        }
    }.use {
        println(it)
    }

kotlin复制代码    httpClient.post{

        url {
            url = "/response-body"
        }

        header {
            "key1" to "value1"
            "key2" to "value2"
        }

        body {
            form {
                "form1" to "value1"
                "form2" to "value2"
            }
        }
    }.use {
        println(it)
    }

kotlin复制代码    httpClient.post{

        url {
            url = "/response-body"
        }

        body("application/json") {
            json {
                "key1" to "value1"
                "key2" to "value2"
                "key3" to "value3"
            }
        }
    }.use {
        println(it)
    }

kotlin复制代码    val file = File("/Users/tony/Downloads/xxx.png")

    httpClient.post{

        url {
            url = "/upload"
        }

        multipartBody {
            +part("file", file.name) {
                file(file)
            }
        }
    }.use {
        println(it)
    }

kotlin复制代码    httpClient.put{

        url {
            url = "/response-body"
        }

        header {
            "key1" to "value1"
            "key2" to "value2"
        }

        body("application/json") {
            string("content")
        }
    }.use {
        println(it)
    }

kotlin复制代码    httpClient.delete{

        url {
            url = "/users/tony"
        }
    }.use {
        println(it)
    }

kotlin复制代码    httpClient.head{

        url {
            url = "/response-headers"
        }

        header {
            "key1" to "value1"
            "key2" to "value2"
            "key3" to "value3"
        }
    }.use {
        println(it)
    }

kotlin复制代码    httpClient.patch{

        url {
            url = "/response-body"
        }

        header {
            "key1" to "value1"
            "key2" to "value2"
        }

        body("application/json") {
            string("content")
        }
    }.use {
        println(it)
    }

kotlin复制代码class ApiService(client: HttpClient) : AbstractHttpService(client) {

    fun testGet(name: String) = get<Response> {
        url = "/sayHi/$name"
    }

    fun testGetWithPath(path: Map<String, String>) = get<Response> {
        url = "/sayHi/{name}"
        pathParams = Params.from(path)
    }

    fun testGetWithHeader(headers: Map<String, String>) = get<Response> {
        url = "/response-headers"
        headersParams = Params.from(headers)
    }

    fun testGetWithHeaderAndQuery(headers: Map<String, String>, queries: Map<String,String>) = get<Response> {
        url = "/response-headers-queries"
        headersParams = Params.from(headers)
        queriesParams = Params.from(queries)
    }

    fun testPost(body: Params) = post<Response> {
        url = "/response-body"
        bodyParams = body
    }

    fun testPostWithModel(model: RequestModel) = post<Response>{
        url = "/response-body"
        bodyModel = model
    }

    fun testPostWithJsonModel(model: RequestModel) = jsonPost<Response>{
        url = "/response-body-with-model"
        jsonModel = model
    }

    fun testPostWithResponseMapper(model: RequestModel) = jsonPost<ResponseData>{
        url = "/response-body-with-model"
        jsonModel = model
        responseMapper = ResponseDataMapper::class
    }
}

kotlin复制代码val apiService by lazy {
    ApiService(httpClient)
}

val requestModel = RequestModel()
apiService.testPostWithModel(requestModel).sync()

kotlin复制代码val apiService by lazy {
    ApiService(httpClient)
}

val requestModel = RequestModel()
apiService.testPostWithModel(requestModel).async()

kotlin复制代码    httpClient.get{

        url {
            url = "/response-headers-queries"

            "param1" to "value1"
            "param2" to "value2"
        }

        header {
            "key1" to "value1"
            "key2" to "value2"
        }
    }.use {
        println(it)
    }

vbnet复制代码curl:
╔══════════════════════════════════════════════════════════════════════════════════════════════════
║ curl -X GET -H "key1: value1" -H "key2: value2" "http://localhost:8080/response-headers-queries?param1=value1&param2=value2"
╚══════════════════════════════════════════════════════════════════════════════════════════════════

kotlin复制代码const val TIME_STAMP = "timestamp"
const val NONCE = "nonce"
const val SIGN = "sign"

private val extraMap:MutableMap<String,String> = mutableMapOf<String,String>().apply {
    this[TIME_STAMP] = System.currentTimeMillis().toString()
    this[NONCE]  = UUID.randomUUID().toString()
}

private val signingInterceptor = SigningInterceptor(SIGN, extraMap, signer = {
    val paramMap = TreeMap<String, String>()
    val url = this.url

    for (name in url.queryParameterNames) {
        val value = url.queryParameterValues(name)[0]?:""
        paramMap[name] = value
    }

    //增加公共参数
    paramMap[TIME_STAMP] = extraMap[TIME_STAMP].toString()
    paramMap[NONCE]  = extraMap[NONCE].toString()

    //所有参数自然排序后拼接
    var paramsStr = join("",paramMap.entries
        .filter { it.key!= SIGN }
        .map { entry -> String.format("%s", entry.value) })

    //生成签名
    sha256HMAC(updateAppSecret,paramsStr)
})

kotlin复制代码interface TraceIdProvider {

    fun getTraceId():String
}

kotlin复制代码interface OAuth2Provider {

    fun getOauthToken():String

    /**
     * 刷新token
     * @return String?
     */
    fun refreshToken(): String?
}

kotlin复制代码interface JWTProvider {

    fun getJWTToken():String

    /**
     * 刷新token
     * @return String?
     */
    fun refreshToken(): String?
}

kotlin复制代码   "https://baidu.com".asyncGet()
       .await()
       .use {
           println(it)
       }

kotlin复制代码        httpClient.asyncGet{

            url{
                url = "/response-headers-queries"

                "param1" to "value1"
                "param2" to "value2"
            }

            header {
                "key1" to "value1"
                "key2" to "value2"
            }
        }.await().use {
            println(it)
        }

kotlin复制代码        httpClient.asyncPost{

            url {
                url = "/response-body"
            }

            header {
                "key1" to "value1"
                "key2" to "value2"
            }

            body("application/json") {
                json {
                    "key1" to "value1"
                    "key2" to "value2"
                    "key3" to "value3"
                }
            }
        }.await().use{
            println(it)
        }

kotlin复制代码        httpClient.flowGet{

            url{
                url = "/response-headers-queries"

                "param1" to "value1"
                "param2" to "value2"
            }

            header {
                "key1" to "value1"
                "key2" to "value2"
            }
        }.collect {
            println(it)
        }

kotlin复制代码        httpClient.flowPost{

            url {
                url = "/response-body"
            }

            header {
                "key1" to "value1"
                "key2" to "value2"
            }

            body("application/json") {
                json {
                    "key1" to "value1"
                    "key2" to "value2"
                    "key3" to "value3"
                }
            }
        }.collect{
            println(it)
        }

kotlin复制代码    // 支持重试的 WebSocket 客户端
    ws = httpClient.websocket("http://127.0.0.1:9876/ws",listener = object : WebSocketListener() {

        override fun onOpen(webSocket: WebSocket, response: Response) {
            logger.info("connection opened...")

            websocket = webSocket

            disposable = Observable.interval(0, 15000,TimeUnit.MILLISECONDS) // 每隔 15 秒发一次业务上的心跳
                .subscribe({
                    heartbeat()
                }, {

                })
        }

        override fun onMessage(webSocket: WebSocket, text: String) {
            logger.info("received instruction: $text")
        }

        override fun onMessage(webSocket: WebSocket, bytes: ByteString) {
        }

        override fun onClosing(webSocket: WebSocket, code: Int, reason: String) {
            logger.info("connection closing: $code, $reason")

            websocket = null

            disposable?.takeIf { !it.isDisposed }?.let {
                it.dispose()
            }
        }

        override fun onClosed(webSocket: WebSocket, code: Int, reason: String) {
            logger.error("connection closed: $code, $reason")

            websocket = null

            disposable?.takeIf { !it.isDisposed }?.let {
                it.dispose()
            }
        }

        override fun onFailure(webSocket: WebSocket, t: Throwable, response: Response?) {
            logger.error("websocket connection error")

            websocket = null

            disposable?.takeIf { !it.isDisposed }?.let {
                it.dispose()
            }
        }
    },wsConfig = WSConfig())

kotlin复制代码    ws?.onConnectStatusChangeListener = {
        logger.info("${it.name}")
        status = it
    }

java复制代码final ChannelFuture initAndRegister() {
    Channel channel = null;
    try {
        channel = channelFactory.newChannel();
        init(channel);
    } catch (Throwable t) {
       
    }
   //注册到boss线程的selector上。
    ChannelFuture regFuture = config().group().register(channel);
    if (regFuture.cause() != null) {
        if (channel.isRegistered()) {
            channel.close();
        } else {
            channel.unsafe().closeForcibly();
        }
    }
    return regFuture;
}

java复制代码@Override
protected void doRegister() throws Exception {
    boolean selected = false;
    for (;;) {
        try {
			//调用ServerSocketChannel的register方法，把当前服务端对象注册到boss线程的selector上
            selectionKey = javaChannel().register(eventLoop().unwrappedSelector(), 0, this);
            return;
        } catch (CancelledKeyException e) {
            if (!selected) {
                // Force the Selector to select now as the "canceled" SelectionKey may still be
                // cached and not removed because no Select.select(..) operation was called yet.
                eventLoop().selectNow();
                selected = true;
            } else {
                // We forced a select operation on the selector before but the SelectionKey is still cached
                // for whatever reason. JDK bug ?
                throw e;
            }
        }
    }
}

java复制代码private static void doBind0(
    final ChannelFuture regFuture, final Channel channel,
    final SocketAddress localAddress, final ChannelPromise promise) {

    //启动
    channel.eventLoop().execute(new Runnable() {
        @Override
        public void run() {
            if (regFuture.isSuccess()) {
                channel.bind(localAddress, promise).addListener(ChannelFutureListener.CLOSE_ON_FAILURE);
            } else {
                promise.setFailure(regFuture.cause());
            }
        }
    });
}

java复制代码private void execute(Runnable task, boolean immediate) {
    boolean inEventLoop = inEventLoop();
    addTask(task);
    if (!inEventLoop) {
        startThread(); //启动线程
        if (isShutdown()) {
            boolean reject = false;
            try {
                if (removeTask(task)) {
                    reject = true;
                }
            } catch (UnsupportedOperationException e) {
                // The task queue does not support removal so the best thing we can do is to just move on and
                // hope we will be able to pick-up the task before its completely terminated.
                // In worst case we will log on termination.
            }
            if (reject) {
                reject();
            }
        }
    }

    if (!addTaskWakesUp && immediate) {
        wakeup(inEventLoop);
    }
}

java复制代码private void startThread() {
    if (state == ST_NOT_STARTED) {
        if (STATE_UPDATER.compareAndSet(this, ST_NOT_STARTED, ST_STARTED)) {
            boolean success = false;
            try {
                doStartThread(); //执行启动过程
                success = true;
            } finally {
                if (!success) {
                    STATE_UPDATER.compareAndSet(this, ST_STARTED, ST_NOT_STARTED);
                }
            }
        }
    }
}

java复制代码private void doStartThread() {
    assert thread == null;
    executor.execute(new Runnable() { //通过线程池执行一个任务
        @Override
        public void run() {
            thread = Thread.currentThread();
            if (interrupted) {
                thread.interrupt();
            }

            boolean success = false;
            updateLastExecutionTime();
            try {
                SingleThreadEventExecutor.this.run(); //调用boss的NioEventLoop的run方法，开启轮询
            }
            //省略....
        }
    });
}

java复制代码protected void run() {
    int selectCnt = 0;
    for (;;) {
        try {
            int strategy;
            try {
                strategy = selectStrategy.calculateStrategy(selectNowSupplier, hasTasks());
                switch (strategy) {
                    case SelectStrategy.CONTINUE:
                        continue;

                    case SelectStrategy.BUSY_WAIT:

                    case SelectStrategy.SELECT:
                        long curDeadlineNanos = nextScheduledTaskDeadlineNanos();
                        if (curDeadlineNanos == -1L) {
                            curDeadlineNanos = NONE; // nothing on the calendar
                        }
                        nextWakeupNanos.set(curDeadlineNanos);
                        try {
                            if (!hasTasks()) {
                                strategy = select(curDeadlineNanos);
                            }
                        } finally {
                            // This update is just to help block unnecessary selector wakeups
                            // so use of lazySet is ok (no race condition)
                            nextWakeupNanos.lazySet(AWAKE);
                        }
                        // fall through
                    default:
                }
            } catch (IOException e) {
                // If we receive an IOException here its because the Selector is messed up. Let's rebuild
                // the selector and retry. https://github.com/netty/netty/issues/8566
                rebuildSelector0();
                selectCnt = 0;
                handleLoopException(e);
                continue;
            }

            selectCnt++;
            cancelledKeys = 0;
            needsToSelectAgain = false;
            final int ioRatio = this.ioRatio;
            boolean ranTasks;
            if (ioRatio == 100) {
                try {
                    if (strategy > 0) {
                        processSelectedKeys();
                    }
                } finally {
                    // Ensure we always run tasks.
                    ranTasks = runAllTasks();
                }
            } else if (strategy > 0) {
                final long ioStartTime = System.nanoTime();
                try {
                    processSelectedKeys();
                } finally {
                    // Ensure we always run tasks.
                    final long ioTime = System.nanoTime() - ioStartTime;
                    ranTasks = runAllTasks(ioTime * (100 - ioRatio) / ioRatio);
                }
            } else {
                ranTasks = runAllTasks(0); // This will run the minimum number of tasks
            }

            if (ranTasks || strategy > 0) {
                if (selectCnt > MIN_PREMATURE_SELECTOR_RETURNS && logger.isDebugEnabled()) {
                    logger.debug("Selector.select() returned prematurely {} times in a row for Selector {}.",
                                 selectCnt - 1, selector);
                }
                selectCnt = 0;
            } else if (unexpectedSelectorWakeup(selectCnt)) { // Unexpected wakeup (unusual case)
                selectCnt = 0;
            }
        } catch (CancelledKeyException e) {
            // Harmless exception - log anyway
            if (logger.isDebugEnabled()) {
                logger.debug(CancelledKeyException.class.getSimpleName() + " raised by a Selector {} - JDK bug?",
                             selector, e);
            }
        } catch (Error e) {
            throw (Error) e;
        } catch (Throwable t) {
            handleLoopException(t);
        } finally {
            // Always handle shutdown even if the loop processing threw an exception.
            try {
                if (isShuttingDown()) {
                    closeAll();
                    if (confirmShutdown()) {
                        return;
                    }
                }
            } catch (Error e) {
                throw (Error) e;
            } catch (Throwable t) {
                handleLoopException(t);
            }
        }
    }
}

java复制代码protected void run() {
        int selectCnt = 0;
        for (;;) {
            try {
                int strategy;
                try {
                    strategy = selectStrategy.calculateStrategy(selectNowSupplier, hasTasks());
                    switch (strategy) {
                    case SelectStrategy.CONTINUE:
                        continue;

                    case SelectStrategy.BUSY_WAIT:
                        // fall-through to SELECT since the busy-wait is not supported with NIO

                    case SelectStrategy.SELECT:
                        long curDeadlineNanos = nextScheduledTaskDeadlineNanos();
                        if (curDeadlineNanos == -1L) {
                            curDeadlineNanos = NONE; // nothing on the calendar
                        }
                        nextWakeupNanos.set(curDeadlineNanos);
                        try {
                            if (!hasTasks()) {
                                strategy = select(curDeadlineNanos);
                            }
                        } finally {
                            // This update is just to help block unnecessary selector wakeups
                            // so use of lazySet is ok (no race condition)
                            nextWakeupNanos.lazySet(AWAKE);
                        }
                        // fall through
                    default:
                    }
                }
                //省略....
            }
        }
}

java复制代码@Override
public int calculateStrategy(IntSupplier selectSupplier, boolean hasTasks) throws Exception {
    return hasTasks ? selectSupplier.get() : SelectStrategy.SELECT;
}

java复制代码private final IntSupplier selectNowSupplier = new IntSupplier() {
    @Override
    public int get() throws Exception {
        return selectNow();
    }
};

java复制代码switch (strategy) {
    case SelectStrategy.CONTINUE:
        continue;

    case SelectStrategy.BUSY_WAIT:
        // fall-through to SELECT since the busy-wait is not supported with NIO

    case SelectStrategy.SELECT:
        long curDeadlineNanos = nextScheduledTaskDeadlineNanos();
        if (curDeadlineNanos == -1L) {
            curDeadlineNanos = NONE; // nothing on the calendar
        }
        nextWakeupNanos.set(curDeadlineNanos);
        try {
            if (!hasTasks()) {
                strategy = select(curDeadlineNanos);
            }
        } finally {
            // This update is just to help block unnecessary selector wakeups
            // so use of lazySet is ok (no race condition)
            nextWakeupNanos.lazySet(AWAKE);
        }
        // fall through
    default:
}

java复制代码//下一次定时任务触发截至时间，默认不是定时任务，返回 -1L
long curDeadlineNanos = nextScheduledTaskDeadlineNanos();
if (curDeadlineNanos == -1L) {
    curDeadlineNanos = NONE; // nothing on the calendar
}
nextWakeupNanos.set(curDeadlineNanos);
try {
    if (!hasTasks()) {
        //2. taskQueue中任务执行完，开始执行select进行阻塞
        strategy = select(curDeadlineNanos);
    }
} finally {
    // This update is just to help block unnecessary selector wakeups
    // so use of lazySet is ok (no race condition)
    nextWakeupNanos.lazySet(AWAKE);
}

java复制代码private int select(long deadlineNanos) throws IOException {
    if (deadlineNanos == NONE) {
        return selector.select();
    }
    //计算select()方法的阻塞超时时间
    long timeoutMillis = deadlineToDelayNanos(deadlineNanos + 995000L) / 1000000L;
    return timeoutMillis <= 0 ? selector.selectNow() : selector.select(timeoutMillis);
}

java复制代码@Override
protected void run() {
    int selectCnt = 0;
    for (;;) {
        //省略....
        selectCnt++;//selectCnt记录的是无功而返的select次数，即eventLoop空转的次数，为解决NIO BUG
        cancelledKeys = 0;
        needsToSelectAgain = false;
        final int ioRatio = this.ioRatio;
        boolean ranTasks;
        if (ioRatio == 100) { //ioRadio执行时间占比是100%，默认是50%
            try {
                if (strategy > 0) { //strategy>0表示存在就绪的SocketChannel
                    processSelectedKeys(); //执行就绪SocketChannel的任务
                }
            } finally {
             //注意，将ioRatio设置为100，并不代表任务不执行，反而是每次将任务队列执行完
                ranTasks = runAllTasks(); //确保总是执行队列中的任务
            }
        } else if (strategy > 0) { //strategy>0表示存在就绪的SocketChannel
            final long ioStartTime = System.nanoTime(); //io时间处理开始时间
            try {
                processSelectedKeys(); //开始处理IO就绪事件
            } finally {
                // io事件执行结束时间
                final long ioTime = System.nanoTime() - ioStartTime;
                //基于本次循环处理IO的时间，ioRatio，计算出执行任务耗时的上限，也就是只允许处理多长时间异步任务
                ranTasks = runAllTasks(ioTime * (100 - ioRatio) / ioRatio);
            }
        } else {
            //这个分支代表：strategy=0，ioRatio<100，此时任务限时=0，意为：尽量少地执行异步任务
            //这个分支和strategy>0实际是一码事，代码简化了一下而已
            ranTasks = runAllTasks(0); // This will run the minimum number of tasks
        }

        if (ranTasks || strategy > 0) { //ranTasks=true，或strategy>0，说明eventLoop干活了，没有空转，清空selectCnt
            if (selectCnt > MIN_PREMATURE_SELECTOR_RETURNS && logger.isDebugEnabled()) {
                logger.debug("Selector.select() returned prematurely {} times in a row for Selector {}.",
                             selectCnt - 1, selector);
            }
            selectCnt = 0;
        } 
         //unexpectedSelectorWakeup处理NIO BUG
        else if (unexpectedSelectorWakeup(selectCnt)) { // Unexpected wakeup (unusual case)
            selectCnt = 0;
        }
    }
}

java复制代码private void processSelectedKeys() {
    if (selectedKeys != null) {
        processSelectedKeysOptimized();
    } else {
        processSelectedKeysPlain(selector.selectedKeys());
    }
}

java复制代码private void processSelectedKeysOptimized() {
    for (int i = 0; i < selectedKeys.size; ++i) {
        //1. 取出IO事件以及对应的channel
        final SelectionKey k = selectedKeys.keys[i];
        selectedKeys.keys[i] = null;//k的引用置null，便于gc回收，也表示该channel的事件处理完成避免重复处理

        final Object a = k.attachment(); //获取保存在当前channel中的attachment，此时应该是NioServerSocketChannel
		//处理当前的channel
        if (a instanceof AbstractNioChannel) {
             //对于boss NioEventLoop，轮询到的基本是连接事件，后续的事情就是通过他的pipeline将连接扔给一个worker NioEventLoop处理
            //对于worker NioEventLoop来说，轮循道的基本商是IO读写事件，后续的事情就是通过他的pipeline将读取到的字节流传递给每个channelHandler来处理
            processSelectedKey(k, (AbstractNioChannel) a);
        } else {
            @SuppressWarnings("unchecked")
            NioTask<SelectableChannel> task = (NioTask<SelectableChannel>) a;
            processSelectedKey(k, task);
        }
		
        if (needsToSelectAgain) {
            // null out entries in the array to allow to have it GC'ed once the Channel close
            // See https://github.com/netty/netty/issues/2363
            selectedKeys.reset(i + 1);

            selectAgain();
            i = -1;
        }
    }
}

java复制代码private void processSelectedKey(SelectionKey k, AbstractNioChannel ch) {
    final AbstractNioChannel.NioUnsafe unsafe = ch.unsafe();
    if (!k.isValid()) {
        final EventLoop eventLoop;
        try {
            eventLoop = ch.eventLoop();
        } catch (Throwable ignored) {
           
        }
        if (eventLoop == this) {
            // close the channel if the key is not valid anymore
            unsafe.close(unsafe.voidPromise());
        }
        return;
    }

    try {
        int readyOps = k.readyOps(); //获取当前key所属的操作类型
      
        if ((readyOps & SelectionKey.OP_CONNECT) != 0) {//如果是连接类型
            int ops = k.interestOps();
            ops &= ~SelectionKey.OP_CONNECT;
            k.interestOps(ops);

            unsafe.finishConnect();
        }
        if ((readyOps & SelectionKey.OP_WRITE) != 0) { //如果是写类型
            ch.unsafe().forceFlush();
        }
		//如果是读类型或者ACCEPT类型。则执行unsafe.read()方法，unsafe的实例对象为 NioMessageUnsafe
        if ((readyOps & (SelectionKey.OP_READ | SelectionKey.OP_ACCEPT)) != 0 || readyOps == 0) {
            unsafe.read();
        }
    } catch (CancelledKeyException ignored) {
        unsafe.close(unsafe.voidPromise());
    }
}

java复制代码@Override
public void read() {
    assert eventLoop().inEventLoop();
    final ChannelConfig config = config();
    final ChannelPipeline pipeline = pipeline(); //如果是第一次建立连接，此时的pipeline是ServerBootstrapAcceptor
    final RecvByteBufAllocator.Handle allocHandle = unsafe().recvBufAllocHandle();
    allocHandle.reset(config);

    boolean closed = false;
    Throwable exception = null;
    try {
        try {
            do {
                int localRead = doReadMessages(readBuf);
                if (localRead == 0) {
                    break;
                }
                if (localRead < 0) {
                    closed = true;
                    break;
                }

                allocHandle.incMessagesRead(localRead);
            } while (continueReading(allocHandle));
        } catch (Throwable t) {
            exception = t;
        }

        int size = readBuf.size();
        for (int i = 0; i < size; i ++) {
            readPending = false;
            pipeline.fireChannelRead(readBuf.get(i));  //调用pipeline中的channelRead方法
        }
        readBuf.clear();
        allocHandle.readComplete();
        pipeline.fireChannelReadComplete();

        if (exception != null) {
            closed = closeOnReadError(exception);

            pipeline.fireExceptionCaught(exception); //调用pipeline中的ExceptionCaught方法
        }

        if (closed) {
            inputShutdown = true;
            if (isOpen()) {
                close(voidPromise());
            }
        }
    } finally {
        if (!readPending && !config.isAutoRead()) {
            removeReadOp();
        }
    }
}

java复制代码final class SelectedSelectionKeySet extends AbstractSet<SelectionKey> {

    SelectionKey[] keys;
    int size;

    SelectedSelectionKeySet() {
        keys = new SelectionKey[1024];
    }

    @Override
    public boolean add(SelectionKey o) {
        if (o == null) {
            return false;
        }

        keys[size++] = o;
        if (size == keys.length) {
            increaseCapacity();
        }

        return true;
    }
}

java复制代码private SelectorTuple openSelector() {
    final Class<?> selectorImplClass = (Class<?>) maybeSelectorImplClass;
    final SelectedSelectionKeySet selectedKeySet = new SelectedSelectionKeySet();
    //使用反射
    Object maybeException = AccessController.doPrivileged(new PrivilegedAction<Object>() {
        @Override
        public Object run() {
            try {
                //Selector内部的selectedKeys字段
                Field selectedKeysField = selectorImplClass.getDeclaredField("selectedKeys");
                //Selector内部的publicSelectedKeys字段
                Field publicSelectedKeysField = selectorImplClass.getDeclaredField("publicSelectedKeys");

                if (PlatformDependent.javaVersion() >= 9 && PlatformDependent.hasUnsafe()) {
                    //获取selectedKeysField字段偏移量
                    long selectedKeysFieldOffset = PlatformDependent.objectFieldOffset(selectedKeysField);
                    //获取publicSelectedKeysField字段偏移量
                    long publicSelectedKeysFieldOffset =
                        PlatformDependent.objectFieldOffset(publicSelectedKeysField);

                    if (selectedKeysFieldOffset != -1 && publicSelectedKeysFieldOffset != -1) {
                        //替换为selectedKeySet
                        PlatformDependent.putObject(
                            unwrappedSelector, selectedKeysFieldOffset, selectedKeySet);
                        PlatformDependent.putObject(
                            unwrappedSelector, publicSelectedKeysFieldOffset, selectedKeySet);
                        return null;
                    }
                    // We could not retrieve the offset, lets try reflection as last-resort.
                }
                Throwable cause = ReflectionUtil.trySetAccessible(selectedKeysField, true);
                if (cause != null) {
                    return cause;
                }
                cause = ReflectionUtil.trySetAccessible(publicSelectedKeysField, true);
                if (cause != null) {
                    return cause;
                }
                selectedKeysField.set(unwrappedSelector, selectedKeySet);
                publicSelectedKeysField.set(unwrappedSelector, selectedKeySet);
                return null;
            } catch (NoSuchFieldException e) {
                return e;
            } catch (IllegalAccessException e) {
                return e;
            }
        }
    });
    if (maybeException instanceof Exception) {
        selectedKeys = null;
        Exception e = (Exception) maybeException;
        logger.trace("failed to instrument a special java.util.Set into: {}", unwrappedSelector, e);
        return new SelectorTuple(unwrappedSelector);
    }
    selectedKeys = selectedKeySet;
}

java复制代码@Override
protected void run() {
    int selectCnt = 0;
    for (;;) {
        ranTasks = runAllTasks();
    }
}

java复制代码protected boolean runAllTasks() {
    assert inEventLoop();
    boolean fetchedAll;
    boolean ranAtLeastOne = false;

    do {
        fetchedAll = fetchFromScheduledTaskQueue(); //合并定时任务到普通任务队列
        if (runAllTasksFrom(taskQueue)) { //循环执行taskQueue中的任务
            ranAtLeastOne = true;
        }
    } while (!fetchedAll);  

    if (ranAtLeastOne) { //如果任务全部执行完成，记录执行完完成时间
        lastExecutionTime = ScheduledFutureTask.nanoTime();
    }
    afterRunningAllTasks();//执行收尾任务
    return ranAtLeastOne;
}

java复制代码private boolean fetchFromScheduledTaskQueue() {
    if (scheduledTaskQueue == null || scheduledTaskQueue.isEmpty()) {
        return true;
    }
    long nanoTime = AbstractScheduledEventExecutor.nanoTime();
    for (;;) {
        Runnable scheduledTask = pollScheduledTask(nanoTime);
        if (scheduledTask == null) {
            return true;
        }
        if (!taskQueue.offer(scheduledTask)) {
            // No space left in the task queue add it back to the scheduledTaskQueue so we pick it up again.
            scheduledTaskQueue.add((ScheduledFutureTask<?>) scheduledTask);
            return false;
        }
    }
}

java复制代码private void execute(Runnable task, boolean immediate) {
    boolean inEventLoop = inEventLoop();
    addTask(task); //把当前任务添加到阻塞队列中
    if (!inEventLoop) { //如果是非NioEventLoop
        startThread(); //启动线程
        if (isShutdown()) { //如果当前NioEventLoop已经是停止状态
            boolean reject = false;
            try {
                if (removeTask(task)) { 
                    reject = true;
                }
            } catch (UnsupportedOperationException e) {
                // The task queue does not support removal so the best thing we can do is to just move on and
                // hope we will be able to pick-up the task before its completely terminated.
                // In worst case we will log on termination.
            }
            if (reject) {
                reject();
            }
        }
    }

    if (!addTaskWakesUp && immediate) {
        wakeup(inEventLoop);
    }
}

java复制代码@Override
protected void run() {
    int selectCnt = 0;
    for (;;) {
        //selectCnt记录的是无功而返的select次数，即eventLoop空转的次数，为解决NIO BUG
        selectCnt++; 
        //ranTasks=true，或strategy>0，说明eventLoop干活了，没有空转，清空selectCnt
        if (ranTasks || strategy > 0) {
            //如果选择操作计数器的值，大于最小选择器重构阈值，则输出log
            if (selectCnt > MIN_PREMATURE_SELECTOR_RETURNS && logger.isDebugEnabled()) {
                logger.debug("Selector.select() returned prematurely {} times in a row for Selector {}.",
                             selectCnt - 1, selector);
            }
            selectCnt = 0;
        } 
        //unexpectedSelectorWakeup处理NIO BUG
        else if (unexpectedSelectorWakeup(selectCnt)) { // Unexpected wakeup (unusual case)
            selectCnt = 0;
        }
    }
}

java复制代码private boolean unexpectedSelectorWakeup(int selectCnt) {
    if (Thread.interrupted()) {
        if (logger.isDebugEnabled()) {
            logger.debug("Selector.select() returned prematurely because " +
                         "Thread.currentThread().interrupt() was called. Use " +
                         "NioEventLoop.shutdownGracefully() to shutdown the NioEventLoop.");
        }
        return true;
    }
    //如果选择重构的阈值大于0， 默认值是512次、 并且当前触发的空轮询次数大于 512次。，则触发重构
    if (SELECTOR_AUTO_REBUILD_THRESHOLD > 0 &&
        selectCnt >= SELECTOR_AUTO_REBUILD_THRESHOLD) {
        // The selector returned prematurely many times in a row.
        // Rebuild the selector to work around the problem.
        logger.warn("Selector.select() returned prematurely {} times in a row; rebuilding Selector {}.",
                    selectCnt, selector);
        rebuildSelector();
        return true;
    }
    return false;
}

java复制代码public void rebuildSelector() {
    if (!inEventLoop()) { //如果不是在eventLoop中执行，则使用异步线程执行
        execute(new Runnable() {
            @Override
            public void run() {
                rebuildSelector0();
            }
        });
        return;
    }
    rebuildSelector0();
}

java复制代码private void rebuildSelector0() {
    final Selector oldSelector = selector; //获取老的selector选择器
    final SelectorTuple newSelectorTuple; //定义新的选择器

    if (oldSelector == null) { //如果老的选择器为空，直接返回
        return;
    }

    try {
        newSelectorTuple = openSelector(); //创建一个新的选择器
    } catch (Exception e) {
        logger.warn("Failed to create a new Selector.", e);
        return;
    }

    // Register all channels to the new Selector.
    int nChannels = 0;
    for (SelectionKey key: oldSelector.keys()) {//遍历注册到选择器的选择key集合
        Object a = key.attachment();
        try {
             //如果选择key无效或选择关联的通道已经注册到新的选择器，则跳出当前循环
            if (!key.isValid() || key.channel().keyFor(newSelectorTuple.unwrappedSelector) != null) {
                continue;
            }
 			//获取key的选择关注事件集
            int interestOps = key.interestOps();
            key.cancel();//取消选择key
	      //注册选择key到新的选择器
            SelectionKey newKey = key.channel().register(newSelectorTuple.unwrappedSelector, interestOps, a);
            if (a instanceof AbstractNioChannel) {//如果是nio通道，则更新通道的选择key
                // Update SelectionKey
                ((AbstractNioChannel) a).selectionKey = newKey;
            }
            nChannels ++;
        } catch (Exception e) {
            logger.warn("Failed to re-register a Channel to the new Selector.", e);
            if (a instanceof AbstractNioChannel) {
                AbstractNioChannel ch = (AbstractNioChannel) a;
                ch.unsafe().close(ch.unsafe().voidPromise());
            } else {
                @SuppressWarnings("unchecked")
                NioTask<SelectableChannel> task = (NioTask<SelectableChannel>) a;
                invokeChannelUnregistered(task, key, e);
            }
        }
    }
	//更新当前事件循环选择器
    selector = newSelectorTuple.selector;
    unwrappedSelector = newSelectorTuple.unwrappedSelector;

    try {
        // time to close the old selector as everything else is registered to the new one
        oldSelector.close(); //关闭原始选择器
    } catch (Throwable t) {
        if (logger.isWarnEnabled()) {
            logger.warn("Failed to close the old Selector.", t);
        }
    }

    if (logger.isInfoEnabled()) {
        logger.info("Migrated " + nChannels + " channel(s) to the new Selector.");
    }
}

java复制代码public void read() {
    assert eventLoop().inEventLoop();
    final ChannelConfig config = config();
    final ChannelPipeline pipeline = pipeline();
    final RecvByteBufAllocator.Handle allocHandle = unsafe().recvBufAllocHandle();
    allocHandle.reset(config);

    boolean closed = false;
    Throwable exception = null;
    try {
        try {
            do {
                //如果有客户端连接进来，则localRead为1，否则返回0
                int localRead = doReadMessages(readBuf);
                if (localRead == 0) {
                    break;
                }
                if (localRead < 0) {
                    closed = true;
                    break;
                }
				
                allocHandle.incMessagesRead(localRead); //累计增加read消息数量
            } while (continueReading(allocHandle));
        } catch (Throwable t) {
            exception = t;
        }

        int size = readBuf.size(); //遍历客户端连接列表
        for (int i = 0; i < size; i ++) {
            readPending = false;
            pipeline.fireChannelRead(readBuf.get(i)); //调用pipeline中handler的channelRead方法。
        }
        readBuf.clear(); //清空集合
        allocHandle.readComplete();
        pipeline.fireChannelReadComplete(); //触发pipeline中handler的readComplete方法

        if (exception != null) {
            closed = closeOnReadError(exception);

            pipeline.fireExceptionCaught(exception);
        }

        if (closed) {
            inputShutdown = true;
            if (isOpen()) {
                close(voidPromise());
            }
        }
    } finally {
        if (!readPending && !config.isAutoRead()) {
            removeReadOp();
        }
    }
}

java复制代码static void invokeChannelRead(final AbstractChannelHandlerContext next, Object msg) {
    final Object m = next.pipeline.touch(ObjectUtil.checkNotNull(msg, "msg"), next);
    EventExecutor executor = next.executor();
    if (executor.inEventLoop()) {
        next.invokeChannelRead(m); //获取pipeline中的下一个节点，调用该handler的channelRead方法
    } else {
        executor.execute(new Runnable() {
            @Override
            public void run() {
                next.invokeChannelRead(m);
            }
        });
    }
}

java复制代码public void channelRead(ChannelHandlerContext ctx, Object msg) {
    final Channel child = (Channel) msg;

    child.pipeline().addLast(childHandler);  //把服务端配置的childHandler，添加到当前NioSocketChannel中的pipeline中

    setChannelOptions(child, childOptions, logger); //设置NioSocketChannel的属性
    setAttributes(child, childAttrs); 

    try {
        //把当前的NioSocketChannel注册到Selector上，并且监听一个异步事件。
        childGroup.register(child).addListener(new ChannelFutureListener() {
            @Override
            public void operationComplete(ChannelFuture future) throws Exception {
                if (!future.isSuccess()) {
                    forceClose(child, future.cause());
                }
            }
        });
    } catch (Throwable t) {
        forceClose(child, t);
    }
}

java复制代码@Override
protected int doReadMessages(List<Object> buf) throws Exception {
    SocketChannel ch = SocketUtils.accept(javaChannel());

    try {
        if (ch != null) {
            buf.add(new NioSocketChannel(this, ch)); //这里
            return 1;
        }
    } catch (Throwable t) {
        logger.warn("Failed to create a new channel from an accepted socket.", t);

        try {
            ch.close();
        } catch (Throwable t2) {
            logger.warn("Failed to close a socket.", t2);
        }
    }

    return 0;
}

java复制代码protected AbstractChannel(Channel parent) {
    this.parent = parent;
    id = newId();
    unsafe = newUnsafe();
    pipeline = newChannelPipeline();
}

java复制代码protected DefaultChannelPipeline(Channel channel) {
    this.channel = ObjectUtil.checkNotNull(channel, "channel");
    succeededFuture = new SucceededChannelFuture(channel, null);
    voidPromise =  new VoidChannelPromise(channel, true);

    tail = new TailContext(this);
    head = new HeadContext(this);

    head.next = tail;
    tail.prev = head;
}

java复制代码@Override
public final ChannelPipeline addLast(EventExecutorGroup executor, ChannelHandler... handlers) {
   ObjectUtil.checkNotNull(handlers, "handlers");

   for (ChannelHandler h: handlers) { //遍历handlers列表，此时这里的handler是ChannelInitializer回调方法
       if (h == null) {
           break;
       }
       addLast(executor, null, h);
   }

   return this;
}

java复制代码@Override
public final ChannelPipeline addLast(EventExecutorGroup group, String name, ChannelHandler handler) {
    final AbstractChannelHandlerContext newCtx;
    synchronized (this) {
        checkMultiplicity(handler); //检查是否有重复的handler
		//创建新的DefaultChannelHandlerContext节点
        newCtx = newContext(group, filterName(name, handler), handler);

        addLast0(newCtx);  //添加新的DefaultChannelHandlerContext到ChannelPipeline

      
        if (!registered) { 
            newCtx.setAddPending();
            callHandlerCallbackLater(newCtx, true);
            return this;
        }

        EventExecutor executor = newCtx.executor();
        if (!executor.inEventLoop()) {
            callHandlerAddedInEventLoop(newCtx, executor);
            return this;
        }
    }
    callHandlerAdded0(newCtx);
    return this;
}

java复制代码childGroup.register(child).addListener(new ChannelFutureListener() {}