要在异步调用上实现优雅关闭检查锁定，或处理异常？

Question

我正在开发一个 Java 应用程序，在大部分时间（包括关闭时）都必须处理来自外部框架的大量传入异步调用。在正常操作期间，这些传入调用需要再次异步分派到另一个框架。

目前，我让我的模块成为一个“好”公民，并围绕关闭标志进行一些锁定，该标志一旦设置，将优雅地停止发送任何进一步的传出呼叫。

令人不安的是，因为传入和传出调用都是异步的，所以我必须使每个“工作人员”任务执行两组锁定（见下文）以执行相同的关闭标志检查（编辑 ：有人在另一个问题中向我指出，使用信号量每个工作人员只需要一次获取/释放）。它有效，但有许多工作任务需要处理，我担心性能的累积下降。一旦框架稍微扩展一下，分析就会很快出现，但无论结果如何，遵循最佳实践都是很好的。

另一种方法是简单地不进行关闭标志检查锁定，并处理在异步调用完成处理之前关闭外部框架时生成的预期异常。我应该补充一点，如果采取这种方法，不会对运营产生不利影响。两种方法都会导致干净关闭。

请问您认为哪种做法更好？没有例外的严厉锁定，与没有锁定但有一系列例外的情况不同。

使用锁，工作任务代码看起来像这样：

final private ReentrantReadWriteLock shutdownLock = new ReentrantReadWriteLock();
private boolean isShutdown;

private void workerTask()
{
   try
   {
      shutdownLock.readLock().lock();

      if (isShutdown)
         return;

      executeAsynchronously(new Runnable()
      {
         @Override
         final public void run()
         {
            try
            {
               shutdownLock.readLock().lock();

               if (isShutdown)
                  return;

               // Do something here.
            }
            finally
            {
               shutdownLock.readLock().unlock();
            }
         }
      });
   }
   finally
   {
      shutdownLock.readLock().unlock();
   }
}

shutdown 方法请求 shutdownLock.writeLock()，然后设置 isShutdown 标志。

不锁定和预期关闭生成的异常的替代方案如下所示：

volatile private boolean isShutdown;

private void workerTask()
{
   try
   {
      executeAsynchronously(new Runnable()
      {
         @Override
         final public void run()
         {
            try
            {
               // Do something here.
            }
            catch (final FrameworkRuntimeException exception)
            {
               if ((! isShutdown) || (exception.type != 
                                      FrameworkRuntimeException.FrameworkIsDisposed))
                  throw exception;
            }
         }
      });
   }
   catch (final FrameworkRuntimeException exception)
   {
      if ((! isShutdown) || (exception.type != 
                             FrameworkRuntimeException.FrameworkIsDisposed))
         throw exception;
   }
}

此实现的关闭方法将 易失性 isShutdown 标志设置为 true。

预先感谢您的任何反馈，

Russ

编辑：在另一个问题中有人向我指出，我可以使用信号量来避免第一种方法中的双重锁定，所以情况不会如此终究是严厉的，但问题仍然存在。

Answer 1

一般来说，我更喜欢检查关闭然后执行任务的方法。如果您乐观地丢弃您“知道”由于关闭而导致的异常，那么您将面临错误分类错误并错过真正问题的风险。

就简化代码而言，您可以摆脱所有锁定，只需确保您的 executeAsynchronously 方法使用 ExecutorService - 然后您的 shutdown 方法只是在服务上调用 shutdown ，如果 isShutdown 返回 true 则可以跳过任务创建，如果您需要等待任务完成后再返回，您可以使用有帮助的awaitTermination 方法。

Answer 2

好吧，我认为这里应该可以工作，并且不会妨碍太多的运行时开销（注意这里是凌晨 4.30，所以最好仔细检查一下;)）。另请注意，添加一些好的 try{}finally{} 代码块将是一个非常好的主意，但为了清楚起见，将其省略。

public static final AtomicInteger activeConnections = new AtomicInteger();
public static volatile boolean shutdown = false;

public static void shutdown() {
    shutdown = true;
    while (activeConnections.get() > 0) {
        synchronized(activeConnections) {
            try {
                activeConnections.wait();
            }
            catch(InterruptedException e) {
            }
        }
    }
    // proceed shutdown
}

public static void run() {
    if (shutdown) return;
    activeConnections.incrementAndGet();
    if (shutdown) {
        leave();
        return;
    }
    // do stuff
    leave();
}

private static void leave() {
    int outstandingConnections = activeConnections.decrementAndGet();
    if (shutdown && outstandingConnections == 0) {
        synchronized(activeConnections) {
            activeConnections.notifyAll();
        }
    }       
}

一旦设置了关闭标志，就没有新线程开始工作。每个线程在与外部框架通信时都会递增一个整数，并在完成时递减它。只有当没有线程再进行通信时，关闭才会继续 - 请注意，由于首先设置了关闭标志，因此不会再启动新线程。

这样你就可以获得相当轻量级的 AtomicInteger（它是作为 CAS 循环实现的，你无法获得更低的开销）和易失性内存屏障。

现在我仍然坚持我的第一条评论，并说捕获异常更简单、更高效、更短，但我喜欢这个问题:)

Answer 3

如果您使用 spring，下面的代码可以正常关闭。您可以更改重试次数。

package com.avea.vpspg.test.schedulers;

import java.util.Map;
import java.util.concurrent.TimeUnit;

import org.apache.log4j.Logger;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.config.BeanPostProcessor;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.context.ApplicationListener;
import org.springframework.context.event.ContextClosedEvent;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;
import org.springframework.stereotype.Component;

import com.avea.vpspg.core.VasProvLogger;

@Component
class ContextClosedHandler implements ApplicationListener<ContextClosedEvent> , ApplicationContextAware,BeanPostProcessor{


private ApplicationContext context;

public Logger logger = XProvLogger.getInstance().x;

public void onApplicationEvent(ContextClosedEvent event) {


    Map<String, ThreadPoolTaskScheduler> schedulers = context.getBeansOfType(ThreadPoolTaskScheduler.class);

    for (ThreadPoolTaskScheduler scheduler : schedulers.values()) {         
        scheduler.getScheduledExecutor().shutdown();
        try {
            scheduler.getScheduledExecutor().awaitTermination(20000, TimeUnit.MILLISECONDS);
            if(scheduler.getScheduledExecutor().isTerminated() || scheduler.getScheduledExecutor().isShutdown())
                logger.info("Scheduler "+scheduler.getThreadNamePrefix() + " has stoped");
            else{
                logger.info("Scheduler "+scheduler.getThreadNamePrefix() + " has not stoped normally and will be shut down immediately");
                scheduler.getScheduledExecutor().shutdownNow();
                logger.info("Scheduler "+scheduler.getThreadNamePrefix() + " has shut down immediately");
            }
        } catch (IllegalStateException e) {
            e.printStackTrace();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    Map<String, ThreadPoolTaskExecutor> executers = context.getBeansOfType(ThreadPoolTaskExecutor.class);

    for (ThreadPoolTaskExecutor executor: executers.values()) {
        int retryCount = 0;
        while(executor.getActiveCount()>0 && ++retryCount<51){
            try {
                logger.info("Executer "+executor.getThreadNamePrefix()+" is still working with active " + executor.getActiveCount()+" work. Retry count is "+retryCount);
                Thread.sleep(1000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
        if(!(retryCount<51))
            logger.info("Executer "+executor.getThreadNamePrefix()+" is still working.Since Retry count exceeded max value "+retryCount+", will be killed immediately");
        executor.shutdown();
        logger.info("Executer "+executor.getThreadNamePrefix()+" with active " + executor.getActiveCount()+" work has killed");
    }
}


@Override
public void setApplicationContext(ApplicationContext context)
        throws BeansException {
    this.context = context;

}


@Override
public Object postProcessAfterInitialization(Object object, String arg1)
        throws BeansException {
    return object;
}


@Override
public Object postProcessBeforeInitialization(Object object, String arg1)
        throws BeansException {
    if(object instanceof ThreadPoolTaskScheduler)
        ((ThreadPoolTaskScheduler)object).setWaitForTasksToCompleteOnShutdown(true);
    if(object instanceof ThreadPoolTaskExecutor)
        ((ThreadPoolTaskExecutor)object).setWaitForTasksToCompleteOnShutdown(true);
    return object;
}