ThreadPoolExecutor: получить конкретный Runnable, который выполняется

Question

Я использую ThreadPoolExecutor для выполнения нескольких длинных задач в фоновом режиме, размер пула ThreadPoolExecutor равен 4, поэтому при добавлении более 4 задач они помещаются в очередь и когда один из 4 задач завершается, одна задача выставляется из очереди для выполнения.

Я хочу знать, есть ли способ доступа к объекту Runnable, которые в настоящее время исполняются, а не в очереди, то есть первые 4 задачи.

Цель: Я хочу сделать это, чтобы получить текущее состояние задачи в любой заданной точке, с помощью mThreadPoolExecutor.getQueue() Я получаю доступ к задачам, стоящим в очереди и готовым к выполнению, пожалуйста, предложите мне способ доступа к задачам, которые в настоящее время выполняются так что я могу прикрепить и удалить на нем слушатель/обработчик, когда это потребуется.

Мой Runnable класс:

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


    } 
} 

В Task держателе представления списка прогресс:

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 фокусируется на проблеме: «как знать, какие runnables в настоящее время выполняются».

Такого подход ведет параллельный набор активного Runnables:

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

И Runnables представляемого в 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's run() или инициализировать такие экземпляры с помощью набора MessageHandler или использовать этот подход к декорированию, чтобы не допустить его из класса VideoFileUploadRunner.

Answer 3

Этот ответ касается моего комментария выше.

Вместо того, чтобы пытаться найти исполняемый файл через исполнителя и прикрепить к нему прослушиватель, привяжите прослушиватель к runnable при его создании и опубликуйте свои события слушателю из исполняемого кода runnable.

Только текущие активные файлы будут публиковать свои обновления.

Вот пример.

Создайте интерфейс для реализации вашего слушателя. Ваш слушатель мог быть исполнителем пула потоков, частный внутренний класс и т.д.

/** 
* 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; 
} 

В способе выполнения, почтовые обновления слушателя, как вы считаете нужным. Реализация

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