首先简单介绍CountDownLatch 的使用：同步工具类
(1) 概念CountDownLatch可以使一个或多个线程等待其他线程各自执行完毕后再执行。
CountDownLatch 定义了一个计数器，和一个阻塞队列， 当计数器的值递减为0之前，阻塞队列里面的线程处于挂起状态，当计数器递减到0时会唤醒阻塞队列所有线程，这里的计数器是一个标志，可以表示一个任务一个线程，也可以表示一个倒计时器，CountDownLatch可以解决那些一个或者多个线程在执行之前必须依赖于某些必要的前提业务先执行的场景。
(2) 方法- CountDownLatch(int count)：构造方法，创建一个值为count 的计数器。
- await()：阻塞当前线程，将当前线程加入阻塞队列。
- await(long timeout, TimeUnit unit)：在timeout的时间之内阻塞当前线程,时间一过则当前线程可以执行，
- countDown()：对计数器进行递减1操作，当计数器递减至0时，当前线程会去唤醒阻塞队列里的所有线程。
讲解：开辟map个线程，开启多线程进行数据的批量入库。但是由于，这里batchInsertMembers是单例，导致数据不安全。
其实：单例和多线程没有直接关系。当然，你多个线程去访问单例对象其实是一个共享资源。所以，你的单例对象要保证是线程安全的。任何线程对单例对象的访问都需要线程安全，你的单例类中需要对一些共享访问进行互斥
AtomicInteger executeNum = new AtomicInteger(0);
Map> map = new HashMap<>();
map.put(1, Lists.newArrayList("a1","b1","c1"));
map.put(2, Lists.newArrayList("a2","b2","c2"));
CountDownLatch latch = new CountDownLatch(map.size());
for (Map.Entry> entry : map.entrySet()) {
ThreadUtil.getDefaultExecutor().execute(() -> {
executeNum.updateAndGet(v -> v + batchInsertMembers(entry.getValue()));
latch.countDown();
});;
}
try {
latch.await();
}
catch (InterruptedException e) {
}
batchInsertMembers方法
public static Integer batchInsertMembers(Listparam) { int i = 0; while (i < param.size()) { if ((i + 1) < param.size()) { System.out.println(param.subList(i, i + 1)); } else { System.out.println(param.subList(i, param.size())); } i += 1; System.out.println("i------" + i); } return param.size(); }
我们来看一下运行结果：
正常情况下，打印两个i------3之后，线程就应该结束了，但是真实情况并不是这样，所以这里多个线程去访问单例对象相当于共享资源，导致了数据不安全。
[a1] [a2] i------1 i------1 [b1] i------2 [b2] i------2 [c1] [c2] i------3 i------3 [a2] i------1 [b2] i------2 [c2] i------3二：下面我们把方法稍微做一下改造：
每次都去new TestThread(entry.getValue(), latch)
每次都new一个新的线程，这样相当于解决了单例方法共享资源的问题，而是多线程每个线程都有新开辟一个新的资源，不存在数据不安全的问题。
AtomicInteger executeNum = new AtomicInteger(0);
Map> map = new HashMap<>();
map.put(1, Lists.newArrayList("a1","b1","c1"));
map.put(2, Lists.newArrayList("a2","b2","c2"));
CountDownLatch latch = new CountDownLatch(map.size());
for (Map.Entry> entry : map.entrySet()) {
ThreadUtil.getDefaultExecutor().execute(new TestThread(entry.getValue(), latch));
}
try {
latch.await();
}
catch (InterruptedException e) {
}
static class TestThread extends Thread {
List testList;
CountDownLatch latch;
public TestThread(List testList, CountDownLatch latch) {
this.testList = testList;
this.latch = latch;
}
@Override
public void run() {
batchInsertMembers(testList);
latch.countDown();
}
}
运行结果如下：
正常
[a1] [a2] i------1 [b2] i------1 i------2 [c2] i------3 [b1] i------2 [c1] i------3三：但是这并不符合我们的诉求，因为我们想要得到执行之后的数量。这里的线程执行完毕之后，并没有进行计数，我们应该怎么做？
定义一个原子类型的计数器，保证线程安全。
AtomicInteger executeNum = new AtomicInteger(0);
这里我们不使用CountDownLatch 来进行开辟线程数，用来等待所有线程执行完毕。而使用Future方法。线程使用Callable而不使用Runnable。
原理和上面ThreadUtil.getDefaultExecutor().execute(new TestThread(entry.getValue(), latch));一样，不过此方法可以用来统计线程返回值。
AtomicInteger executeNum = new AtomicInteger(0);
Map> map = new HashMap<>();
map.put(1, Lists.newArrayList("a1","b1","c1"));
map.put(2, Lists.newArrayList("a2","b2","c2"));
List futureList = new ArrayList<>();
for (Map.Entry> entry : map.entrySet()) {
futureList.add(ThreadUtil.getDefaultExecutor()
.submit(new TestCallable(entry.getValue())));
}
// 汇总所有线程的查询数据
for (Future future : futureList) {
try {
executeNum.addAndGet((Integer) future.get());
}
catch (Exception e) {
System.out.println("error");
}
}
System.out.println(executeNum.get());
static class TestCallable implements Callable {
List testList;
public TestCallable(List testList) {
this.testList = testList;
}
@Override
public Integer call() throws Exception {
return batchInsertMembers(testList);
}
}
