java - 为什么 ExecutorService.awaitTermination() 在提交的任务完成之前成功

Question

在我的代码中，我有许多作为管道运行的 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;
            }
        }

Answer 1

我认为这很愚蠢但找不到它，但问题是我将任务提交给了错误的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 时，它​​可以立即关闭!