网络编程框架t-io的编程基本知识介绍

t-io作为目前国内最流行的开源网络编程框架软件，以简单易懂，上手容易而著称，相同的功能比起netty实现起来，要简单的多，代码量也大大减少，如果要使用好t-io，还是要先学习t-io的一些基本知识，这篇文章主要从8个方面介绍了t-io的基础知识。具体请参考： https://www.wanetech.com/doc/tio/88

t-io收发消息过程

t-io收发消息及处理过程，可以用一张图清晰地表达出来

应用层包：Packet

Packet是用于表述业务数据结构的，我们通过继承Packet来实现自己的业务数据结构，对于各位而言，把Packet看作是一个普通的VO对象即可。

注意：不建议直接使用Packet对象，而是要继承Packet

一个简单的Packet可能长这样

1package org.tio.study.helloworld.common;
2import org.tio.core.intf.Packet;
3/**
4 * @author tanyaowu
5 */
6public class HelloPacket extends Packet {
7    private static final long serialVersionUID = -172060606924066412L;
8    public static final int HEADER_LENGTH = 4;//消息头的长度
9    public static final String CHARSET = "utf-8";
10    private byte[] body;
11    /**
12     * @return the body
13     */
14    public byte[] getBody() {
15        return body;
16    }
17    /**
18     * @param body the body to set
19     */
20    public void setBody(byte[] body) {
21        this.body = body;
22    }
23}

可以结合AioHandler.java理解Packet

1import java.nio.ByteBuffer;
2import org.tio.core.ChannelContext;
3import org.tio.core.TioConfig;
4import org.tio.core.exception.AioDecodeException;
5/**
6 * 
7 * @author tanyaowu 
8 * 2017年10月19日 上午9:40:15
9 */
10public interface AioHandler {
11    /**
12     * 根据ByteBuffer解码成业务需要的Packet对象.
13     * 如果收到的数据不全，导致解码失败，请返回null，在下次消息来时框架层会自动续上前面的收到的数据
14     * @param buffer 参与本次希望解码的ByteBuffer
15     * @param limit ByteBuffer的limit
16     * @param position ByteBuffer的position，不一定是0哦
17     * @param readableLength ByteBuffer参与本次解码的有效数据（= limit - position）
18     * @param channelContext
19     * @return
20     * @throws AioDecodeException
21     */
22    Packet decode(ByteBuffer buffer, int limit, int position, int readableLength, ChannelContext channelContext) throws AioDecodeException;
23    /**
24     * 编码
25     * @param packet
26     * @param tioConfig
27     * @param channelContext
28     * @return
29     * @author: tanyaowu
30     */
31    ByteBuffer encode(Packet packet, TioConfig tioConfig, ChannelContext channelContext);
32    /**
33     * 处理消息包
34     * @param packet
35     * @param channelContext
36     * @throws Exception
37     * @author: tanyaowu
38     */
39    void handler(Packet packet, ChannelContext channelContext) throws Exception;
40}

单条TCP连接上下文：ChannelContext

每一个tcp连接的建立都会产生一个ChannelContext对象，这是个抽象类，如果你是用t-io作tcp客户端，那么就是ClientChannelContext，如果你是用tio作tcp服务器，那么就是ServerChannelContext 用户可以把业务数据通过ChannelContext对象和TCP连接关联起来，像下面这样设置属性

ChannelContext.set(String key, Object value)

然后用下面的方式获取属性

 ChannelContext.get(String key)

当然最最常用的还是用t-io提供的强到没对手的bind功能，譬如用下面的代码绑定userid

Tio.bindUser(ChannelContext channelContext, String userid)

然后可以通过userid进行操作，示范代码如下

1//获取某用户的ChannelContext集合
2SetWithLock<ChannelContext> set = Tio.getChannelContextsByUserid(tioConfig, userid);
3//给某用户发消息
4Tio.sendToUser(TioConfig, userid, Packet)

除了可以绑定userid，t-io还内置了如下绑定API

无序列表绑定业务id

Tio.bindBsId(ChannelContext channelContext, String bsId)

绑定token

Tio.bindToken(ChannelContext channelContext, String token)

绑定群组

Tio.bindGroup(ChannelContext channelContext, String group)

ChannelContext对象包含的信息非常多，主要对象见下图说明 ChannelContext是t-io中非常重要的类，他是业务和连接的沟通桥梁！

服务配置与维护：TioConfig

场景：我们在写TCP Server时，都会先选好一个端口以监听客户端连接，再创建N组线程池来执行相关的任务，譬如发送消息、解码数据包、处理数据包等任务，还要维护客户端连接的各种数据，为了和业务互动，还要把这些客户端连接和各种业务数据绑定起来，譬如把某个客户端绑定到一个群组，绑定到一个userid，绑定到一个token等。 TioConfig就是解决以上场景的：配置线程池、监听端口，维护客户端各种数据等的。

TioConfig是个抽象类

如果你是用tio作tcp客户端，那么你需要创建ClientTioConfig对象服务器端对应一个ClientTioConfig对象如果你是用tio作tcp服务器，那么你需要创建ServerTioConfig 一个监听端口对应一个ServerTioConfig ，一个jvm可以监听多个端口，所以一个jvm可以有多个ServerTioConfig对象 TioConfig对象包含的信息非常多，主要对象见下图如何获取TioConfig对象见：https://www.wanetech.com/doc/tio/253?pageNumber=1

编码、解码、处理：AioHandler

AioHandler是处理消息的核心接口，它有两个子接口，ClientAioHandler和ServerAioHandler，当用tio作tcp客户端时需要实现ClientAioHandler，当用tio作tcp服务器时需要实现ServerAioHandler，它主要定义了3个方法，见下

1import java.nio.ByteBuffer;
2import org.tio.core.ChannelContext;
3import org.tio.core.TioConfig;
4import org.tio.core.exception.AioDecodeException;
5/**
6 * 
7 * @author tanyaowu 
8 * 2017年10月19日 上午9:40:15
9 */
10public interface AioHandler {
11    /**
12     * 根据ByteBuffer解码成业务需要的Packet对象.
13     * 如果收到的数据不全，导致解码失败，请返回null，在下次消息来时框架层会自动续上前面的收到的数据
14     * @param buffer 参与本次希望解码的ByteBuffer
15     * @param limit ByteBuffer的limit
16     * @param position ByteBuffer的position，不一定是0哦
17     * @param readableLength ByteBuffer参与本次解码的有效数据（= limit - position）
18     * @param channelContext
19     * @return
20     * @throws AioDecodeException
21     */
22    Packet decode(ByteBuffer buffer, int limit, int position, int readableLength, ChannelContext channelContext) throws AioDecodeException;
23    /**
24     * 编码
25     * @param packet
26     * @param tioConfig
27     * @param channelContext
28     * @return
29     * @author: tanyaowu
30     */
31    ByteBuffer encode(Packet packet, TioConfig tioConfig, ChannelContext channelContext);
32    /**
33     * 处理消息包
34     * @param packet
35     * @param channelContext
36     * @throws Exception
37     * @author: tanyaowu
38     */
39    void handler(Packet packet, ChannelContext channelContext) throws Exception;
40}

消息来往监听：AioListener

AioListener是处理消息的核心接口，它有两个子接口：ClientAioListener和ServerAioListener

当用tio作tcp客户端时需要实现ClientAioListener 当用tio作tcp服务器时需要实现ServerAioListener 它主要定义了如下方法

1package org.tio.core.intf;
2import org.tio.core.ChannelContext;
3/**
4 *
5 * @author tanyaowu
6 * 2017年4月1日 上午9:34:08
7 */
8public interface AioListener {
9    /**
10     * 建链后触发本方法，注：建链不一定成功，需要关注参数isConnected
11     * @param channelContext
12     * @param isConnected 是否连接成功,true:表示连接成功，false:表示连接失败
13     * @param isReconnect 是否是重连, true: 表示这是重新连接，false: 表示这是第一次连接
14     * @throws Exception
15     * @author: tanyaowu
16     */
17    public void onAfterConnected(ChannelContext channelContext, boolean isConnected, boolean isReconnect) throws Exception;
18    /**
19     * 原方法名：onAfterDecoded
20     * 解码成功后触发本方法
21     * @param channelContext
22     * @param packet
23     * @param packetSize
24     * @throws Exception
25     * @author: tanyaowu
26     */
27    public void onAfterDecoded(ChannelContext channelContext, Packet packet, int packetSize) throws Exception;
28    /**
29     * 接收到TCP层传过来的数据后
30     * @param channelContext
31     * @param receivedBytes 本次接收了多少字节
32     * @throws Exception
33     */
34    public void onAfterReceivedBytes(ChannelContext channelContext, int receivedBytes) throws Exception;
35    /**
36     * 消息包发送之后触发本方法
37     * @param channelContext
38     * @param packet
39     * @param isSentSuccess true:发送成功，false:发送失败
40     * @throws Exception
41     * @author tanyaowu
42     */
43    public void onAfterSent(ChannelContext channelContext, Packet packet, boolean isSentSuccess) throws Exception;
44    /**
45     * 处理一个消息包后
46     * @param channelContext
47     * @param packet
48     * @param cost 本次处理消息耗时，单位：毫秒
49     * @throws Exception
50     */
51    public void onAfterHandled(ChannelContext channelContext, Packet packet, long cost) throws Exception;
52    /**
53     * 连接关闭前触发本方法
54     * @param channelContext the channelcontext
55     * @param throwable the throwable 有可能为空
56     * @param remark the remark 有可能为空
57     * @param isRemove
58     * @author tanyaowu
59     * @throws Exception 
60     */
61    public void onBeforeClose(ChannelContext channelContext, Throwable throwable, String remark, boolean isRemove) throws Exception;
62    /**
63     * 连接关闭前后触发本方法
64     * 警告：走到这个里面时，很多绑定的业务都已经解绑了，所以这个方法一般是空着不实现的
65     * @param channelContext the channelcontext
66     * @param throwable the throwable 有可能为空
67     * @param remark the remark 有可能为空
68     * @param isRemove 是否是删除
69     * @throws Exception
70     * @author: tanyaowu
71     */
72//    public void onAfterClose(ChannelContext channelContext, Throwable throwable, String remark, boolean isRemove) throws Exception;
73}

服务器端入口：TioServer

这个对象大家稍微了解一下即可，服务器启动时会用到这个对象，简单贴一下它的源代码吧，大家只需要关注它有一个start()方法是用来启动网络服务的即可

1import java.io.IOException;
2import java.lang.management.ManagementFactory;
3import java.lang.management.RuntimeMXBean;
4import java.net.InetSocketAddress;
5import java.net.StandardSocketOptions;
6import java.nio.channels.AsynchronousChannelGroup;
7import java.nio.channels.AsynchronousServerSocketChannel;
8import java.util.ArrayList;
9import java.util.Date;
10import java.util.List;
11import java.util.concurrent.TimeUnit;
12import org.slf4j.Logger;
13import org.slf4j.LoggerFactory;
14import org.tio.core.Node;
15import org.tio.utils.SysConst;
16import org.tio.utils.date.DateUtils;
17import org.tio.utils.hutool.StrUtil;
18/**
19 * @author tanyaowu
20 *
21 */
22public class TioServer {
23    private static Logger log = LoggerFactory.getLogger(TioServer.class);
24    private ServerTioConfig serverTioConfig;
25    private AsynchronousServerSocketChannel serverSocketChannel;
26    private AsynchronousChannelGroup channelGroup = null;
27    private Node serverNode;
28    private boolean isWaitingStop = false;
29    /**
30     *
31     * @param serverTioConfig
32     *
33     * @author tanyaowu
34     * 2017年1月2日 下午5:53:06
35     *
36     */
37    public TioServer(ServerTioConfig serverTioConfig) {
38        super();
39        this.serverTioConfig = serverTioConfig;
40    }
41    /**
42     * @return the serverTioConfig
43     */
44    public ServerTioConfig getServerTioConfig() {
45        return serverTioConfig;
46    }
47    /**
48     * @return the serverNode
49     */
50    public Node getServerNode() {
51        return serverNode;
52    }
53    /**
54     * @return the serverSocketChannel
55     */
56    public AsynchronousServerSocketChannel getServerSocketChannel() {
57        return serverSocketChannel;
58    }
59    /**
60     * @return the isWaitingStop
61     */
62    public boolean isWaitingStop() {
63        return isWaitingStop;
64    }
65    /**
66     * @param serverTioConfig the serverTioConfig to set
67     */
68    public void setServerTioConfig(ServerTioConfig serverTioConfig) {
69        this.serverTioConfig = serverTioConfig;
70    }
71    /**
72     * @param isWaitingStop the isWaitingStop to set
73     */
74    public void setWaitingStop(boolean isWaitingStop) {
75        this.isWaitingStop = isWaitingStop;
76    }
77    public void start(String serverIp, int serverPort) throws IOException {
78        long start = System.currentTimeMillis();
79        this.serverNode = new Node(serverIp, serverPort);
80        channelGroup = AsynchronousChannelGroup.withThreadPool(serverTioConfig.groupExecutor);
81        serverSocketChannel = AsynchronousServerSocketChannel.open(channelGroup);
82        serverSocketChannel.setOption(StandardSocketOptions.SO_REUSEADDR, true);
83        serverSocketChannel.setOption(StandardSocketOptions.SO_RCVBUF, 64 * 1024);
84        InetSocketAddress listenAddress = null;
85        if (StrUtil.isBlank(serverIp)) {
86            listenAddress = new InetSocketAddress(serverPort);
87        } else {
88            listenAddress = new InetSocketAddress(serverIp, serverPort);
89        }
90        serverSocketChannel.bind(listenAddress, 0);
91        AcceptCompletionHandler acceptCompletionHandler = serverTioConfig.getAcceptCompletionHandler();
92        serverSocketChannel.accept(this, acceptCompletionHandler);
93        serverTioConfig.startTime = System.currentTimeMillis();
94        //下面这段代码有点无聊，写得随意，纯粹是为了打印好看些
95        String baseStr = "|----------------------------------------------------------------------------------------|";
96        int baseLen = baseStr.length();
97        StackTraceElement[] ses = Thread.currentThread().getStackTrace();
98        StackTraceElement se = ses[ses.length - 1];
99        int xxLen = 18;
100        int aaLen = baseLen - 3;
101        List<String> infoList = new ArrayList<>();
102        infoList.add(StrUtil.fillAfter("Tio gitee address", ' ', xxLen) + "| " + SysConst.TIO_URL_GITEE);
103        infoList.add(StrUtil.fillAfter("Tio site address", ' ', xxLen) + "| " + SysConst.TIO_URL_SITE);
104        infoList.add(StrUtil.fillAfter("Tio version", ' ', xxLen) + "| " + SysConst.TIO_CORE_VERSION);
105        infoList.add(StrUtil.fillAfter("-", '-', aaLen));
106        infoList.add(StrUtil.fillAfter("TioConfig name", ' ', xxLen) + "| " + serverTioConfig.getName());
107        infoList.add(StrUtil.fillAfter("Started at", ' ', xxLen) + "| " + DateUtils.formatDateTime(new Date()));
108        infoList.add(StrUtil.fillAfter("Listen on", ' ', xxLen) + "| " + this.serverNode);
109        infoList.add(StrUtil.fillAfter("Main Class", ' ', xxLen) + "| " + se.getClassName());
110        try {
111            RuntimeMXBean runtimeMxBean = ManagementFactory.getRuntimeMXBean();
112            String runtimeName = runtimeMxBean.getName();
113            String pid = runtimeName.split("@")[0];
114            long startTime = runtimeMxBean.getStartTime();
115            long startCost = System.currentTimeMillis() - startTime;
116            infoList.add(StrUtil.fillAfter("Jvm start time", ' ', xxLen) + "| " + startCost + " ms");
117            infoList.add(StrUtil.fillAfter("Tio start time", ' ', xxLen) + "| " + (System.currentTimeMillis() - start) + " ms");
118            infoList.add(StrUtil.fillAfter("Pid", ' ', xxLen) + "| " + pid);
119        } catch (Exception e) {
120        }
121        //100
122        String printStr = "\r\n"+baseStr+"\r\n";
123        //        printStr += "|--" + leftStr + " " + info + " " + rightStr + "--|\r\n";
124        for (String string : infoList) {
125            printStr += "| " + StrUtil.fillAfter(string, ' ', aaLen) + "|\r\n";
126        }
127        printStr += baseStr + "\r\n";
128        if (log.isInfoEnabled()) {
129            log.info(printStr);
130        } else {
131            System.out.println(printStr);
132        }
133    }
134    /**
135     * 
136     * @return
137     * @author tanyaowu
138     */
139    public boolean stop() {
140        isWaitingStop = true;
141        boolean ret = true;
142        try {
143            channelGroup.shutdownNow();
144        } catch (Exception e) {
145            log.error("channelGroup.shutdownNow()时报错", e);
146        }
147        try {
148            serverSocketChannel.close();
149        } catch (Exception e1) {
150            log.error("serverSocketChannel.close()时报错", e1);
151        }
152        try {
153            serverTioConfig.groupExecutor.shutdown();
154        } catch (Exception e1) {
155            log.error(e1.toString(), e1);
156        }
157        try {
158            serverTioConfig.tioExecutor.shutdown();
159        } catch (Exception e1) {
160            log.error(e1.toString(), e1);
161        }
162        serverTioConfig.setStopped(true);
163        try {
164            ret = ret && serverTioConfig.groupExecutor.awaitTermination(6000, TimeUnit.SECONDS);
165            ret = ret && serverTioConfig.tioExecutor.awaitTermination(6000, TimeUnit.SECONDS);
166        } catch (InterruptedException e) {
167            log.error(e.getLocalizedMessage(), e);
168        }
169        log.info(this.serverNode + " stopped");
170        return ret;
171    }
172}

客户端入口：TioClient

只有当你在用t-io作为TCP客户端时，才用得到TioClient，此处简单贴一下它的源代码，它的用法，见后面的showcase示范工程

1package org.tio.client;
2import java.io.IOException;
3import java.net.InetSocketAddress;
4import java.net.StandardSocketOptions;
5import java.nio.channels.AsynchronousChannelGroup;
6import java.nio.channels.AsynchronousSocketChannel;
7import java.util.Set;
8import java.util.concurrent.CountDownLatch;
9import java.util.concurrent.LinkedBlockingQueue;
10import java.util.concurrent.TimeUnit;
11import java.util.concurrent.locks.ReentrantReadWriteLock;
12import java.util.concurrent.locks.ReentrantReadWriteLock.ReadLock;
13import java.util.concurrent.locks.ReentrantReadWriteLock.WriteLock;
14import org.slf4j.Logger;
15import org.slf4j.LoggerFactory;
16import org.tio.client.intf.ClientAioHandler;
17import org.tio.core.ChannelContext;
18import org.tio.core.Node;
19import org.tio.core.Tio;
20import org.tio.core.intf.Packet;
21import org.tio.core.ssl.SslFacadeContext;
22import org.tio.core.stat.ChannelStat;
23import org.tio.utils.SystemTimer;
24import org.tio.utils.hutool.StrUtil;
25import org.tio.utils.lock.SetWithLock;
26/**
27 *
28 * @author tanyaowu
29 * 2017年4月1日 上午9:29:58
30 */
31public class TioClient {
32    /**
33     * 自动重连任务
34     * @author tanyaowu
35     *
36     */
37    private static class ReconnRunnable implements Runnable {
38        ClientChannelContext channelContext = null;
39        TioClient tioClient = null;
40        //        private static Map<Node, Long> cacheMap = new HashMap<>();
41        public ReconnRunnable(ClientChannelContext channelContext, TioClient tioClient) {
42            this.channelContext = channelContext;
43            this.tioClient = tioClient;
44        }
45        /**
46         * @see java.lang.Runnable#run()
47         *
48         * @author tanyaowu
49         * 2017年2月2日 下午8:24:40
50         *
51         */
52        @Override
53        public void run() {
54            ReentrantReadWriteLock closeLock = channelContext.closeLock;
55            WriteLock writeLock = closeLock.writeLock();
56            writeLock.lock();
57            try {
58                if (!channelContext.isClosed) //已经连上了，不需要再重连了
59                {
60                    return;
61                }
62                long start = SystemTimer.currTime;
63                tioClient.reconnect(channelContext, 2);
64                long end = SystemTimer.currTime;
65                long iv = end - start;
66                if (iv >= 100) {
67                    log.error("{},重连耗时:{} ms", channelContext, iv);
68                } else {
69                    log.info("{},重连耗时:{} ms", channelContext, iv);
70                }
71                if (channelContext.isClosed) {
72                    channelContext.setReconnCount(channelContext.getReconnCount() + 1);
73                    //                    cacheMap.put(channelContext.getServerNode(), SystemTimer.currTime);
74                    return;
75                }
76            } catch (java.lang.Throwable e) {
77                log.error(e.toString(), e);
78            } finally {
79                writeLock.unlock();
80            }
81        }
82    }
83    private static Logger log = LoggerFactory.getLogger(TioClient.class);
84    private AsynchronousChannelGroup channelGroup;
85    private ClientTioConfig clientTioConfig;
86    /**
87     * @param serverIp 可以为空
88     * @param serverPort
89     * @param aioDecoder
90     * @param aioEncoder
91     * @param aioHandler
92     *
93     * @author tanyaowu
94     * @throws IOException
95     *
96     */
97    public TioClient(final ClientTioConfig clientTioConfig) throws IOException {
98        super();
99        this.clientTioConfig = clientTioConfig;
100        this.channelGroup = AsynchronousChannelGroup.withThreadPool(clientTioConfig.groupExecutor);
101        startHeartbeatTask();
102        startReconnTask();
103    }
104    /**
105     *
106     * @param serverNode
107     * @throws Exception
108     *
109     * @author tanyaowu
110     *
111     */
112    public void asynConnect(Node serverNode) throws Exception {
113        asynConnect(serverNode, null);
114    }
115    /**
116     *
117     * @param serverNode
118     * @param timeout
119     * @throws Exception
120     *
121     * @author tanyaowu
122     *
123     */
124    public void asynConnect(Node serverNode, Integer timeout) throws Exception {
125        asynConnect(serverNode, null, null, timeout);
126    }
127    /**
128     *
129     * @param serverNode
130     * @param bindIp
131     * @param bindPort
132     * @param timeout
133     * @throws Exception
134     *
135     * @author tanyaowu
136     *
137     */
138    public void asynConnect(Node serverNode, String bindIp, Integer bindPort, Integer timeout) throws Exception {
139        connect(serverNode, bindIp, bindPort, null, timeout, false);
140    }
141    /**
142     *
143     * @param serverNode
144     * @return
145     * @throws Exception
146     *
147     * @author tanyaowu
148     *
149     */
150    public ClientChannelContext connect(Node serverNode) throws Exception {
151        return connect(serverNode, null);
152    }
153    /**
154     *
155     * @param serverNode
156     * @param timeout
157     * @return
158     * @throws Exception
159     * @author tanyaowu
160     */
161    public ClientChannelContext connect(Node serverNode, Integer timeout) throws Exception {
162        return connect(serverNode, null, 0, timeout);
163    }
164    /**
165     *
166     * @param serverNode
167     * @param bindIp
168     * @param bindPort
169     * @param initClientChannelContext
170     * @param timeout 超时时间，单位秒
171     * @return
172     * @throws Exception
173     * @author tanyaowu
174     */
175    public ClientChannelContext connect(Node serverNode, String bindIp, Integer bindPort, ClientChannelContext initClientChannelContext, Integer timeout) throws Exception {
176        return connect(serverNode, bindIp, bindPort, initClientChannelContext, timeout, true);
177    }
178    /**
179     *
180     * @param serverNode
181     * @param bindIp
182     * @param bindPort
183     * @param initClientChannelContext
184     * @param timeout 超时时间，单位秒
185     * @param isSyn true: 同步, false: 异步
186     * @return
187     * @throws Exception
188     * @author tanyaowu
189     */
190    private ClientChannelContext connect(Node serverNode, String bindIp, Integer bindPort, ClientChannelContext initClientChannelContext, Integer timeout, boolean isSyn)
191            throws Exception {
192        AsynchronousSocketChannel asynchronousSocketChannel = null;
193        ClientChannelContext channelContext = null;
194        boolean isReconnect = initClientChannelContext != null;
195        //        ClientAioListener clientAioListener = clientTioConfig.getClientAioListener();
196        long start = SystemTimer.currTime;
197        asynchronousSocketChannel = AsynchronousSocketChannel.open(channelGroup);
198        long end = SystemTimer.currTime;
199        long iv = end - start;
200        if (iv >= 100) {
201            log.error("{}, open 耗时:{} ms", channelContext, iv);
202        }
203        asynchronousSocketChannel.setOption(StandardSocketOptions.TCP_NODELAY, true);
204        asynchronousSocketChannel.setOption(StandardSocketOptions.SO_REUSEADDR, true);
205        asynchronousSocketChannel.setOption(StandardSocketOptions.SO_KEEPALIVE, true);
206        InetSocketAddress bind = null;
207        if (bindPort != null && bindPort > 0) {
208            if (false == StrUtil.isBlank(bindIp)) {
209                bind = new InetSocketAddress(bindIp, bindPort);
210            } else {
211                bind = new InetSocketAddress(bindPort);
212            }
213        }
214        if (bind != null) {
215            asynchronousSocketChannel.bind(bind);
216        }
217        channelContext = initClientChannelContext;
218        start = SystemTimer.currTime;
219        InetSocketAddress inetSocketAddress = new InetSocketAddress(serverNode.getIp(), serverNode.getPort());
220        ConnectionCompletionVo attachment = new ConnectionCompletionVo(channelContext, this, isReconnect, asynchronousSocketChannel, serverNode, bindIp, bindPort);
221        if (isSyn) {
222            Integer realTimeout = timeout;
223            if (realTimeout == null) {
224                realTimeout = 5;
225            }
226            CountDownLatch countDownLatch = new CountDownLatch(1);
227            attachment.setCountDownLatch(countDownLatch);
228            asynchronousSocketChannel.connect(inetSocketAddress, attachment, clientTioConfig.getConnectionCompletionHandler());
229            boolean f = countDownLatch.await(realTimeout, TimeUnit.SECONDS);
230            if (f) {
231                return attachment.getChannelContext();
232            } else {
233                log.error("countDownLatch.await(realTimeout, TimeUnit.SECONDS) 返回false ");
234                return attachment.getChannelContext();
235            }
236        } else {
237            asynchronousSocketChannel.connect(inetSocketAddress, attachment, clientTioConfig.getConnectionCompletionHandler());
238            return null;
239        }
240    }
241    /**
242     *
243     * @param serverNode
244     * @param bindIp
245     * @param bindPort
246     * @param timeout 超时时间，单位秒
247     * @return
248     * @throws Exception
249     *
250     * @author tanyaowu
251     *
252     */
253    public ClientChannelContext connect(Node serverNode, String bindIp, Integer bindPort, Integer timeout) throws Exception {
254        return connect(serverNode, bindIp, bindPort, null, timeout);
255    }
256    /**
257     * @return the channelGroup
258     */
259    public AsynchronousChannelGroup getChannelGroup() {
260        return channelGroup;
261    }
262    /**
263     * @return the clientTioConfig
264     */
265    public ClientTioConfig getClientTioConfig() {
266        return clientTioConfig;
267    }
268    /**
269     *
270     * @param channelContext
271     * @param timeout
272     * @return
273     * @throws Exception
274     *
275     * @author tanyaowu
276     *
277     */
278    public void reconnect(ClientChannelContext channelContext, Integer timeout) throws Exception {
279        connect(channelContext.getServerNode(), channelContext.getBindIp(), channelContext.getBindPort(), channelContext, timeout);
280    }
281    /**
282     * @param clientTioConfig the clientTioConfig to set
283     */
284    public void setClientTioConfig(ClientTioConfig clientTioConfig) {
285        this.clientTioConfig = clientTioConfig;
286    }
287    /**
288     * 定时任务：发心跳
289     * @author tanyaowu
290     *
291     */
292    private void startHeartbeatTask() {
293        final ClientGroupStat clientGroupStat = (ClientGroupStat)clientTioConfig.groupStat;
294        final ClientAioHandler aioHandler = clientTioConfig.getClientAioHandler();
295        final String id = clientTioConfig.getId();
296        new Thread(new Runnable() {
297            @Override
298            public void run() {
299                while (!clientTioConfig.isStopped()) {
300//                    final long heartbeatTimeout = clientTioConfig.heartbeatTimeout;
301                    if (clientTioConfig.heartbeatTimeout <= 0) {
302                        log.warn("用户取消了框架层面的心跳定时发送功能，请用户自己去完成心跳机制");
303                        break;
304                    }
305                    SetWithLock<ChannelContext> setWithLock = clientTioConfig.connecteds;
306                    ReadLock readLock = setWithLock.readLock();
307                    readLock.lock();
308                    try {
309                        Set<ChannelContext> set = setWithLock.getObj();
310                        long currtime = SystemTimer.currTime;
311                        for (ChannelContext entry : set) {
312                            ClientChannelContext channelContext = (ClientChannelContext) entry;
313                            if (channelContext.isClosed || channelContext.isRemoved) {
314                                continue;
315                            }
316                            ChannelStat stat = channelContext.stat;
317                            long compareTime = Math.max(stat.latestTimeOfReceivedByte, stat.latestTimeOfSentPacket);
318                            long interval = currtime - compareTime;
319                            if (interval >= clientTioConfig.heartbeatTimeout / 2) {
320                                Packet packet = aioHandler.heartbeatPacket(channelContext);
321                                if (packet != null) {
322                                    if (log.isInfoEnabled()) {
323                                        log.info("{}发送心跳包", channelContext.toString());
324                                    }
325                                    Tio.send(channelContext, packet);
326                                }
327                            }
328                        }
329                        if (log.isInfoEnabled()) {
330                            log.info("[{}]: curr:{}, closed:{}, received:({}p)({}b), handled:{}, sent:({}p)({}b)", id, set.size(), clientGroupStat.closed.get(),
331                                    clientGroupStat.receivedPackets.get(), clientGroupStat.receivedBytes.get(), clientGroupStat.handledPackets.get(),
332                                    clientGroupStat.sentPackets.get(), clientGroupStat.sentBytes.get());
333                        }
334                    } catch (Throwable e) {
335                        log.error("", e);
336                    } finally {
337                        try {
338                            readLock.unlock();
339                            Thread.sleep(clientTioConfig.heartbeatTimeout / 4);
340                        } catch (Throwable e) {
341                            log.error(e.toString(), e);
342                        } finally {
343                        }
344                    }
345                }
346            }
347        }, "tio-timer-heartbeat" + id).start();
348    }
349    /**
350     * 启动重连任务
351     *
352     *
353     * @author tanyaowu
354     *
355     */
356    private void startReconnTask() {
357        final ReconnConf reconnConf = clientTioConfig.getReconnConf();
358        if (reconnConf == null || reconnConf.getInterval() <= 0) {
359            return;
360        }
361        final String id = clientTioConfig.getId();
362        Thread thread = new Thread(new Runnable() {
363            @Override
364            public void run() {
365                while (!clientTioConfig.isStopped()) {
366                    //log.info("准备重连");
367                    LinkedBlockingQueue<ChannelContext> queue = reconnConf.getQueue();
368                    ClientChannelContext channelContext = null;
369                    try {
370                        channelContext = (ClientChannelContext) queue.take();
371                    } catch (InterruptedException e1) {
372                        log.error(e1.toString(), e1);
373                    }
374                    if (channelContext == null) {
375                        continue;
376                        //                        return;
377                    }
378                    if (channelContext.isRemoved) //已经删除的，不需要重新再连
379                    {
380                        continue;
381                    }
382                    SslFacadeContext sslFacadeContext = channelContext.sslFacadeContext;
383                    if (sslFacadeContext != null) {
384                        sslFacadeContext.setHandshakeCompleted(false);
385                    }
386                    long sleeptime = reconnConf.getInterval() - (SystemTimer.currTime - channelContext.stat.timeInReconnQueue);
387                    //log.info("sleeptime:{}, closetime:{}", sleeptime, timeInReconnQueue);
388                    if (sleeptime > 0) {
389                        try {
390                            Thread.sleep(sleeptime);
391                        } catch (InterruptedException e) {
392                            log.error(e.toString(), e);
393                        }
394                    }
395                    if (channelContext.isRemoved || !channelContext.isClosed) //已经删除的和已经连上的，不需要重新再连
396                    {
397                        continue;
398                    }
399                    ReconnRunnable runnable = new ReconnRunnable(channelContext, TioClient.this);
400                    reconnConf.getThreadPoolExecutor().execute(runnable);
401                }
402            }
403        });
404        thread.setName("tio-timer-reconnect-" + id);
405        thread.setDaemon(true);
406        thread.start();
407    }
408    /**
409     * 
410     * @return
411     * @author tanyaowu
412     */
413    public boolean stop() {
414        boolean ret = true;
415        try {
416            clientTioConfig.groupExecutor.shutdown();
417        } catch (Exception e1) {
418            log.error(e1.toString(), e1);
419        }
420        try {
421            clientTioConfig.tioExecutor.shutdown();
422        } catch (Exception e1) {
423            log.error(e1.toString(), e1);
424        }
425        clientTioConfig.setStopped(true);
426        try {
427            ret = ret && clientTioConfig.groupExecutor.awaitTermination(6000, TimeUnit.SECONDS);
428            ret = ret && clientTioConfig.tioExecutor.awaitTermination(6000, TimeUnit.SECONDS);
429        } catch (InterruptedException e) {
430            log.error(e.getLocalizedMessage(), e);
431        }
432        log.info("client resource has released");
433        return ret;
434    }
435}