25 | CompletionService：如何批量执行异步任务？

我觉得问题出在return m这里需要等待三个线程执行完成，但是并没有。 ... AtomicReference<Integer> m = new AtomicReference<>(Integer.MAX_VALUE); CountDownLatch latch = new CountDownLatch(3); for(int i=0; i<3; i++) { executor.execute(()->{ Integer r = null; try { r = cs.take().get(); } catch(Exception e) {} save(r); m.set(Integer.min(m.get(), r)); latch.countDown(); }); latch.await(); return m; }

看老师的意图是要等三个比较报假的线程都执行完才能执行主线程的的return m，但是代码无法保证三个线程都执行完，和主线程执行return的顺序，因此，m的值不是准确的，可以加个线程栈栏，线程执行完计数器，来达到这效果

试过返回值是2147483647，也就是int的最大值。没有等待操作完成就猴急的返回了。 m.set(Integer.min(m.get(), r)... 这个操作也不是原子操作。 试着自己弄了一下： public Integer run(){ // 创建线程池 ExecutorService executor = Executors.newFixedThreadPool(3); // 创建 CompletionService CompletionService<Integer> cs = new ExecutorCompletionService<>(executor); AtomicReference<Integer> m = new AtomicReference<>(Integer.MAX_VALUE); // 异步向电商 S1 询价 cs.submit(()->getPriceByS1()); // 异步向电商 S2 询价 cs.submit(()->getPriceByS2()); // 异步向电商 S3 询价 cs.submit(()->getPriceByS3()); // 将询价结果异步保存到数据库 // 并计算最低报价 for (int i=0; i<3; i++) { Integer r = logIfError(()->cs.take().get()); executor.execute(()-> save(r)); m.getAndUpdate(v->Integer.min(v, r)); } return m.get(); } 不知道可不可行

老师，并发工具类，这整个一章，感觉听完似懂非懂的，因为实践中没用过，我要如何弥补这部分，还是说只要听说过，然后用的时候再去查看demo吗

老师stampedLock的获取锁源码，老师能帮忙解惑下么？阻塞的读线程cowait是挂在写节点的下方么？老师能解惑下基于的理论模型 private long acquireWrite(boolean interruptible, long deadline) { WNode node = null, p; for (int spins = -1;;) { // spin while enqueuing long m, s, ns; //如果当前的state是无锁状态即100000000 if ((m = (s = state) & ABITS) == 0L) { //设置成写锁 if (U.compareAndSwapLong(this, STATE, s, ns = s + WBIT)) return ns; } else if (spins < 0) //当前锁状态为写锁状态，并且队列为空，设置自旋值 spins = (m == WBIT && wtail == whead) ? SPINS : 0; else if (spins > 0) { //自旋操作，就是让线程在此自旋 if (LockSupport.nextSecondarySeed() >= 0) --spins; } //如果队列尾元素为空，初始化队列 else if ((p = wtail) == null) { // initialize queue WNode hd = new WNode(WMODE, null); if (U.compareAndSwapObject(this, WHEAD, null, hd)) wtail = hd; } //当前要加入的元素为空，初始化当前元素，前置节点为尾节点 else if (node == null) node = new WNode(WMODE, p); //队列的稳定性判断，当前的前置节点是否改变，重新设置 else if (node.prev != p) node.prev = p; //将当前节点加入尾节点中 else if (U.compareAndSwapObject(this, WTAIL, p, node)) { p.next = node; break; } }

// 获取电商 S1 报价并保存 r=f1.get(); executor.execute(()->save(r)); 如果把r=f1.get（）放进execute里应该是也能保证先执行完的先保存

我实际测试了第一段代码，确实是异步的，f1.get不会阻塞主线程。。。 public static void main(String[] args) { ExecutorService executor = Executors.newFixedThreadPool(3); Future<Integer> f1 = executor.submit(()->getPriceByS1()); Future<Integer> f2 = executor.submit(()->getPriceByS2()); Future<Integer> f3 = executor.submit(()->getPriceByS3()); executor.execute(()-> { try { save(f1.get()); } catch (InterruptedException e) { e.printStackTrace(); } catch (ExecutionException e) { e.printStackTrace(); } }); executor.execute(()-> { try { save(f2.get()); } catch (InterruptedException e) { e.printStackTrace(); } catch (ExecutionException e) { e.printStackTrace(); } }); executor.execute(()-> { try { save(f3.get()); } catch (InterruptedException e) { e.printStackTrace(); } catch (ExecutionException e) { e.printStackTrace(); } }); } private static Integer getPriceByS1() { try { Thread.sleep(10000); } catch (InterruptedException e) { e.printStackTrace(); } return 1; } private static Integer getPriceByS2() { try { Thread.sleep(1000); } catch (InterruptedException e) { e.printStackTrace(); } return 2; } private static Integer getPriceByS3() { try { Thread.sleep(1000); } catch (InterruptedException e) { e.printStackTrace(); } return 3; } private static void save(Integer i) { System.out.println("save " + i); }

1.AtomicReference<Integer>的get方法应该改成使用cas方法 2.最后筛选最小结果的任务是异步执行的，应该在return之前做同步，所以最好使用sumit提交该任务便于判断任务的完成 最后请教老师一下，第一个例子中为什么主线程会阻塞在f1.get()方法呢？

算低价的时候是用三个不同的线程去计算，是异步的，因此可能算出来并不是预期的结果 老师，这样理解对吗？

老师讲的挺不错的，看了这个例子，有几点疑问，还希望老师说明下： // 这个是老师例子： // 创建线程池 ExecutorService executor = Executors.newFixedThreadPool(3); // 创建CompletionService CompletionService<Integer> cs = new ExecutorCompletionService<>(executor); // 异步向电商S1询价 cs.submit(()->getPriceByS1()); // 异步向电商S2询价 cs.submit(()->getPriceByS2()); // 异步向电商S3询价 cs.submit(()->getPriceByS3()); // 将询价结果异步保存到数据库 for (int i=0; i<3; i++) { Integer r = cs.take().get(); executor.execute(()->save(r)); } 首先，CompletionService应该是要绑定泛型，代表异步任务的返回结果，实际应用中，几乎不太可能所有的异步任务的返回类型是一样的，除非设置成Object这种通用型，那又会导致拿到结果后需要强转，代码看起来更难受； 其次，对于返回的结果的处理方式，实际应用中几乎也是不同的，那就要针对每一个take出来的结果做判断，这实际上也是会导致代码很难维护； 综上，CompletionService 看来能够做批量处理异步任务的事情，实际应用中，我感觉不太实用！ 以上两点是个人见解，有不对之处请老师指教！