在我的代码中，我有许多作为管道运行的 executorservices，因为第一个 executorService 可以将任务提交给任何后续的 executor 服务，但绝不会反过来。
services.add(songLoaderService);
services.add(AcoustIdMatcher.getExecutorService());
services.add(SongPrematcherMatcher.getExecutorService());
services.add(MusicBrainzMetadataMatcher.getExecutorService());
//Start Loading Songs
songLoaderService.submit(loader);
我们只为第一次服务提交一个任务，然后我可以请求关闭。在此任务完成之前，这不会成功，到那时它将把一些任务交给第二个服务，依此类推。
所以这段代码已经运行了很多年，shutdown() 直到所有要提交的任务都已经完成才会被调用，而 awaitTermination() 方法直到所有提交的任务都完成才被调用.
int count = 0;
for (ExecutorService service : services)
{
MainWindow.logger.severe("Requested Shutdown Task:" + count + ":"+((SongKongThreadFactory)((TimeoutThreadPoolExecutor) service).getThreadFactory()).getName());
//Request Shutdown
service.shutdown();
//Now wait for service to terminate
service.awaitTermination(10, TimeUnit.DAYS);
MainWindow.logger.severe("Completed Shutdown Task:" + count);
if(count==2)
{
MainWindow.logger.severe("Report:"+currentReportId+":SongPreMatcher:" + SongPrematcherMatcher.getPipelineQueuedCount()+":"+ SongPrematcherMatcher.getPipelineCallCount()+":"+ SongPrematcherMatcher.getPipelineCompletedCount()+":"+SongPrematcherMatcher.getPipelineFileCount());
}
count++;
}
但我现在看到一个 ExecutorService 无法以这种方式工作的问题。 关闭 SongPrematcherMatcher 服务的请求在之前的 (AcoustIdMatcher) 服务添加到此服务的所有任务都已提交并启动后成功，但其中一个完成之前，如下面的调试所示行
Report:353:SongPreMatcher:init:57:started:57:Finished:56
丢失的任务并没有失败，因为我们可以看到它在日志输出的末尾完成，但重点是它在其运行的服务成功终止后完成。
这会产生重大后果，因为这意味着此任务尝试提交给 MusicBrainzMetadataMatcher 服务的所有任务都会失败，因为自从上一个服务 (PrematcherMatched) 已关闭以来，已经为此提出了关闭请求。
PrematcherMatcher 是最近添加的，所以我的假设是它有问题，但我看不出它可能是什么。
toplevelanalyzer.FixSongsController:start:SEVERE: Requested Shutdown Task:0
analyser.AcoustIdMatcher:<init>:SEVERE: GROUP 115:C:\Users\Paul\Desktop\FlatRandomFolder:true:false:true:false
toplevelanalyzer.FixSongsController:start:SEVERE: Completed Shutdown Task:0
toplevelanalyzer.FixSongsController:start:SEVERE: Requested Shutdown Task:1:analyser.AcoustIdMatcher
analyser.SongPrematcherMatcher:<init>:SEVERE: Queue:GROUP 791:C:\Users\Paul\Desktop\FlatRandomFolder:true:false:true:false
analyser.SongPrematcherMatcher:call:SEVERE: Start:GROUP 791:C:\Users\Paul\Desktop\FlatRandomFolder:true:false:true:false
toplevelanalyzer.FixSongsController:start:SEVERE: Completed Shutdown Task:1
toplevelanalyzer.FixSongsController:start:SEVERE: Requested Shutdown Task:2:analyser.SongPrematcherMatcher
toplevelanalyzer.FixSongsController:start:SEVERE: Completed Shutdown Task:2
toplevelanalyzer.FixSongsController:start:SEVERE: Report:353:SongPreMatcher:init:57:started:57:Finished:56
toplevelanalyzer.FixSongsController:start:SEVERE: Requested Shutdown Task:3:analyser.MusicBrainzMetadataMatcher
analyser.MusicBrainzMetadataMatcher:<init>:SEVERE: Queue:GROUP 795:C:\Users\Paul\Desktop\FlatRandomFolder:true:false:true:false
analyser.MusicBrainzMetadataMatcher:<init>:SEVERE: Queue:GROUP 797:C:\Users\Paul\Desktop\FlatRandomFolder:true:false:true:false
analyser.MusicBrainzMetadataMatcher:<init>:SEVERE: Queue:GROUP 799:C:\Users\Paul\Desktop\FlatRandomFolder:true:false:true:false
analyser.MusicBrainzMetadataMatcher:<init>:SEVERE: Queue:GROUP 821:C:\Users\Paul\Desktop\FlatRandomFolder:true:false:true:false
analyser.MusicBrainzMetadataMatcher:<init>:SEVERE: Queue:GROUP 823:C:\Users\Paul\Desktop\FlatRandomFolder:true:false:true:false
analyser.SongPrematcherMatcher:call:SEVERE: Finish:GROUP 791:C:\Users\Paul\Desktop\FlatRandomFolder:true:false:true:false
如果我在循环中人为地添加一个延迟，那么至少对于这个特定的测试用例它是有效的，但这不是一个解决方案，因为当存在一系列执行程序服务时它会引入延迟，现在关闭它们是有效的一个接一个。也不清楚为什么此修复有效，以及它是否始终有效。
for (ExecutorService service : services)
{
Thread.sleep(5000);
//Request Shutdown
service.shutdown();
......
}
每个任务都有一个执行器服务，只能将一种类型的任务添加到特定的执行器服务。 executorService 确实有特殊处理允许用户取消任务并防止长时间运行的任务，但这不是这里的问题。
我看不出 PreMatcherMatcher 代码与任何其他任务有什么不同。
package com.jthink.songkong.analyse.analyser;
import com.jthink.songkong.analyse.general.Errors;
import com.jthink.songkong.cmdline.SongKong;
import com.jthink.songkong.ui.MainWindow;
import com.jthink.songkong.util.SongKongThreadFactory;
import java.util.List;
import java.util.concurrent.*;
/**
* From http://stackoverflow.com/questions/2758612/executorservice-that-interrupts-tasks-after-a-timeout
* With additional support for caller running task when bounded queue is full
*/
public class TimeoutThreadPoolExecutor extends ThreadPoolExecutor {
private final long timeout;
private final TimeUnit timeoutUnit;
private final ScheduledExecutorService timeoutExecutor = Executors.newSingleThreadScheduledExecutor();
private final ConcurrentMap<Runnable, ScheduledFuture> runningTasks = new ConcurrentHashMap<Runnable, ScheduledFuture>();
private final static int WAIT_BEFORE_STOP = 10000;
public long getTimeout()
{
return timeout;
}
public TimeUnit getTimeoutUnit()
{
return timeoutUnit;
}
public TimeoutThreadPoolExecutor(int workerSize, ThreadFactory threadFactory, LinkedBlockingQueue<Runnable> queue,long timeout, TimeUnit timeoutUnit)
{
super(workerSize, workerSize, 0L, TimeUnit.MILLISECONDS, queue, threadFactory, new ThreadPoolExecutor.CallerRunsPolicy());
this.timeout = timeout;
this.timeoutUnit = timeoutUnit;
}
@Override
public List<Runnable> shutdownNow() {
timeoutExecutor.shutdownNow();
return super.shutdownNow();
}
@Override
public <T> FutureCallable<T> newTaskFor(Callable<T> callable) {
return new FutureCallable<T>(callable);
}
@Override
protected void beforeExecute(Thread t, Runnable r) {
MainWindow.logger.warning("beforeExecute:"+t.getName()+":"+r.toString());
SongKong.checkIn();
if(timeout > 0) {
final ScheduledFuture<?> scheduled = timeoutExecutor.schedule(new TimeoutTask(t,r), timeout, timeoutUnit);
runningTasks.put(r, scheduled);
}
}
@Override
protected void afterExecute(Runnable r, Throwable t) {
MainWindow.logger.warning("afterExecute:"+r.toString());
//AfterExecute will be called after the task has completed, either of its own accord or because it
//took too long and was interrupted by corresponding timeout task
//Remove mapping and cancel timeout task
ScheduledFuture timeoutTask = runningTasks.remove(r);
if(timeoutTask != null) {
timeoutTask.cancel(false);
}
}
@Override
protected void terminated()
{
//All tasks have completed either naturally or via being cancelled by timeout task so close the timeout task
MainWindow.logger.warning("---Terminated:"+((SongKongThreadFactory)getThreadFactory()).getName());
timeoutExecutor.shutdown();
}
class TimeoutTask implements Runnable {
private final Thread thread;
private Callable c;
public TimeoutTask(Thread thread, Runnable c) {
this.thread = thread;
if(c instanceof FutureCallable)
{
this.c = ((FutureCallable) c).getCallable();
}
}
@Override
public void run()
{
String msg = "";
if (c != null)
{
if (c instanceof AcoustIdMatcher)
{
msg = c.getClass() + ":" + ((AcoustIdMatcher) c).getSongGroup().getKey();
}
else if (c instanceof SongPrematcherMatcher)
{
msg = c.getClass() + ":" + ((SongPrematcherMatcher) c).getSongGroup().getKey();
}
else if (c instanceof MusicBrainzSongGroupMatcher)
{
msg = c.getClass() + ":" + ((MusicBrainzSongGroupMatcher) c).getSongGroup().getKey();
}
else if (c instanceof MusicBrainzMetadataMatcher)
{
msg = c.getClass() + ":" + ((MusicBrainzMetadataMatcher) c).getSongGroup().getKey();
}
else if (c instanceof MusicBrainzUpdateSongOnly)
{
msg = c.getClass() + ":" + ((MusicBrainzUpdateSongOnly) c).getSongGroup().getKey();
}
else if (c instanceof DiscogsSongGroupMatcher)
{
msg = c.getClass() + ":" + ((DiscogsSongGroupMatcher) c).getSongGroup().getKey();
}
else if (c instanceof MusicBrainzSongMatcher)
{
msg = c.getClass() + ":" + String.valueOf(((MusicBrainzSongMatcher) c).getSongId());
}
else if (c instanceof SongSaver)
{
msg = c.getClass() + ":" + String.valueOf(((SongSaver) c).getSongId());
}
else
{
msg = c.getClass().getName();
}
}
if (c != null && c instanceof CancelableTask)
{
MainWindow.logger.warning("+++Cancelling " + msg + " task because taking too long");
((CancelableTask) c).setCancelTask(true);
StackTraceElement[] stackTrace = thread.getStackTrace();
Errors.addError("Cancelled " + msg + " because taken too long", stackTrace);
Counters.getErrors().getCounter().incrementAndGet();
if(stackTrace.length>0)
{
boolean isKnownProblem = false;
for(int i=0;i<stackTrace.length;i++)
{
if(
(stackTrace[i].getClassName().contains("CosineSimilarity")) ||
(stackTrace[i].getClassName().contains("com.jthink.songkong.fileloader.FileFilters"))
)
{
isKnownProblem=true;
break;
}
}
if(isKnownProblem)
{
MainWindow.logger.warning("+++Interrupting " + msg + " task because taking too long");
thread.interrupt();
try
{
Thread.sleep(WAIT_BEFORE_STOP);
}
catch (InterruptedException ie)
{
MainWindow.logger.warning("+++Interrupted TimeoutTask " + msg + " task because taking too long");
}
if(thread.isAlive())
{
MainWindow.logger.warning("+++Stopping CosineSimailarity task");
thread.stop();
}
}
}
}
}
}
}
public class AnalyserService
{
protected static final int BOUNDED_QUEUE_SIZE = 500;
protected String threadGroup;
public AnalyserService(String threadGroup)
{
this.threadGroup=threadGroup;
}
protected ExecutorService executorService;
protected void initExecutorService()
{
int workerSize = Runtime.getRuntime().availableProcessors();
executorService = new PausableExecutor(workerSize, workerSize,0L, TimeUnit.MILLISECONDS,new LinkedBlockingQueue<Runnable>(BOUNDED_QUEUE_SIZE),new SongKongThreadFactory(threadGroup));
}
public ExecutorService getExecutorService()
{
if (executorService == null || executorService.isShutdown())
{
initExecutorService();
}
return executorService;
}
/** Submit and return immediately
*
* @param task
*/
public void submit(Callable<Boolean> task) //throws Exception
{
executorService.submit(task);
}
}
public class AnalyserServiceWithTimeout extends AnalyserService
{
private static final int TIMEOUT_PER_TASK = 30;
public AnalyserServiceWithTimeout(String threadGroup)
{
super(threadGroup);
}
@Override
protected void initExecutorService()
{
int workerSize = Runtime.getRuntime().availableProcessors();
executorService = new TimeoutThreadPoolExecutor(workerSize,
new SongKongThreadFactory(threadGroup),
new LinkedBlockingQueue<Runnable>(BOUNDED_QUEUE_SIZE),
TIMEOUT_PER_TASK,
TimeUnit.MINUTES);
}
}
package com.jthink.songkong.analyse.analyser;
import com.google.common.base.Strings;
import com.jthink.songkong.analyse.general.Errors;
import com.jthink.songkong.cmdline.SongKong;
import com.jthink.songkong.db.SongCache;
import com.jthink.songkong.match.MetadataGatherer;
import com.jthink.songkong.preferences.UserPreferences;
import com.jthink.songkong.ui.MainWindow;
import com.jthink.songkong.util.SongKongThreadGroup;
import com.jthink.songlayer.Song;
import com.jthink.songlayer.hibernate.HibernateUtil;
import org.hibernate.Session;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutorService;
import java.util.logging.Level;
/**
* Try and match songs to acoustid only first as a starting point
*
* Use when we have no or little metadata
*/
public class SongPrematcherMatcher extends CancelableTask implements Callable<Boolean> {
private static PipelineCount pipelineCount = new PipelineCount();
public static int getPipelineQueuedCount()
{
return pipelineCount.getQueuedCount();
}
public static int getPipelineCallCount()
{
return pipelineCount.getCallCount();
}
public static void resetPipelineCount()
{
pipelineCount.resetCounts();
}
public static int getPipelineFileCount()
{
return pipelineCount.getFileCount();
}
public static int getPipelineCompletedCount()
{
return pipelineCount.getCompletedCount();
}
private static AnalyserService analyserService = new AnalyserServiceWithTimeout(SongKongThreadGroup.THREAD_PREMATCHER_WORKER);
private Session session;
private SongGroup songGroup;
public SongGroup getSongGroup()
{
return songGroup;
}
public SongPrematcherMatcher(SongGroup songGroup)
{
SongKong.logger.severe("Queue:"+ songGroup.getKey());
pipelineCount.incQueuedCount();
pipelineCount.incFileCount(songGroup.getSongIds().size());
this.songGroup = songGroup;
}
public static ExecutorService getExecutorService()
{
return analyserService.getExecutorService();
}
public static AnalyserService getService()
{
return analyserService;
}
public Boolean call()
{
try
{
SongKong.logger.severe("Start:" + songGroup.getKey());
if (SongKong.isStopTask() || isCancelTask())
{
return false;
}
SongKong.checkIn();
pipelineCount.incCallCount();
session = HibernateUtil.beginTransaction();
AnalysisStats stats = new AnalysisStats();
List<Song> songs = SongCache.loadSongsFromDatabase(session, songGroup.getSongIds());
//Try to match acoustid this should allow more to be grouped and matched by metadata on first pass
try
{
new RecordingOnlyMatcher().matchRecordingsOnlyByAcoustid(session, songGroup, songs, stats);
}
catch(Exception ex)
{
MainWindow.logger.log(Level.SEVERE, Strings.nullToEmpty(ex.getMessage()), ex);
Errors.addError(Strings.nullToEmpty(ex.getMessage()));
}
session.getTransaction().commit();
HibernateUtil.closeSession(session);
processSongsWithNewMetadata(songGroup, songs);
pipelineCount.incCompletedCount();
SongKong.logger.severe("Finish:" + songGroup.getKey());
return true;
}
catch (Exception e)
{
SongKong.logger.severe("FinishFail:" + songGroup.getKey());
MainWindow.logger.log(Level.SEVERE, "SongPrematcherMatcher:" + e.getMessage(), e);
if (session.getTransaction() != null)
{
session.getTransaction().rollback();
}
return false;
}
catch (Error e)
{
SongKong.logger.severe("FinishFail:" + songGroup.getKey());
MainWindow.logger.log(Level.SEVERE, "SongPrematcherMatcher:" + e.getMessage(), e);
if (session.getTransaction() != null)
{
session.getTransaction().rollback();
}
return false;
}
catch (Throwable t)
{
SongKong.logger.severe("FinishFail:" + songGroup.getKey());
MainWindow.logger.log(Level.SEVERE, "SongPrematcherMatcher:" + t.getMessage(), t);
if (session.getTransaction() != null)
{
session.getTransaction().rollback();
}
return false;
}
finally
{
if(session.isOpen())
{
session.getTransaction().commit();
HibernateUtil.closeSession(session);
}
}
}
private boolean processSongsWithNewMetadata(SongGroup songGroup, List<Song> songs)
{
MainWindow.logger.info("Prematcher:" + songGroup.getKey() + ":totalcount:" + songs.size());
int count = 0;
//Group based on actual metadata only
MetadataGatherer mg = new MetadataGatherer(songs);
for (String album : mg.getAlbums().keySet())
{
List<Song> songsInGrouping = mg.getAlbums().get(album);
count+=songsInGrouping.size();
MainWindow.logger.warning("Prematcher:" + songGroup.getKey() + ":" + album + ":count:" + songsInGrouping.size());
SongGroup sg = SongGroup.createSongGroupForSongs(songGroup, songsInGrouping);
sg.setRandomFolderNoMetadata(false);
sg.setRandomFolder(false);
processRandomFolder(sg, songsInGrouping);
}
List<Song> songsWithNoInfo = new ArrayList<>(mg.getSongsWithNoRelease());
if(songsWithNoInfo.size()>0)
{
count+=songsWithNoInfo.size();
SongGroup sgWithNoInfo = SongGroup.createSongGroupForSongs(songGroup, songsWithNoInfo);
MainWindow.logger.warning("Prematcher:" + songGroup.getKey() + ":NoMetadata:" + ":count:" + songsWithNoInfo.size());
processRandomFolderNoMetadata(sgWithNoInfo, songsWithNoInfo);
}
if(count<songs.size())
{
MainWindow.logger.warning(songGroup.getKey()+":Not all songs have been processed"+songs.size());
Errors.addErrorWithoutStackTrace(songGroup.getKey()+":Not all songs have been processed:"+songs.size());
}
return true;
}
private boolean processRandomFolder(SongGroup songGroup, List<Song> songs)
{
if(UserPreferences.getInstance().isSearchMusicBrainz())
{
MusicBrainzMetadataMatcher.getService().submit(new MusicBrainzMetadataMatcher(songGroup));
}
else if(UserPreferences.getInstance().isSearchDiscogs())
{
if(songGroup.getSubSongGroups().size() > 1)
{
DiscogsMultiFolderSongGroupMatcher.getService().submit(new DiscogsMultiFolderSongGroupMatcher(songGroup));
}
else if(songGroup.getSongIds().size()==1)
{
DiscogsSongMatcher.getService().submit(new DiscogsSongMatcher(songGroup, songGroup.getSongIds().get(0)));
}
else
{
DiscogsSongGroupMatcher.getService().submit(new DiscogsSongGroupMatcher(songGroup));
}
}
else
{
for (Integer songId : songGroup.getSongIds())
{
SongSaver.getService().submit(new SongSaver(songId));
}
}
return true;
}
/**
* Process a group of files that are in a Random folder and dont seem to have anything in common so should not be grouped
* together.
*
* @param songGroup
* @param songs
* @return
*/
private boolean processRandomFolderNoMetadata(SongGroup songGroup, List<Song> songs)
{
if(UserPreferences.getInstance().isSearchMusicBrainz())
{
for (Song song : songs)
{
MusicBrainzSongMatcher.getService().submit(new MusicBrainzSongMatcher(songGroup, song.getRecNo()));
}
}
else if(UserPreferences.getInstance().isSearchDiscogs())
{
for (Song song : songs)
{
DiscogsSongMatcher.getService().submit(new DiscogsSongMatcher(songGroup, song.getRecNo()));
}
}
else
{
for (Integer songId : songGroup.getSongIds())
{
SongSaver.getService().submit(new SongSaver(songId));
}
}
return true;
}
}
最佳答案
我认为这很愚蠢但找不到它，但问题是我将任务提交给了错误的ExecutorService
在 AcoustidMatcher 我有
private boolean processFolderWithPoorMetadata(SongGroup songGroup)
{
MusicBrainzMetadataMatcher.getService().submit(new SongPrematcherMatcher(songGroup));
return true;
}
当我应该拥有
private boolean processFolderWithPoorMetadata(SongGroup songGroup)
{
SongPrematcherMatcher.getService().submit(new SongPrematcherMatcher(songGroup));
return true;
}
所以这意味着所有 SongPrematcher 任务都提交给了错误的 ExecutorService，所以当我请求关闭 SongPrematcher ExecutorService 时，它可以立即关闭!
关于java - 为什么 ExecutorService.awaitTermination() 在提交的任务完成之前成功，我们在Stack Overflow上找到一个类似的问题： https://stackoverflow.com/questions/43693141/