ThreadPoolExecutor:获取正在执行的特定Runnable

Question

我使用ThreadPoolExecutor在后台执行多个长时间运行的任务，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关注的问题是："如何知道哪些可运行程序正在执行"。

这种方法保持一组并发的活动Runnables:

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

提交给ThreadPoolExecutor的Runnables应该用一个更新该集的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

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

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

只有当前活动的可运行程序才会发布其更新。

下面是一个例子。

为侦听器创建一个要实现的接口。您的监听器可以是线程池执行器、私有内部类等。

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