代码小析 - 异步回调

/**
 * 异步取得channel
 * @param index
 * @param callback
 */
public void asynGetChannel(int index,final Callback callback) {
    // 1. 随机获取一条channel
    final int pos = ThreadLocalRandom.current().nextInt(MAX_CONNECTIONS);
    Channel target = channels[pos];

    // 2. 如果获取到了连接，直接返回
    if (target != null && target.isActive()) {
        logger.info("direct success "+index);
        callback.onSuccess(target);
        return;
    }

    synchronized (locks[pos]) {
        target = channels[pos];
        // 2. 如果获取到了连接，直接返回
        if (target != null && target.isActive()) {
            callback.onSuccess(target);
            return;
        }

        // 3.如果连接正在创建中，则加入queue
        if (target instanceof EmptyChannel) {
            boolean result = jobs.offer(callback);
            if (result) {
                return;
            } else {
                throw new RuntimeException("Can't connet to target server and the waiting queue is full");
            }
        }

        // 4. 连接尚未创建
        channels[pos] = new EmptyChannel();

       Connector.connect(host, port, new Callback() {
           @Override
           public void onSuccess(Channel channel) {
               logger.info(index + " ------------connect success---------"+pos + " channel:" +channels[pos].getClass().getName());
               List<Callback> tmpJobs;//建立一个tempJobs，快速释放锁
               synchronized (locks[pos]) {
                   // 设置channels，拷贝jobs队列，释放锁
                   channels[pos] = channel;
                   tmpJobs = drainJobs();
               }
               for(Callback pendingCallback : tmpJobs) {
                   try {
                       if(pendingCallback != callback) {
                           pendingCallback.onSuccess(channel);
                       }
                   } catch (Exception e) {
                       logger.error("call connectionCallback fail", e);
                   }
               }
           }

           @Override
           public void onError(Throwable e) {
               List<Callback> tmpJobs;//建立一个tempJobs，快速释放锁
               synchronized (locks[pos]) {
                   // 设置channels，拷贝jobs队列，释放锁
                   channels[pos] = null;
                   tmpJobs = drainJobs();
               }
               for(Callback pendingCallback : tmpJobs) {
                   try {
                       if(pendingCallback != callback) {
                           pendingCallback.onError(e);
                       }
                   } catch (Exception x) {
                       logger.error("call connectionCallback fail", x);
                   }
               }
           }
       });
    }
}

Future future = doTask1();
doTask2();
doTask3();
Result result = future.get();