的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()); 
    } 
};