Java读&写锁要求，具有锁和从不同线程释放 [英] Java read & write lock requirement, with lock and release from different threads

问题描述

我试图找到一个不太笨重的解决方案来解决Java并发问题。

I'm trying to find a less clunky solution to a Java concurrency problem.

问题的要点是我需要一个关闭调用来阻止仍然是工作线程活动的，但关键的方面是工作者任务都是异步生成和完成的，因此保持和释放必须由不同的线程完成。我需要他们以某种方式发送信号到关闭线程一旦他们的工作完成。只是为了使事情更有趣，工作线程不能互相阻止，所以我不确定在这个特定的实例中的信号量的应用。

The gist of the problem is that I need a shutdown call to block while there are still worker threads active, but the crucial aspect is that the worker tasks are each spawned and completed asynchronously so the hold and release must be done by different threads. I need them to somehow send a signal to the shutdown thread once their work has completed. Just to make things more interesting, the worker threads cannot block each other so I'm unsure about the application of a Semaphore in this particular instance.

我有一个解决方案我认为安全地做这项工作，但我不熟悉Java并发工具使我认为可能有一个更容易或更优雅的模式。

I have a solution which I think safely does the job, but my unfamiliarity with the Java concurrency utils leads me to think that there might be a much easier or more elegant pattern. Any help in this regard would be greatly appreciated.

这里是我迄今为止，相当稀疏，除了评论：

Here's what I have so far, fairly sparse except for the comments:

final private ReentrantReadWriteLock shutdownLock = new ReentrantReadWriteLock();
volatile private int activeWorkerThreads;
private boolean isShutdown;

private void workerTask()
{
   try
   {
      // Point A: Worker tasks mustn't block each other.
      shutdownLock.readLock().lock();

      // Point B: I only want worker tasks to continue if the shutdown signal
      // hasn't already been received.
      if (isShutdown)
         return;

      activeWorkerThreads ++;

      // Point C: This async method call returns immediately, soon after which
      // we release our lock. The shutdown thread may then acquire the write lock
      // but we want it to continue blocking until all of the asynchronous tasks
      // have completed.
      executeAsynchronously(new Runnable()
      {
         @Override
         final public void run()
         {
            try
            {
              // Do stuff.
            }
            finally
            {
               // Point D: Release of shutdown thread loop, if there are no other
               // active worker tasks.
               activeWorkerThreads --;
            }
         }
      });
   }
   finally
   {
      shutdownLock.readLock().unlock();
   }
}


final public void shutdown()
{
   try
   {
      // Point E: Shutdown thread must block while any worker threads
      // have breached Point A.
      shutdownLock.writeLock().lock();

      isShutdown = true;

      // Point F: Is there a better way to wait for this signal?
      while (activeWorkerThreads > 0)
         ;

      // Do shutdown operation.
   }
   finally
   {
      shutdownLock.writeLock().unlock();
   }
}

感谢您提供任何帮助！

Russ

推荐答案

您可以在这种情况下使用信号量，等待shutdown（）调用。认为它的方式是作为一组票，发给工人，以表明他们在飞行。如果shutdown（）方法可以获取所有票据，那么它知道它已经耗尽所有工作者，并且没有活动。因为#acquire（）是一个阻塞调用，shutdown（）不会旋转。我已经使用这种方法为一个分布式的主 - 工人库，它很容易扩展它处理超时和重试。

You can use a semaphore in this scenario and not require a busy wait for the shutdown() call. The way to think of it is as a set of tickets that are handed out to workers to indicate that they are in-flight. If the shutdown() method can acquire all of the tickets then it knows that it has drained all workers and there is no activity. Because #acquire() is a blocking call the shutdown() won't spin. I've used this approach for a distributed master-worker library and its easy extend it to handle timeouts and retrials.

Executor executor = // ...
final int permits = // ...
final Semaphore semaphore = new Semaphore(permits);

void schedule(final Runnable task) {
  semaphore.acquire();
  try {
    executor.execute(new Runnable() {
      @Override public run() {
        try {
          task.run();
        } finally {
          semaphore.release();
        }
      }
    });
  } catch (RejectedExecutionException e) {
    semaphore.release();
    throw e;
  }
}

void shutDown() {
  semaphore.acquireUninterruptibly(permits);

  // do stuff
}

这篇关于Java读&写锁要求，具有锁和从不同线程释放的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！