标签: countdownlatch

有人可以帮我理解Java CountDownLatch是什么以及何时使用它？

我对这个程序的工作原理并不十分清楚.据我所知,所有三个线程立即启动,每个线程将在3000ms后调用CountDownLatch.倒数会逐一减少.在锁存器变为零之后,程序打印"已完成".也许我理解的方式不正确.

import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

class Processor implements Runnable {
    private CountDownLatch latch;

    public Processor(CountDownLatch latch) {
        this.latch = latch;
    }

    public void run() {
        System.out.println("Started.");

        try {
            Thread.sleep(3000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

        latch.countDown();
    }
}

// ------------------------------------------------ -----

public class App {

    public static void main(String[] args) {

        CountDownLatch latch = new CountDownLatch(3); // coundown from 3 to 0

        ExecutorService executor = Executors.newFixedThreadPool(3); // 3 Threads in pool

        for(int i=0; i …

我正在阅读java.util.concurrent API,并发现了

• CountDownLatch:允许一个或多个线程等待直到在其他线程中执行的一组操作完成的同步辅助.
• CyclicBarrier:一种同步辅助,允许一组线程全部等待彼此到达公共障碍点.

对我来说,两者似乎都是平等的,但我相信还有更多的东西.

例如,在CoundownLatch, the countdown value could not be reset, that can happen in the case of CyclicBarrier.

这两者之间还有其他区别吗？
有什么use cases地方会有人想重置倒计时的价值？

使用有什么好处

java.util.concurrent.CountdownLatch

代替

java.util.concurrent.Semaphore？

据我所知,以下片段几乎相同:

1.信号量

final Semaphore sem = new Semaphore(0);
for (int i = 0; i < num_threads; ++ i)
{
  Thread t = new Thread() {
    public void run()
    {
      try
      {
        doStuff();
      }
      finally
      {
        sem.release();
      }
    }
  };
  t.start();
}

sem.acquire(num_threads);

2:CountDownLatch

final CountDownLatch latch = new CountDownLatch(num_threads);
for (int i = 0; i < num_threads; ++ i)
{
  Thread t = new Thread() {
    public void run()
    {
      try
      {
        doStuff();
      }
      finally …

我现在遇到了两次问题,即生产者线程生成N个工作项,将它们提交给a ExecutorService,然后需要等到所有N个项都被处理完毕.

注意事项

• N事先不知道.如果是这样的话,我只需创建一个CountDownLatch然后拥有生产者线程,await()直到所有工作完成.
• 使用a CompletionService是不合适的,因为虽然我的生产者线程需要阻塞(即通过调用take()),但是没有办法表明所有工作都已完成,导致生产者线程停止等待.

我目前最喜欢的解决方案是使用整数计数器,并在提交工作项时递增它,并在处理工作项时递减它.在所有N个任务的提交之后,我的生产者线程将需要等待锁定,检查是否counter == 0通知它.如果消费者线程已经递减计数器并且新值为0,则消费者线程将需要通知生产者.

有没有更好的方法解决这个问题,或者java.util.concurrent我应该使用合适的构造而不是"滚动自己的"？

提前致谢.

我在多线程环境中经历了不同的并发模型(http://tutorials.jenkov.com/java-concurrency/concurrency-models.html)

本文重点介绍了三种并发模型.

1. 并行工人

   第一个并发模型就是我所说的并行工作模型.传入的工作分配给不同的工作人员.

2. 流水线

   工人们正在举办类似的工人在工厂的流水线.每个工人只执行完整工作的一部分.当该部分完成时,工人将工作转发给下一个工人.

   每个工作者都在自己的线程中运行,并且不与其他工作者共享任何状态.这有时也称为无共享并发模型.

3. 功能并行

   函数并行的基本思想是使用函数调用实现程序.函数可以被视为彼此发送消息的" 代理 "或" 参与者 ",就像在流水线并发模型(AKA反应或事件驱动系统)中一样.当一个函数调用另一个函数时,这类似于发送消息.

现在我想为这三个概念映射java API支持

1. 并行工作者:它是ExecutorService,ThreadPoolExecutor,CountDownLatch API吗？

2. 装配线:将事件发送到JMS等消息传递系统并使用队列和主题的消息传递概念.

3. 功能并行:ForkJoinPool在某种程度上和java 8流.与溪流相比,ForkJoin池很容易理解.

我是否正确映射这些并发模型？如果没有,请纠正我.

在等待其他线程完成时,我们可以使用join或者CountdownLatch.使用这两种机制中的任何一种的利弊是什么？

是否有一个C#等同于Java的CountDownLatch？

我需要的东西直接等同于CountDownLatch,但可以重置(保持线程安全!).我不能使用经典的同步结构,因为它们在这种情况下根本不起作用(复杂的锁定问题).目前,我正在创建许多CountDownLatch对象,每个对象都替换前一个对象.我相信这是在GC中的年轻一代(由于物体数量庞大).您可以看到使用下面的锁存器的代码(它java.net是ns-3网络模拟器接口的模拟的一部分).

一些想法可能是尝试CyclicBarrier(JDK5 +)或Phaser(JDK7)

我可以测试代码并回到找到解决此问题的任何人,因为我是唯一可以将其插入正在运行的系统中以查看发生的情况的人:)

/**
 *
 */
package kokunet;

import java.io.IOException;
import java.nio.channels.ClosedSelectorException;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import kokuks.IConnectionSocket;
import kokuks.KKSAddress;
import kokuks.KKSSocket;
import kokuks.KKSSocketListener;

/**
 * KSelector
 * @version 1.0
 * @author Chris Dennett
 */
public class KSelector extends SelectorImpl {
    // True if this Selector has been closed
    private volatile boolean closed = false;

    // Lock for close and cleanup
    final class CloseLock {} …

以下是两个完成(我认为)同样事情的代码块.

我基本上是在尝试学习如何使用Java 1.5的并发来摆脱Thread.sleep(long).第一个示例使用ReentrantLock,第二个示例使用CountDownLatch.我想要做的就是让一个线程进入休眠状态,直到另一个线程中的条件得到解决.

ReentrantLock提供了一个布尔锁定,我用来决定是否唤醒另一个线程,然后我使用条件和等待/信号来休眠另一个线程.据我所知,我需要使用锁的唯一原因是,如果多个线程需要对boolean进行写访问.

CountDownLatch似乎提供与ReentrantLock相同的功能,但没有(不必要的？)锁.然而,感觉就像我通过初始化它只需要一次倒计时来劫持它的预期用途.我认为它应该在多个线程要处理同一个任务时使用,而不是在多个线程在等待一个任务时使用.

所以,问题:

1. 我在ReentrantLock代码中使用锁定"正确的东西"吗？如果我只在一个线程中写入布尔值,那么锁是否必要？只要我在唤醒任何其他线程之前重置布尔值我不会导致问题,我可以吗？

2. 是否有一个类似于CountDownLatch的类,我可以使用它来避免锁定(假设我应该在这个实例中避免它们),这更适合这个任务？

3. 有没有其他方法来改进我应该注意的代码？

例1:

import java.util.concurrent.locks.*;

public class ReentrantLockExample extends Thread {

//boolean - Is the service down?
boolean serviceDown;

// I am using this lock to synchronize access to sDown
Lock serviceLock; 
// and this condition to sleep any threads waiting on the service.
Condition serviceCondition;

public static void main(String[] args) {
    Lock l = new ReentrantLock();
    Condition c = l.newCondition(); 
    ReentrantLockExample rle = new ReentrantLockExample(l, c);

    //Imagine this thread figures out the …

我正在尝试测试一个在单独的线程中工作的方法,简化它是这样的:

public void methodToTest()
{
    Thread thread = new Thread()
    {
        @Override
        public void run() {
            Clazz.i = 2;
        }
    };
    thread.start();
}

在我的单元测试中,我想测试Clazz.i == 2,但我不能这样做,因为我认为断言是在线程更改值之前运行的.我想使用另一个线程来测试它,然后使用join等待,但它仍然无法正常工作.

SSCCE:

@Test
public void sscce() throws InterruptedException
{
    Thread thread = new Thread()
    {
        @Override
        public void run() {
            methodToTest()
        }
    };
    thread.start();     
    thread.join();  
    AssertEquals(2, Clazz.i);
}

public static class Clazz
{
    public static int i = 0;
}

我认为这是因为测试主代码创​​建了一个等待(加入)第二个线程的线程,但是第二个线程没有完成工作,它创建另一个线程来完成工作然后完成,这继续第一个线程,而第三个线程Clazz.i = 2在断言后执行.

我怎样才能使第一个线程等待它启动的线程以及该线程启动的任何线程？