JUC 中的多线程协作工具类：CountDownLatch 和 CyclicBarrier

运行改造后的 producer 方法，结果如下：

引入 CountDownLatch 之后，我们的消息就发送正常了。

提到 CountDownLatch，那么肯定会联想到 CyclicBarrier 类。

CyclicBarrier 类称作循环屏障，跟 CountDownLatch 的功能很相似，但是又不太一样，主要有以下几种区别：

• 工作原理不一样，CountDownLatch 是使一批线程等待另一批线程执行完之后再执行，而 CyclicBarrier 是当等待的线程达到一定数目后，就开始继续工作。
• CyclicBarrier 是可以循环使用的，而 CountDownLatch 类不一样。
• 服务对象不一样，CountDownLatch 是等待后，其他线程继续执行，而 CyclicBarrier 是等待线程数达到一定屏障后，自己继续执行。就好比吃饭，等人到齐后，才开始吃，等的人是你，吃饭的人还是你。
• 实现不一样，CountDownLatch 是基于 AQS 实现的，CyclicBarrier 是基于 ReentrantLock 实现的。

CyclicBarrier 类的源代码也不是太多，就不贴了，有兴趣的可以研究研究。直接实战一把。在原有的异步消息中，添加一个功能，当所有的消息发送完成后，每个消息上报自己消息发送的情况。代码如下：

public class AsySendProducer {

    public static void main(String[] args) throws MQClientException, UnsupportedEncodingException, RemotingException, InterruptedException {

 DefaultMQProducer producer = new DefaultMQProducer("test-producer-group");
 // 设置nameserver 地址
 producer.setNamesrvAddr("xxxxx:9876");
 //启动生产者
 producer.start();
 System.out.println("-------------启动服务--------------");
 // 设置异步发送失败重试次数，默认为2
 producer.setRetryTimesWhenSendAsyncFailed(0);

 // 实例化一个 countDownLatch 实例，我们需要发送 10条消息，所以计数器为 10.
 CountDownLatch countDownLatch = new CountDownLatch(5);
		// 实例化 CyclicBarrier 
 CyclicBarrier cyclicBarrier = new CyclicBarrier(5);

 for (int i = 0; i < 5; i++) {//
     final int index = i;
     // ID110：业务数据的ID，比如用户ID、订单编号等等
     Message msg = new Message("TopicTest",
      "TagA",
      "ID110",
      ("Hello RocketMQ " +
i).getBytes(RemotingHelper.DEFAULT_CHARSET) 
     );
     // 不管消息发送成功还是失败，CountDownLatch 中的计数器都需要减一
     producer.send(msg, new SendCallback() {
  
  @Override
  public void onSuccess(SendResult sendResult) {
      System.out.printf(Thread.currentThread().getName() + " %-10d OK %s%n", index, sendResult.getMsgId());
      countDownLatch.countDown();
      try {
	      // 等待所有线程执行完
   cyclicBarrier.await();
   System.out.println(Thread.currentThread().getName()+"：我执行成功！");
      } catch (InterruptedException e) {
   e.printStackTrace();
      } catch (BrokenBarrierException e) {
   e.printStackTrace();
      }
  }

  
  @Override
  public void onException(Throwable e) {
      System.out.printf(Thread.currentThread().getName() + " %-10d Exception %s %n", index, e);
      countDownLatch.countDown();
      try {
	      // 等待所有线程执行完
   cyclicBarrier.await();
   System.out.println(Thread.currentThread().getName()+"：我执行失败！");
      } catch (InterruptedException ex) {
   ex.printStackTrace();
      } catch (BrokenBarrierException ex) {
   ex.printStackTrace();
      }

  }
     });

 }
 //关闭生产者连接
 countDownLatch.await();
 producer.shutdown();
 System.out.println("--------------关闭服务器-----------");

    }
}