的ThreadPoolExecutor：获取正在执行

Question

我使用ThreadPoolExecutor执行在后台多长时间运行的任务的具体Runnable接口的ThreadPoolExecutor池大小是4所以正在增加超过4任务时，他们都推到队列中，并且当一个4个任务完成后，一个任务从队列中移出执行。

我想知道有什么方法可以访问Runnable当前正在执行的对象，而不是队列，即前4个任务。

目标：我想要在任何给定点上获得当前任务的状态，在mThreadPoolExecutor.getQueue()的帮助下，我正在访问正在排队并准备执行的任务，请告诉我如何访问当前正在执行的任务这样我就可以在需要时附加和移除监听器/处理器。

我运行的类：

public class VideoFileUploadRunner implements Runnable { 

    private final VideoFileSync mVideoFileSync; 
    private final DataService dataService; 

    private Handler handler; 

    public VideoFileUploadRunner(VideoFileSync videoFileSync, DataService dataService) { 
     this.mVideoFileSync = videoFileSync; 
     this.dataService = dataService; 

    } 

    public int getPK() 
    { 
     return mVideoFileSync.get_idPrimaryKey(); 
    } 

    public void setHandler(Handler handler) { 
     this.handler = handler; 
    } 

    @Override 
    public void run() { 
     try { 

      if (mVideoFileSync.get_idPrimaryKey() < 0) { 
       addEntryToDataBase(); 
      } 
      updateStatus(VideoUploadStatus.IN_PROGRESS); 
      FileUploader uploader = new FileUploader(); 
      updateStatus(uploader.uploadFile(mVideoFileSync.getVideoFile())); 



     } catch (Exception e) { 
      updateStatus(VideoUploadStatus.FAILED); 
      e.printStackTrace(); 
     } 
    } 

    private void addEntryToDataBase() { 
     int pk = dataService.saveVideoRecordForSync(mVideoFileSync); 
     mVideoFileSync.set_idPrimaryKey(pk); 
    } 

    private void updateStatus(VideoUploadStatus status) { 
     if (handler != null) { 
      Message msg = new Message(); 
      Bundle b = new Bundle(); 
      b.putString(AppConstants.Sync_Status, status.toString()); 
      msg.setData(b); 
      handler.sendMessage(msg); 
     } 
     dataService.updateUploadStatus(mVideoFileSync.get_idPrimaryKey(), status.toString()); 


    } 
} 

在任务进度列表视图持有人：

public void setData(VideoFileSync fileSync) { 
     tvIso.setText(fileSync.getVideoFile().getISO_LOOP_EQUP()); 
     tvUnit.setText(fileSync.getVideoFile().getUnit()); 
     tvName.setText(fileSync.getVideoFile().getLocalPath()); 
     tvStatus.setText(fileSync.getCurentStatus().toString()); 
     addHandleForUpdate(fileSync); 
    } 

    private void addHandleForUpdate(VideoFileSync fileSync) { 

     Handler.Callback callBack = new Handler.Callback() { 
      @Override 
      public boolean handleMessage(Message msg) { 
       if(msg.getData()!=null) 
       { 
        tvStatus.setText(msg.getData().getString(AppConstants.Sync_Status)); 

       } 
       return false; 
      } 
     }; 
     mHadler = new Handler(Looper.getMainLooper(),callBack); 

     VideoFileUploadRunner runner = VideoUploadManager.getInstance().getRunnerForSyncFile(fileSync); 
     if(runner!=null) 
     runner.setHandler(mHadler); 
    } 

在VideoUploadManager

我有以下方法返回Runnable对象，在这里，我想帮助，使我可以返回任务正在执行。

public synchronized VideoFileUploadRunner getRunnerForSyncFile(VideoFileSync fileSync) { 
     Iterator<Runnable> itr = mThreadPoolExecutor.getQueue().iterator(); 
     while (itr.hasNext()) { 
      VideoFileUploadRunner runner = (VideoFileUploadRunner) itr.next(); 
      if (runner.getPK() == fileSync.get_idPrimaryKey()) { 
       return runner; 
      } 
     } 
     return null; 

    } 

Answer 1

，最好的办法是揭露同步变量保存在当前正在执行的任务的信息。

public MyTask implements Runnable { 
    private String id; 
    private Map<String, MyTask> mapTasks; 

    public MyTask(String id, Map<String, MyTask> mapTasks) { 
     this.id = id; 
     this.mapTasks = mapTasks; 
    } 

    public void run() { 
     synchronized(mapTasks) { 
      mapTasks.put(id, this); 
     } 

     ... 

     synchronized(mapTasks) { 
      mapTasks.remove(id); 
     } 
    } 
} 


// Create a map of tasks 
Map<String, MyTask> mapTasks = new HashMap<String, MyTask>(); 

// How to create tasks 
MyTask myTask1 = new MyTask("task1", mapTasks); 
MyTask myTask2 = new MyTask("task2", mapTasks); 

executorService.execute(myTask1); 
executorService.execute(myTask2); 

.... 

而且当前打印任务列表中执行：

public void printCurrentExecutingTasks(Map<String, MyTask> tasks) { 
    for (String id: tasks.keySet()) { 
     System.out.println("Executing task with id: " + id); 
    } 
} 

Answer 2

我anwser的重点问题：“如何知道哪些是执行可运行”。

这种方法保持活跃的Runnable的并发的设定：

private final Set<VideoFileUploadRunner> active = Collections.newSetFromMap(new ConcurrentHashMap<>()); 

和的Runnable被提交到ThreadPoolExecutor的应该用一个Runnable的更新来装饰这一套：

class DecoratedRunnable implements Runnable { 

    final VideoFileUploadRunner runnable; 

    public DecoratedRunnable(VideoFileUploadRunner runnable) { 
     this.runnable = runnable; 
    } 

    @Override 
    public void run() { 
     active.add(runnable); // add to set 
     try { 
      runnable.run(); 
     } finally { 
      active.remove(runnable); // finally remove from set (even when something goes wrong) 
     } 
    } 
} 

所以我们可以在提交它们之前装饰VideoFileUploadRunner实例：

executorService.submit(new DecoratedRunnable(videoFileUploadRunner)); 

getRunnerForSyncFile会那么简单地这样来实现的方法：

public VideoFileUploadRunner getRunnerForSyncFile(VideoFileSync fileSync) { 
    return active.stream() 
      .filter(videoFileUploadRunner -> videoFileUploadRunner.getPK() == fileSync.get_idPrimaryKey()) 
      .findAny() 
      .orElse(null); 
} 

备注：作为@Charlie评论，这不是一个监听器附加到一个Runnable的最佳途径。您可以请求从VideoFileUploadRunner的run()方法中设置消息处理程序，或使用MessageHandler集初始化此类实例，或者使用此装饰方法将其保留在VideoFileUploadRunner类之外。

Answer 3

这个答案与我上面的评论有关。

不是试图通过执行程序找到可运行程序并将侦听程序附加到它，而是在创建它时将侦听程序绑定到可运行程序，并将您的事件从可运行程序的执行代码发布到侦听程序。

只有当前活动的runnables会发布其更新。

下面是一个例子。

为您的侦听器创建一个接口来实现。你的听众可能是线程池的执行者，私有内部类等

/** 
* Callback interface to notify when a video upload's state changes 
*/ 
interface IVideoUploadListener { 

    /** 
    * Called when a video upload's state changes 

    * @param pUploadId The ID of the video upload 
    * @param pStatus The new status of the upload 
    */ 
    void onStatusChanged(int pUploadId, VideoUploadStatus pStatus); 
} 

您的状态类型创建一个枚举（例如）

/** 
* Enum to hold different video upload states 
*/ 
enum VideoUploadStatus { 
    IN_PROGRESS, 
    ADDED_TO_DB, 
    FILE_UPLOADED, 
    FINISHED, 
    FAILED 
} 

每个Runnable的抓住听者的参考。

public class VideoFileUploadRunner implements Runnable { 

    private final IVideoUploadListener mUploadListener; 
    private final VideoFileSync mVideoFileSync; 
    private final DataService mDataService; 
    private Handler mHandler; 

    // etc... 
} 

通过构造

public VideoFileUploadRunner(IVideoUploadListener pUploadListener, VideoFileSync pVideoFileSync, DataService pDataService) { 
    mUploadListener = pUploadListener; 
    mVideoFileSync = pVideoFileSync; 
    mDataService = pDataService; 
} 

在run方法传递接口的一个实例，更新后的听众，你认为合适。

@Override 
public void run() { 
    mUploadListener.onStatusChanged(getPrimaryKey(), VideoUploadStatus.IN_PROGRESS); 
    try { 
     if (mVideoFileSync.get_idPrimaryKey() < 0) { 
      addEntryToDataBase(); 
      mUploadListener.onStatusChanged(getPrimaryKey(), VideoUploadStatus.ADDED_TO_DB); 
     } 
     FileUploader uploader = new FileUploader(); 
     uploader.uploadFile(mVideoFileSync.getVideoFile()); 
     mUploadListener.onStatusChanged(getPrimaryKey(), VideoUploadStatus.FILE_UPLOADED); 

     // Other logic here... 

     mUploadListener.onStatusChanged(getPrimaryKey(), VideoUploadStatus.FINISHED); 
    } 

    catch (Exception e) { 
     mUploadListener.onStatusChanged(getPrimaryKey(), VideoUploadStatus.FAILED); 
     e.printStackTrace(); 
    } 
} 

听众实现onStatusChanged（）方法应该是同步。这将有助于避免竞赛状况的错误结果。

private IVideoUploadListener mUploadListener = new IVideoUploadListener() { 
    @Override 
    public synchronized void onStatusChanged(int pUploadId, VideoUploadStatus pStatus) { 
     Log.i("ListenerTag", "Video file with ID " + pUploadId + " has the status " + pStatus.toString()); 
    } 
};