talent-aio源码阅读小记（二）

在上篇文章中，我们提到了talent-aio的四类Task：DecodeRunnable、HandlerRunnable、SendRunnable、CloseRunnable，并且分析了这些task的基类AbstractQueueRunnable。在这篇文章中，我们就来分析一下这几个task是如何相互协作来处理输入、输出数据以及客户端连接的。

talent-aio 基础类介绍

在开始介绍前，我们需要先介绍几个talent-aio的基础类和接口，在处理输入输出的整个流程中，都需要用到它们。两个基础的类为：

ChannelContext
GroupContext

其中ChannelContext类包含：

  1. AsynchronousSocketChannel对象、与该连接有关的
  2. DecodeRunnable、HandlerRunnable、SendRunnable、CloseRunnable
  3. GroupContext
  4. ReadCompletionHandler
    ChannelContext类代表一个socket连接，封装了：连接对应的AsynchronousSocketChannel对象；解码、处理接收到的数据和发送数据、关闭连接的几个Runnable；该连接所属的GroupContext引用；该连接的ReadCompletionHandler等。

在创建ChannelContext的过程中，首先记录accept返回的asynchronousSocketChannel:

    public void setAsynchronousSocketChannel(AsynchronousSocketChannel
                         asynchronousSocketChannel)
    {
        this.asynchronousSocketChannel = asynchronousSocketChannel;

        if (asynchronousSocketChannel != null)
        {
            try
            {
                clientNode = getClientNode(asynchronousSocketChannel);
            } catch (IOException e)
            {
                log.error(e.toString(), e);
            }
        } else
        {
            clientNode = null;
        }
    }

而后设置GroupContext，并创建几个Runnable，最终将自己加入GroupContext管理的连接集合connections和对端集合clientNodes中：

    public void setGroupContext(GroupContext<Ext, P, R> groupContext)
    {
        this.groupContext = groupContext;

        if (groupContext != null)
        {
            decodeRunnable = new DecodeRunnable<>(this, 
groupContext.getDecodeExecutor());
            closeRunnable = new CloseRunnable<>(this, null, 
null, groupContext.getCloseExecutor());

            handlerRunnableNormPrior = new HandlerRunnable<>(this, 
groupContext.getHandlerExecutorNormPrior());

            sendRunnableNormPrior = new SendRunnable<>(this, 
groupContext.getSendExecutorNormPrior());

            groupContext.getConnections().add(this);
            groupContext.getClientNodes().put(this);
        }
    }

而GroupContext管理着当前实例的所有连接connections、连接对应的用户users以及用户间的分组关系groups：


    protected ClientNodes<Ext, P, R> clientNodes = new ClientNodes<>();
    protected Connections<Ext, P, R> connections = new Connections<>();
    protected Groups<Ext, P, R> groups = new Groups<>();
    protected Users<Ext, P, R> users = new Users<>();

除此之外，GroupContext还管理着一些线程池：

/**
 * 解码线程池
*/
private SynThreadPoolExecutor<SynRunnableIntf> decodeExecutor;

/**
 * 关闭连接的线程池
 */
private SynThreadPoolExecutor<SynRunnableIntf> closeExecutor;

/**
 * 业务处理线程池
 */
private SynThreadPoolExecutor<SynRunnableIntf> handlerExecutorNormPrior;

/**
 * 消息发送线程池
 */
private SynThreadPoolExecutor<SynRunnableIntf> sendExecutorNormPrior;

在处理输入输出的过程中我们会用到上面这两个类。

talent-aio的主要的业务逻辑都在AioHandler接口的实现类中，AioHandler接口为：

public interface AioHandler<Ext, P extends Packet, R>
{

    /**
     * 处理消息包
     *
     * @param packet the packet
     * @return the r
     * @author: tanyaowu
     * @创建时间:　2016年11月15日 上午11:38:52
     */
    R handler(P packet,
 ChannelContext<Ext, P, R> channelContext) throws Exception;

    /**
     * 编码
     *
     * @param packet the packet
     * @return the byte buffer
     * @author: tanyaowu
     * @创建时间:　2016年11月15日 上午11:38:52
     */
    ByteBuffer encode(P packet, 
ChannelContext<Ext, P, R> channelContext);

    /**
     * 根据ByteBuffer解码成业务需要的Packet对象.
     *
     * @param buffer the buffer
     * @return the t
     * @throws AioDecodeException the aio decode exception
     */
    P decode(ByteBuffer buffer, ChannelContext<Ext, P, R> channelContext)
 throws AioDecodeException;

}

三个方法分别用来数据解包、数据封包以及处理数据包。

处理流程

读取数据

首先，是处理读取数据：

    public void completed(Integer result, ByteBuffer byteBuffer)
    {
        GroupContext<Ext, P, R> groupContext 
  = channelContext.getGroupContext();
        if (result > 0)
        {
            byteBuffer.limit(byteBuffer.position());
            byteBuffer.position(0);
            DecodeRunnable<Ext, P, R> decodeRunnable
 = channelContext.getDecodeRunnable();
            decodeRunnable.addMsg(byteBuffer);

            groupContext.getDecodeExecutor().execute(decodeRunnable);

        } else if (result == 0)
        {
            log.error("读到的数据长度为0");
        } else if (result < 0)
        {
            Aio.close(channelContext, null, "读数据时返回" + result);
        }

        if (AioUtils.checkBeforeIO(channelContext))
        {
            AsynchronousSocketChannel asynchronousSocketChannel 
= channelContext.getAsynchronousSocketChannel();
            ByteBuffer newByteBuffer = 
      ByteBuffer.allocate(
channelContext.getGroupContext().getReadBufferSize()
);
            asynchronousSocketChannel.read(newByteBuffer,
 newByteBuffer, this);
        }

    }

在读取数据后，将byteBuffer提交到decodeRunnable的数据队列中，而后继续调用read方法读取对端发送来的数据。

数据组包

在decodeRunnable中，调用AioHandler的decode方法来获取数据包，然后提交给HandlerRunnable处理。在处理数据时主要考虑了半包和粘包的情况：

@Override
public void runTask()
{
    ConcurrentLinkedQueue<ByteBuffer> queue
 = getMsgQueue();
        
    ByteBuffer byteBuffer = null;
    label_1: while ((size = queue.size()) > 0)
    {
        byteBuffer = queue.poll();
        if (byteBuffer != null)
        {
            if (lastByteBuffer != null)
            {
                byteBuffer.position(0);
                byteBuffer = 
            ByteBufferUtils.composite(lastByteBuffer, byteBuffer);
                lastByteBuffer = null;
            }
        } else {
            break label_1;
        }

        try
        {
            byteBuffer.position(0);
            label_2: while (true)
            {
                int initPosition = byteBuffer.position();
                P packet = 
        channelContext.
        getGroupContext().
        getAioHandler().
        decode(byteBuffer, channelContext);

                if (packet == null)// 数据不够，组不了包
                {
                    if (log.isDebugEnabled())
                    {
                        log.debug("{},数据不够，组不了包", 
                             channelContext.toString());
                    }
                    byteBuffer.position(initPosition);
                    lastByteBuffer = byteBuffer;
                    continue label_1;
                } else //组包成功
                {       
                    channelContext.getStat().
            setLatestTimeOfReceivedPacket(
              SystemTimer.currentTimeMillis()
            );  
                    int afterDecodePosition = byteBuffer.position();
                    int len = afterDecodePosition - initPosition;
                    AioListener<Ext, P, R> aioListener
       = channelContext.getGroupContext().getAioListener();
                    if (aioListener != null)
                    {
                          aioListener.
onAfterDecoded(channelContext, packet, len);
                    }
                    submit(packet, len);
                    channelContext.getGroupContext().
                                  getGroupStat().
                                  getReceivedPacket().
                                  incrementAndGet();

                                  channelContext.getGroupContext().
                                  getGroupStat().
                                  getReceivedBytes().
                                  addAndGet(len);
                        
                  if (byteBuffer.hasRemaining())//组包后，还剩有数据
                  {
                        if (log.isDebugEnabled())
                        {
                            log.debug("{}组包后，还剩有数据:{}", 
          channelContext, byteBuffer.limit() - byteBuffer.position());
                        }
                            continue label_2;
                        } else//组包后，数据刚好用完
                        {
                            lastByteBuffer = null;
                            log.debug("{},组包后，数据刚好用完", 
                                                channelContext);
                            continue label_1;
                        }
                    }
                }
            } catch (AioDecodeException e)
            {
                log.error(channelContext.toString(), e);
                Aio.close(channelContext, e, "解码异常:" 
                                + e.getMessage());
                break label_1;
            } finally
            {

            }
        }
    }

在这个循环中，首先从队列中获取数据，获取到数据后，看看是否存在粘包多出来的数据lastByteBuffer, 如果存在则将两部分数据合并。而后调用AioHandler的decode方法处理数据，如果由于半包导致解包失败，则继续从队列中获取数据，组合起来尝试解包；如果解包成功，则将多余的数据放在lastByteBuffer，并且更新各种统计信息。最后，将组好的数据包通过submit方法传递给HandlerRunnable处理。

数据处理

HandlerRunnable的逻辑相对比较简单，从数据队列中获取组好的包，并调用doPacket处理数据包：

    @Override
    public void runTask()
    {
        ConcurrentLinkedQueue<P> queue = getMsgQueue();
        P packet = null;
        while ((packet = queue.poll()) != null)
        {
            doPacket(packet);
        }
    }

在doPacket中，调用AioHandler的hanlder解除处理数据包：

groupContext.getAioHandler().handler(packet, channelContext);

发送数据

发送数据使用Aio类的静态方法send将packet添加到SendRunnable的数据队列中，而后将sendRunnable提交到线程池中运行：

public static <Ext, P extends Packet, R> void send(
ChannelContext<Ext, P, R> channelContext,
 P packet
)
{
    if (channelContext == null)
    {
        log.error("channelContext == null");
        return;
    }
    SendRunnable<Ext, P, R> sendRunnable
   = AioUtils.selectSendRunnable(channelContext, packet);
    sendRunnable.addMsg(packet);
    SynThreadPoolExecutor<SynRunnableIntf> synThreadPoolExecutor
   = AioUtils.selectSendExecutor(channelContext, packet);
    synThreadPoolExecutor.execute(sendRunnable);
}

在线程池中，会尝试一次发送所有等待发送的packet，不过对单次发送的packet设置了一个上限，而后对每个packet编码，汇总到一个ByteBuffer中：

for ( int i = 0; i < queueSize; i++ )
{
    if ( (packet = queue.poll() ) != null )
    {
        ByteBuffer byteBuffer = aioHandler.encode( packet, channelContext );
        allBytebufferCapacity += byteBuffer.limit();
        packetCount++;
        byteBuffers[i] = byteBuffer;

        if ( aioListener != null )
        {
            try
            {
                aioListener.onBeforeSent( channelContext, packet );
            } catch ( Exception e )
            {
                log.error( e.toString(), e );
            }
        }
    } else{
        break;
    }
}
ByteBuffer allByteBuffer = ByteBuffer.allocate( allBytebufferCapacity );
for ( ByteBuffer byteBuffer : byteBuffers )
{
    if ( byteBuffer != null )
    {
        byteBuffer.flip();
        allByteBuffer.put( byteBuffer );
    }
}

最终在sendRunnable的sendByteBuffer中完成数据发送，注意发送前需要获取一个信号量，保证同一时间对一个连接只有一个线程在调用发送动作，并在writeCompleteHandler中记录当前连接发送数据，并更新当前连接活动时间，为keepalive做准备：

public void sendByteBuffer( ByteBuffer byteBuffer, Integer packetCount ) {
    if ( byteBuffer == null )
    {
        log.error( "byteBuffer is null" );
        return;
    }

    if ( !AioUtils.checkBeforeIO( channelContext ) )
    {
        return;
    }

    byteBuffer.flip();
    AsynchronousSocketChannel  asynchronousSocketChannel    
                                              = channelContext.getAsynchronousSocketChannel();
    WriteCompletionHandler<Ext, P, R>  writeCompletionHandler       
                                             = channelContext.getWriteCompletionHandler();
    try
    {
        writeCompletionHandler.getWriteSemaphore().acquire();
    } catch ( InterruptedException e )
    {
        log.error( e.toString(), e );
    }
    asynchronousSocketChannel.write( byteBuffer, packetCount, writeCompletionHandler );
}

注意发送时如果队列中有待发送数据将直接在当前进程中继续运行runTask，而不是重新提交Runnable到线程池中，以尽快发送数据：

if (queue.size() > 0) {
    runTask();
}

关闭连接

在关闭连接时，首先将解码，处理，发送runnable全部停止，然后清空其数据队列，最后检查该连接是否已经 调用过关闭，如果没有，就将CloseRunnable提交到线程池中执行：

public static < Ext, P extends Packet, R > void close( ChannelContext<Ext, P, R> channelContext, Throwable t, String remark, boolean isRemove )
{
    channelContext.getDecodeRunnable().clearMsgQueue();
    channelContext.getHandlerRunnableNormPrior().clearMsgQueue();
    channelContext.getSendRunnableNormPrior().clearMsgQueue();

    channelContext.getDecodeRunnable().setCanceled( true );
    channelContext.getHandlerRunnableNormPrior().setCanceled( true );
    channelContext.getSendRunnableNormPrior().setCanceled( true );

    CloseRunnable<Ext, P, R> closeRunnable = channelContext.getCloseRunnable();
    if ( closeRunnable.isWaitingExecute() )
    {
        log.error( "{},已经在等待关闭\r\n本次关闭备注:{}\r\n第一次的备注:{}\r\n本次关闭异常:{}\r\n第一次时异常:{}", channelContext, remark, closeRunnable.getRemark(), t, closeRunnable.getT() );
        return;
    }
    synchronized (closeRunnable) {
        if ( closeRunnable.isWaitingExecute() ) /* double check */
        {
            return;
        }
        closeRunnable.setRemove( isRemove );
        closeRunnable.setRemark( remark );
        closeRunnable.setT( t );
        closeRunnable.getExecutor().execute( closeRunnable );
        closeRunnable.setWaitingExecute( true );
    }
}

在CloseRunnable的runTask中，主要执行关闭连接，清理连接数据的逻辑：

//关闭连接
try
{
    AsynchronousSocketChannel asynchronousSocketChannel = channelContext.getAsynchronousSocketChannel();
    if ( asynchronousSocketChannel != null )
    {
        asynchronousSocketChannel.close();
    }
} catch ( Throwable e )
{
    log.error( e.toString() );
}

//清理连接数据
//删除集合中的维护信息 start
/*删除集合中的维护信息 start */
try
{
    groupContext.getConnections().remove( channelContext );
} catch ( Throwable e )
{
    log.error( e.toString(), e );
}

try
{
    groupContext.getClientNodes().remove( channelContext );
} catch ( Throwable e )
{
    log.error( e.toString(), e );
}

try
{
    groupContext.getUsers().unbind( channelContext );
} catch ( Throwable e )
{
    log.error( e.toString(), e );
}

try
{
    groupContext.getGroups().unbind( channelContext );
} catch ( Throwable e )
{
    log.error( e.toString(), e );
}

channelContext.setClosed( true );
channelContext.getGroupContext().getGroupStat().getClosed().incrementAndGet();
/*删除集合中的维护信息 end */

除此以外，如果连接是因为出错被关闭的，还会根据ReconnConf的配置进行断线重连操作，这个我们将在后文中讲解。

本篇总结

本篇着重讲解了talent-aio如何处理半包、粘包的情况，对输入进行解包，并处理输入数据包，并返回数据的。
在本系列的下一篇中，会讲解talent-aio的断线重连、keepalive等功能是如何实现的，并且接受其是如何定制线程以及executor的。最后，还将给出一个简单的使用talent-io的IM server的示例。预知后事如何，请听下回分解~

最后编辑于 ：
©著作权归作者所有,转载或内容合作请联系作者
  • 人面猴
    序言：七十年代末，一起剥皮案震惊了整个滨河市，随后出现的几起案子，更是在滨河造成了极大的恐慌，老刑警刘岩，带你破解...
    沈念sama阅读 194,088评论 5赞 459
  • 死咒
    序言：滨河连续发生了三起死亡事件，死亡现场离奇诡异，居然都是意外死亡，警方通过查阅死者的电脑和手机，发现死者居然都...
    沈念sama阅读 81,715评论 2赞 371
  • 救了他两次的神仙让他今天三更去死
    文/潘晓璐 我一进店门，熙熙楼的掌柜王于贵愁眉苦脸地迎上来，“玉大人，你说我怎么就摊上这事。” “怎么了？”我有些...
    开封第一讲书人阅读 141,361评论 0赞 319
  • 道士缉凶录：失踪的卖姜人
    文/不坏的土叔 我叫张陵，是天一观的道长。 经常有香客问我，道长，这世上最难降的妖魔是什么？ 我笑而不...
    开封第一讲书人阅读 52,099评论 1赞 263
  • 港岛之恋（遗憾婚礼）
    正文 为了忘掉前任，我火速办了婚礼，结果婚礼上，老公的妹妹穿的比我还像新娘。我一直安慰自己，他们只是感情好，可当我...
    茶点故事阅读 60,987评论 4赞 355
  • 恶毒庶女顶嫁案：这布局不是一般人想出来的
    文/花漫 我一把揭开白布。 她就那样静静地躺着，像睡着了一般。 火红的嫁衣衬着肌肤如雪。 梳的纹丝不乱的头发上，一...
    开封第一讲书人阅读 46,063评论 1赞 272
  • 城市分裂传说
    那天，我揣着相机与录音，去河边找鬼。 笑死，一个胖子当着我的面吹牛，可吹牛的内容都是我干的。 我是一名探鬼主播，决...
    沈念sama阅读 36,486评论 3赞 381
  • 双鸳鸯连环套：你想象不到人心有多黑
    文/苍兰香墨 我猛地睁开眼，长吁一口气：“原来是场噩梦啊……” “哼！你这毒妇竟也来了？” 一声冷哼从身侧响起，我...
    开封第一讲书人阅读 35,175评论 0赞 253
  • 万荣杀人案实录
    序言：老挝万荣一对情侣失踪，失踪者是张志新（化名）和其女友刘颖，没想到半个月后，有当地人在树林里发现了一具尸体，经...
    沈念sama阅读 39,440评论 1赞 290
  • 护林员之死
    正文 独居荒郊野岭守林人离奇死亡，尸身上长有42处带血的脓包…… 初始之章·张勋 以下内容为张勋视角 年9月15日...
    茶点故事阅读 34,518评论 2赞 309
  • 白月光启示录
    正文 我和宋清朗相恋三年，在试婚纱的时候发现自己被绿了。 大学时的朋友给我发了我未婚夫和他白月光在一起吃饭的照片。...
    茶点故事阅读 36,305评论 1赞 326
  • 活死人
    序言：一个原本活蹦乱跳的男人离奇死亡，死状恐怖，灵堂内的尸体忽然破棺而出，到底是诈尸还是另有隐情，我是刑警宁泽，带...
    沈念sama阅读 32,190评论 3赞 312
  • 日本核电站爆炸内幕
    正文 年R本政府宣布，位于F岛的核电站，受9级特大地震影响，放射性物质发生泄漏。R本人自食恶果不足惜，却给世界环境...
    茶点故事阅读 37,550评论 3赞 298
  • 男人毒药：我在死后第九天来索命
    文/蒙蒙 一、第九天 我趴在偏房一处隐蔽的房顶上张望。 院中可真热闹，春花似锦、人声如沸。这庄子的主人今日做“春日...
    开封第一讲书人阅读 28,880评论 0赞 17
  • 一桩弑父案，背后竟有这般阴谋
    文/苍兰香墨 我抬头看了看天上的太阳。三九已至，却和暖如春，着一层夹袄步出监牢的瞬间，已是汗流浃背。 一阵脚步声响...
    开封第一讲书人阅读 30,152评论 1赞 250
  • 情欲美人皮
    我被黑心中介骗来泰国打工， 没想到刚下飞机就差点儿被人妖公主榨干…… 1. 我叫王不留，地道东北人。 一个月前我还...
    沈念sama阅读 41,451评论 2赞 341
  • 代替公主和亲
    正文 我出身青楼，却偏偏与公主长得像，于是被迫代替她去往敌国和亲。 传闻我的和亲对象是个残疾皇子，可洞房花烛夜当晚...
    茶点故事阅读 40,637评论 2赞 335

推荐阅读更多精彩内容