29

有谁知道是否有任何闩锁实现可以执行以下操作：

  • 有一种方法来减少锁存器的值，或者如果值为零则等待
  • 有等待锁存值为零的方法
  • 有一种将数字添加到锁存器值的方法
4

6 回答 6

83

您还可以使用Phaser (java.util.concurrent.Phaser)

final Phaser phaser = new Phaser(1); // register self
while (/* some condition */) {
    phaser.register(); // Equivalent to countUp
    // do some work asynchronously, invoking
    // phaser.arriveAndDeregister() (equiv to countDown) in a finally block
}
phaser.arriveAndAwaitAdvance(); // await any async tasks to complete

我希望这有帮助。

于 2015-09-07T16:20:20.323 回答
9

java.util.concurrent.Semaphore似乎符合要求。

  • 获取（）或获取（n）
  • 也获取（）（不确定我理解这里有什么区别）（*）
  • 释放（）或释放（n）

(*) 好的，没有内置方法可以等到信号量变得不可用。我想您首先会编写自己的包装器acquire，tryAcquire如果失败会触发您的“忙碌事件”（并继续使用 normal acquire）。每个人都需要调用你的包装器。也许是信号量的子类？

于 2013-01-10T09:57:02.993 回答
7

您可以使用如下所示的简单实现，而不是从 AQS 开始。它有点幼稚（它是同步的与 AQS 无锁算法），但除非您希望在满足的场景中使用它，否则它可能已经足够好了。

public class CountUpAndDownLatch {
    private CountDownLatch latch;
    private final Object lock = new Object();

    public CountUpAndDownLatch(int count) {
        this.latch = new CountDownLatch(count);
    }

    public void countDownOrWaitIfZero() throws InterruptedException {
        synchronized(lock) {
            while(latch.getCount() == 0) {
                lock.wait();
            }
            latch.countDown();
            lock.notifyAll();
        }
    }

    public void waitUntilZero() throws InterruptedException {
        synchronized(lock) {
            while(latch.getCount() != 0) {
                lock.wait();
            }
        }
    }

    public void countUp() { //should probably check for Integer.MAX_VALUE
        synchronized(lock) {
            latch = new CountDownLatch((int) latch.getCount() + 1);
            lock.notifyAll();
        }
    }

    public int getCount() {
        synchronized(lock) {
            return (int) latch.getCount();
        }
    }
}

注意：我尚未对其进行深入测试，但它似乎表现得如预期：

public static void main(String[] args) throws InterruptedException {
    final CountUpAndDownLatch latch = new CountUpAndDownLatch(1);
    Runnable up = new Runnable() {
        @Override
        public void run() {
            try {
                System.out.println("IN UP " + latch.getCount());
                latch.countUp();
                System.out.println("UP " + latch.getCount());
            } catch (InterruptedException ex) {
            }
        }
    };

    Runnable downOrWait = new Runnable() {
        @Override
        public void run() {
            try {
                System.out.println("IN DOWN " + latch.getCount());
                latch.countDownOrWaitIfZero();
                System.out.println("DOWN " + latch.getCount());
            } catch (InterruptedException ex) {
            }
        }
    };

    Runnable waitFor0 = new Runnable() {
        @Override
        public void run() {
            try {
                System.out.println("WAIT FOR ZERO " + latch.getCount());
                latch.waitUntilZero();
                System.out.println("ZERO " + latch.getCount());
            } catch (InterruptedException ex) {
            }
        }
    };
    new Thread(waitFor0).start();
    up.run();
    downOrWait.run();
    Thread.sleep(100);
    downOrWait.run();
    new Thread(up).start();
    downOrWait.run();
}

输出：

IN UP 1
UP 2
WAIT FOR ZERO 1
IN DOWN 2
DOWN 1
IN DOWN 1
ZERO 0
DOWN 0
IN DOWN 0
IN UP 0
DOWN 0
UP 0
于 2013-01-10T14:57:05.103 回答
3

对于那些需要基于 AQS 的解决方案的人来说，这对我有用：

public class CountLatch {

    private class Sync extends AbstractQueuedSynchronizer {
        private static final long serialVersionUID = 1L;

        public Sync() {
        }

        @Override
        protected int tryAcquireShared(int arg) {
            return count.get() == releaseValue ? 1 : -1;
        }

        @Override
        protected boolean tryReleaseShared(int arg) {
            return true;
        }
    }

    private final Sync sync;
    private final AtomicLong count;
    private volatile long releaseValue;

    public CountLatch(final long initial, final long releaseValue) {
        this.releaseValue = releaseValue;
        this.count = new AtomicLong(initial);
        this.sync = new Sync();
    }

    public void await() throws InterruptedException {
        sync.acquireSharedInterruptibly(1);
    }

    public long countUp() {
        final long current = count.incrementAndGet();
        if (current == releaseValue) {
            sync.releaseShared(0);
        }
        return current;
    }

    public long countDown() {
        final long current = count.decrementAndGet();
        if (current == releaseValue) {
            sync.releaseShared(0);
        }
        return current;
    }

    public long getCount() {
        return count.get();
    }
}

您使用初始值和目标值初始化同步器。一旦达到目标值（通过向上和/或向下计数），等待的线程将被释放。

于 2015-06-05T07:51:41.003 回答
0

我需要一个并使用与使用 AQS（非阻塞）的 CountDownLatch 相同的策略来构建它，这个类也与为 Apache Camel 创建的一个非常相似（如果不准确的话），我认为它也比 JDK Phaser 更轻，这个就像 JDK 中的 CountDownLact 一样，它不会让您在零以下倒数，并允许您倒数和倒数：

导入 java.util.concurrent.TimeUnit；导入 java.util.concurrent.locks.AbstractQueuedSynchronizer；

public class CountingLatch
{
  /**
   * Synchronization control for CountingLatch.
   * Uses AQS state to represent count.
   */
  private static final class Sync extends AbstractQueuedSynchronizer
  {
    private Sync()
    {
    }

    private Sync(final int initialState)
    {
      setState(initialState);
    }

    int getCount()
    {
      return getState();
    }

    protected int tryAcquireShared(final int acquires)
    {
      return getState()==0 ? 1 : -1;
    }

    protected boolean tryReleaseShared(final int delta)
    {
      // Decrement count; signal when transition to zero
      for(; ; ){
        final int c=getState();
        final int nextc=c+delta;
        if(nextc<0){
          return false;
        }
        if(compareAndSetState(c,nextc)){
          return nextc==0;
        }
      }
    }
  }

  private final Sync sync;

  public CountingLatch()
  {
    sync=new Sync();
  }

  public CountingLatch(final int initialCount)
  {
    sync=new Sync(initialCount);
  }

  public void increment()
  {
    sync.releaseShared(1);
  }

  public int getCount()
  {
    return sync.getCount();
  }

  public void decrement()
  {
    sync.releaseShared(-1);
  }

  public void await() throws InterruptedException
  {
    sync.acquireSharedInterruptibly(1);
  }

  public boolean await(final long timeout) throws InterruptedException
  {
    return sync.tryAcquireSharedNanos(1,TimeUnit.MILLISECONDS.toNanos(timeout));
  }
}
于 2013-09-02T16:07:05.430 回答
0

这是CounterLatchApache 站点上的一个变体。

他们的版本，出于他们自己最熟悉的原因，在变量 ( ) 处于给定值 时阻塞调用者线程。AtomicInteger

但是调整此代码非常容易，这样您就可以选择 Apache 版本的功能，或者......说“在这里等到计数器达到某个值”。可以说，后者将具有更多的适用性。在我的特殊情况下，我把它弄得沙沙作响，因为我想检查所有“块”是否已经在SwingWorker.process()……上发表过，但我后来找到了它的其他用途。

在这里，它是用 Jython 编写的，这是世界上最好的语言 (TM)。我将在适当的时候推出一个 Java 版本。

class CounterLatch():
    def __init__( self, initial = 0, wait_value = 0, lift_on_reached = True ):
        self.count = java.util.concurrent.atomic.AtomicLong( initial )
        self.signal = java.util.concurrent.atomic.AtomicLong( wait_value )

        class Sync( java.util.concurrent.locks.AbstractQueuedSynchronizer ):
            def tryAcquireShared( sync_self, arg ):
                if lift_on_reached:
                    return -1 if (( not self.released.get() ) and self.count.get() != self.signal.get() ) else 1
                else:
                    return -1 if (( not self.released.get() ) and self.count.get() == self.signal.get() ) else 1
            def tryReleaseShared( self, args ):
                return True

        self.sync = Sync()
        self.released = java.util.concurrent.atomic.AtomicBoolean() # initialised at False

    def await( self, *args ):
        if args:
            assert len( args ) == 2
            assert type( args[ 0 ] ) is int
            timeout = args[ 0 ]
            assert type( args[ 1 ] ) is java.util.concurrent.TimeUnit
            unit = args[ 1 ]
            return self.sync.tryAcquireSharedNanos(1, unit.toNanos(timeout))
        else:
            self.sync.acquireSharedInterruptibly( 1 )

    def count_relative( self, n ):
        previous = self.count.addAndGet( n )
        if previous == self.signal.get():
            self.sync.releaseShared( 0 )
        return previous

注意 Apache 版本使用关键字volatileforsignal和released。在 Jython 中，我认为这并不存在，但使用AtomicIntegerandAtomicBoolean应该确保在任何线程中没有值“过时”。

示例用法：

在 SwingWorker 构造函数中：

self.publication_counter_latch = CounterLatch()

在 SW.publish 中：

# increase counter value BEFORE publishing chunks
self.publication_counter_latch.count_relative( len( chunks ) )
self.super__publish( chunks )

在 SW.process 中：

# ... do sthg [HERE] with the chunks!
# AFTER having done what you want to do with your chunks:
self.publication_counter_latch.count_relative( - len( chunks ) )

在等待块处理停止的线程中：

worker.publication_counter_latch.await()
于 2016-03-01T22:05:41.653 回答