怎样实现一个非阻塞的超时重试任务队列

时间  2019-11-16
标签 怎样 实现 一个 阻塞 超时 重试 任务 队列 繁體版
原文   原文链接

原由

最近接手一个项目，要把其中的阻塞任务队列，重构成非阻塞。在客户端不多有机会直接处理任务队列。项目完成须要总结经验html

阻塞的发生

我这里先说明我遇到的阻塞问题，我这里的阻塞不是多线程访问的阻塞，概念上是任务执行的阻塞。具体是：java

  • 任务开始客户端准备数据，经过socket向服务器发送数据。
  • 阻塞等待服务器socket的ack回应。
  • 获得服务器的socket回应完成任务，取出队列的后续任务继续执行。

这样的阻塞队列优势就是：android

  • 代码看起来很是简洁且汇集，开始代码对应结束代码。
  • 逻辑上能够保证任务的完成，由于若是没有完成，就会阻塞直到任务的完成。

可是致命的缺点也是阻塞等待，由于直接的socket通讯使用是不保证送达，若是服务器一直没有回应，客户端的任务队列就一直阻塞在队头。除非经过其余方式强制终止任务队列。git

非阻塞的队列

肯定了问题的发生的缘由，就能够一步步的解决问题。 首先阻塞就是由于在等待回应，只有回应后才能完成任务。任务以本地客户端开启，以服务器回应结束，期间阻塞。构成一个任务的概念。github

拆任务

其实客户端没必要执着等待回应，只要把任务拆分红服务器

  1. 发送任务
  2. 回应任务

而期间再也不阻塞，只要回应任务可以找到对应的发送任务，客户端就能够肯定该任务的完成。多线程

HandlerThread实现任务处理队列

这里socket的通讯确定是发生在子线程的，而子线程想要维护任务处理队列，最好的方式就是直接使用HandlerThread，它封装在子线程中Handler的配置，而Handler自己就是的任务处理队列。app

package com.example.licola.myandroiddemo.java;

import android.os.Handler;
import android.os.HandlerThread;
import java.util.HashSet;

/** * Created by LiCola on 2018/4/10. * 简化版非阻塞任务队列 */
public class Dispatcher {

  private static final String THREAD_NAME="dispatcher-worker";

  private Handler mHandler;
  private HandlerThread handlerThread;

  private HashSet<String> tasks = new HashSet<>();//任务集合

  public void run(){
    handlerThread = new HandlerThread(THREAD_NAME);
    handlerThread.start();
    mHandler = new Handler(handlerThread.getLooper());
  }

  public void postSendTask(String id,String data){
    mHandler.post(new Runnable() {
      @Override
      public void run() {
         //发送任务的操做 如准备数据等

        tasks.add(id);
      }
    });
  }

  public void postAckTask(final String id){
    mHandler.post(new Runnable() {
      @Override
      public void run() {
        //回应任务的操做 如解析回应等

        tasks.remove(id);
      }
    });
  }
}

复制代码

上面的代码已经很是简化，不涉及具体的任务处理，只有关键代码。实现了前文的拆任务的理念。socket

可是拆任务也带来了一个很严重的问题，任务怎样保证完成。由于不阻塞，发送任务只管发送，发送完成迎来的多是下一个发送任务，而对应的回应任务却一直没有到来。概念上这个任务始终没有完成。代码上就是tasks堆积愈来愈多等待回应的任务。ide

超时机制

为了应对可能堆积的tasks任务集合，就须要引入超时机制，就是给一个任务设定最长等待时间，若是超过这个时间尚未完成就重试。有了前面的代码基础加入超时检测处理是很容易的。

  • 首先想到的就是在运行过程当中加入按期循环执行的检测代码。
  • 给发送任务加入时间变量，用于检测超时。
  • 任务集合保存任务id和对应的发送数据，用于重试。

超时重试机制的任务队列

package com.example.licola.myandroiddemo.java;

import android.os.Handler;
import android.os.HandlerThread;
import android.util.Pair;
import com.example.licola.myandroiddemo.utils.Logger;
import java.util.HashMap;
import java.util.Map.Entry;

/** * Created by LiCola on 2018/4/10. * 支持超时重试机制版非阻塞任务队列 */
public class Dispatcher {

  private static final String THREAD_NAME = "dispatcher-worker";

  //超时检测时间
  private static final long CHECK_ACK_TIME_OUT = 10 * 1000;
  //任务限定等待时间，即任务超时时间
  private static final long ACK_TIME_OUT = 4 * 1000;

  private Handler mHandler;
  private HandlerThread handlerThread;

  private HashMap<String, Pair<Long, String>> tasks=new HashMap<>();//任务集合

  public void run() {
    handlerThread = new HandlerThread(THREAD_NAME);
    handlerThread.start();
    mHandler = new Handler(handlerThread.getLooper());

    //开启循环检测
    mHandler.postDelayed(checkTimeOutTask(), CHECK_ACK_TIME_OUT);
  }

  public void postSendTask(final String id, final String data) {
    mHandler.post(new Runnable() {
      @Override
      public void run() {
        //发送任务的操做 如准备数据等

        Logger.d("开始发送任务");
        tasks.put(id, new Pair<>(System.currentTimeMillis(), data));
      }
    });
  }

  public void postAckTask(final String id) {
    mHandler.post(new Runnable() {
      @Override
      public void run() {
        //回应任务的操做 如解析回应等

        Logger.d("开始回应任务");
        tasks.remove(id);
      }
    });
  }

  public Runnable checkTimeOutTask() {
    return new Runnable() {
      @Override
      public void run() {
        int count = 0;
        long curTime = System.currentTimeMillis();

        if (!tasks.isEmpty()) {
          for (Entry<String, Pair<Long, String>> entry : tasks.entrySet()) {
            String id = entry.getKey();
            Pair<Long, String> pair = entry.getValue();
            Long time = pair.first;
            String data = pair.second;
            if (curTime - time >= ACK_TIME_OUT) {
              postSendTask(id, data);
              count++;
            }
          }
        }

        if (count > 0) {
          Logger.d(String.format("检测到超时任务%d", count));
        }

        //循环检测
        mHandler.postDelayed(checkTimeOutTask(), CHECK_ACK_TIME_OUT);

      }
    };
  }
}

复制代码

上面的代码已经实现超时重试机制。仔细想一想这段代码的运行状况。仍是问题和有优化空间的。

检测时机

仔细想一想按期检测的时间和限定的超时时间，二者的关系。

//超时检测时间
  private static final long CHECK_ACK_TIME_OUT = 10 * 1000;
  //任务限定等待时间，即任务超时时间
  private static final long ACK_TIME_OUT = 4 * 1000;
复制代码

为了检测尽量的高效，且不影响整个任务队列处理性能。让检测时间间隔比较大，且大于任务超时时间。 实际的运行状况极可能以下图所示：

运行时间图

咱们以时间点check为基准分析：

  1. check-1时间点以前：开始任务task-一、task-2。
  2. check-1时间点：检测开始，发现任务集合中有2个等待任务，可是它们都没有超时，没有任何处理。
  3. check-2时间点以前：task-1正常完成，任务集合中删除它。
  4. check-2时间定：检测开始，发现任务集合中有1个等待任务，且已经超时。task-2任务重试，task-2的计时重置到当前时间点。

这是一种假设运行状况，可是仍是暴露出了两个问题：

  • 不够高效：虽然检测时间间隔足够大，一个间隔内可以完成整个发送回应的正常任务，可是检测并无很高效，仍是在check-1时间点中观察到了两个不该该被观察到的任务。其中task-1：它刚开始且能够正常完成的。
  • 不精确的超时：在check-2以前任务task-2它已经超时了，可是在超时一段时间后才发现。

这两个问题其实不严重，根据实际状况选择。 若是任务的超时小几率发生，且不要求精确的超时检测。超时重试机制的任务处理队列-非精确控制时间，仍是足够知足开发需求的。

精确的控制超时时间

怎样作到精确的控制超时时间，且让检测更高效。在Android开发中有没有遇到精确控制任务时间的状况，而其余工程师们怎样实现高效处理的。虽然咱们平常开发中没有感知，可是这个状况其实很是很是的广泛存在。把这个问题换个角度：

怎样精确的控制任务时间？

再想一想你开发的各类系统处理：

  • 长按点击事件的监听
  • ANR（Application Not Response）的检测和发生

这两个系统处理本质上就是精确控制任务时间的处理。

源码的智慧

肯定了上面这两个源码目标，咱们来看看系统是怎样实现的。

长按点击事件

一个点击的事件序列由ACTION_DOWN开始，后续的事件action不肯定。

开始：

任务的开始就是在View.onTouchEvent(MotionEvent event)的action事件处理cast:MotionEvent.ACTION_DOWN中的方法checkForLongClick(0, x, y) 核心代码就一行：

private void checkForLongClick(int delayOffset, float x, float y) {
        if ((mViewFlags & LONG_CLICKABLE) == LONG_CLICKABLE || (mViewFlags & TOOLTIP) == TOOLTIP) {
		      
		      //发送延迟任务
            postDelayed(mPendingCheckForLongPress,
                    ViewConfiguration.getLongPressTimeout() - delayOffset);
        }
    }
复制代码

结束：

点击任务处理已经开始，而典型点击任务结束就是ACTION_UP事件，一样在代码中cast:MotionEvent.ACTION_UP中的方法removeLongPressCallback()

private void removeLongPressCallback() {
        if (mPendingCheckForLongPress != null) {
            removeCallbacks(mPendingCheckForLongPress);
        }
    }
复制代码

超时：

由于在开始就已经肯定固定时间点后执行超时处理，在这个时间点以前没有其余action操做来及时remove掉超时处理。从而超时处理获得执行，具体就是执行长按事件。

private final class CheckForLongPress implements Runnable {
    
        @Override
        public void run() {
            if ((mOriginalPressedState == isPressed()) && (mParent != null)
                    && mOriginalWindowAttachCount == mWindowAttachCount) {
                if (performLongClick(mX, mY)) {
                    mHasPerformedLongPress = true;
                }
            }
        }
    }
复制代码

ANR的检查与发生

总所周知ANR的发生有不少种，这里就挑Service的建立超时来举例说明

Service Timeout:好比前台服务在20s内未执行完成。

这里参考理解Android ANR的触发原理的分析流程。做者很形象的总结整个ANR检测的理念：

埋炸弹-拆炸弹

由于ANR的处理比较复杂，咱们省略自动写日志和进程通讯等流程。

开始：埋炸弹

ActiveServices源码部分

private final void realStartServiceLocked(ServiceRecord r, ProcessRecord app, boolean execInFg) throws RemoteException {
    ...
    //发送delay消息(SERVICE_TIMEOUT_MSG)
    bumpServiceExecutingLocked(r, execInFg, "create");
    try {
        ...
        //最终执行服务的onCreate()方法
        app.thread.scheduleCreateService(r, r.serviceInfo,
                mAm.compatibilityInfoForPackageLocked(r.serviceInfo.applicationInfo),
                app.repProcState);
    } catch (DeadObjectException e) {
        mAm.appDiedLocked(app);
        throw e;
    } finally {
        ...
    }
}
复制代码
private final void bumpServiceExecutingLocked(ServiceRecord r, boolean fg, String why) {
    ... 
    scheduleServiceTimeoutLocked(r.app);
}

void scheduleServiceTimeoutLocked(ProcessRecord proc) {
    if (proc.executingServices.size() == 0 || proc.thread == null) {
        return;
    }
    long now = SystemClock.uptimeMillis();
    Message msg = mAm.mHandler.obtainMessage(
            ActivityManagerService.SERVICE_TIMEOUT_MSG);
    msg.obj = proc;
    
    //当超时后仍没有remove该SERVICE_TIMEOUT_MSG消息，则执行service Timeout流程
    mAm.mHandler.sendMessageAtTime(msg,
        proc.execServicesFg ? (now+SERVICE_TIMEOUT) : (now+ SERVICE_BACKGROUND_TIMEOUT));
}
复制代码

结束：拆炸弹

在Service的启动前，已经埋下了炸弹，那就在启动完成后拆掉炸弹。 ActiveServices源码部分

private void serviceDoneExecutingLocked(ServiceRecord r, boolean inDestroying, boolean finishing) {
    ...
    if (r.executeNesting <= 0) {
        if (r.app != null) {
            r.app.execServicesFg = false;
            r.app.executingServices.remove(r);
            if (r.app.executingServices.size() == 0) {
                //当前服务所在进程中没有正在执行的service
                mAm.mHandler.removeMessages(ActivityManagerService.SERVICE_TIMEOUT_MSG, r.app);
        ...
    }
    ...
}
复制代码

超时：炸弹爆炸

若是Service没有限定时间内完成启动，拆掉炸弹，炸弹就会爆炸，就是超时任务执行。 就是ActiveService的serviceTimeout方法执行，写下日志发出ANR弹框。

总结

咱们从精确控制任务超时时间这角度，分析了长按事件和ANR的发生原理。最终发现他们都是基于一样的设计方式：埋炸弹-拆炸弹 在任务开始时设置定时任务，及时完成remove掉定时任务，不然任务超时就会执行超时处理，而定时任务精确的时间执行就保证了超时任务精确控制。这个方式彻底不一样于我前文实现的间隔检测-非精确时间控制。

超时重试机制的任务队列-精确控制时间

有对源码的理解和总结，稍微修改代码就能够获得以下

package com.example.licola.myandroiddemo.java;

import android.os.Handler;
import android.os.HandlerThread;
import com.example.licola.myandroiddemo.utils.Logger;
import java.util.HashMap;

/** * Created by LiCola on 2018/4/10. * 支持超时重试机制版非阻塞任务队列 */
public class DispatcherTime {

  private static final String THREAD_NAME = "dispatcher-worker";

  //任务限定等待时间，即任务超时时间
  private static final long ACK_TIME_OUT = 2 * 1000;

  private Handler mHandler;
  private HandlerThread handlerThread;

  private HashMap<String, Runnable> timeoutTask = new HashMap<>();//超时集合

  public void run() {
    handlerThread = new HandlerThread(THREAD_NAME);
    handlerThread.start();
    mHandler = new Handler(handlerThread.getLooper());
  }

  public void postSendTask(final String id, final String data) {
    mHandler.post(new Runnable() {
      @Override
      public void run() {
        //发送任务的操做 如准备数据等

        Logger.d("开始发送任务",data);
        Runnable checkTimeOutTask = checkTimeOutTask(id, data);
        timeoutTask.put(id, checkTimeOutTask);
        mHandler.postDelayed(checkTimeOutTask,ACK_TIME_OUT);
      }
    });
  }

  public void postAckTask(final String id) {
    mHandler.post(new Runnable() {
      @Override
      public void run() {
        //回应任务的操做 如解析回应等

        Logger.d("开始回应任务",id);
        Runnable runnable = timeoutTask.remove(id);
        mHandler.removeCallbacks(runnable);
      }
    });
  }

  public Runnable checkTimeOutTask(final String id, final String data) {
    return new Runnable() {
      @Override
      public void run() {

        Logger.d("超时任务执行 ",id,data);
        postSendTask(id, data);
      }
    };
  }
}
复制代码

上面实现了每次任务发送都会埋下一个延迟任务，若是没有及时获得回应就会重试。 这个实现的缺点若是要说的就是：

  • 每一个发送任务都会建立一个对应的延迟任务，若是发送任务数量较大，且只有小几率任务超时，就会产生大量建立的任务而又短时间存在且没有机会执行的任务。

固然若是要优化就是使用Handler.handleMessage(Message msg)方法处理超时任务，而不是每次postDelayed都建立Runnable对象。这里只留下思路就不用代码了。

总结

  • 其实源码的理解不是很难，只要找到切入点，从你关心的点出发，就可以理解源码并应用它。咱们从超时任务的处理为切入点就很容易理解长按事件的原理和ANR的发生机制。
  • 当咱们了解到一个新的解决方案，不要急于去应用它，要分析新方案的利弊，和咱们实际项目的匹配程度，才能很好的应用它和改造它。

参考

相关文章
相关标签/搜索
每日一句
    每一个你不满意的现在，都有一个你没有努力的曾经。
本站公众号
   欢迎关注本站公众号,获取更多信息
联系我们 最近搜索 最新文章 沪ICP备13005482号-10 MyBatis教程 SQL 教程 MySQL教程 Java 教程 Thymeleaf 教程 Hibernate教程 Spring教程 Redis教程