Java 并发编程实战-创建和执行任务的最佳实践

Posted 2022-04-25 公众号JavaEdge

若无法通过并行流实现并发，则必须创建并运行自己的任务。运行任务的理想Java 8方法就是CompletableFuture。
Java并发的历史始于非常原始和有问题的机制，并且充满各种尝试的优化。本文将展示一个规范形式，表示创建和运行任务的最简单，最好的方法。
Java初期通过直接创建自己的Thread对象来使用线程，甚至子类化来创建特定“任务线程”对象。手动调用构造函数并自己启动线程。创建所有这些线程的开销变得非常重要，现在不鼓励。Java 5中，添加了类来为你处理线程池。可以将任务创建为单独的类型，然后将其交给ExecutorService运行，而不是为每种不同类型的任务创建新的Thread子类型。ExecutorService为你管理线程，并在运行任务后重新循环线程而不是丢弃线程。

创建任务

这只是个包含run()方法的Runnable类。它没有包含实际运行任务的机制。使用Nap类中的“sleep”：

第二个构造函数在超时的时候，会显示一条消息。TimeUnit.MILLISECONDS.sleep()：获取“当前线程”并在参数中将其置于休眠状态，这意味着该线程被挂起。这并不意味着底层处理器停止。os将其切换到其他任务，例如在你的计算机上运行另一个窗口。OS任务管理器定期检查sleep()是否超时。当它执行时，线程被“唤醒”并给予更多处理时间。

sleep()抛已检查的InterruptedException：通过突然中断它们来终止任务。由于它往往会产生不稳定状态，所以不鼓励用来终止。但我们必须在需要或仍发生终止的情况下捕获该异常。

执行任务

结果：

All tasks submitted
main awaiting termination
main awaiting termination
NapTask[0] pool-1-thread-1
main awaiting termination
NapTask[1] pool-1-thread-1
main awaiting termination
NapTask[2] pool-1-thread-1
main awaiting termination
NapTask[3] pool-1-thread-1
main awaiting termination
NapTask[4] pool-1-thread-1
main awaiting termination
NapTask[5] pool-1-thread-1
main awaiting termination
NapTask[6] pool-1-thread-1
main awaiting termination
NapTask[7] pool-1-thread-1
main awaiting termination
NapTask[8] pool-1-thread-1
main awaiting termination
NapTask[9] pool-1-thread-1

创建十个NapTasks并将它们提交给ExecutorService，它们开始自己运行。然而，期间main()继续运行。当运行至​​exec.shutdown();​​时，main告诉ExecutorService完成已提交的任务，但不再接受新任务。此时，这些任务仍在运行，必须等到它们在退出main()之前完成。这是通过检查​​exec.isTerminated()​​来实现：在所有任务完成后为true。

main()中线程的名称是main，且只有一个其他线程pool-1-thread-1。此外，交错输出显示两个线程确实在同时运行。
若仅调用​​exec.shutdown()​​，程序将完成所有任务，若尝试提交新任务将抛RejectedExecutionException。

exec.shutdown()的替代方法exec.shutdownNow()：除了不接受新任务，还会尝试通过中断任务来停止任何当前正在运行的任务。同样，中断是错误的，容易出错，不鼓励！

使用更多线程

使用线程的重点几乎总是更快地完成任务，那为何要限制自己使用SingleThreadExecutor？Executors还给了我们更多选项，如CachedThreadPool：

运行该程序时，你会发现它完成得更快。这是有道理的，而不是使用相同线程来顺序运行每个任务，每个任务都有自己的线程，所以它们并行运行。似乎没有缺点，很难看出为什么有人会使用SingleThreadExecutor。
要理解这个问题，需要一个更复杂任务：

用CachedThreadPool试一下：

输出结果：

0 pool-1-thread-1 195
3 pool-1-thread-4 400
2 pool-1-thread-3 300
1 pool-1-thread-2 200
5 pool-1-thread-6 600
6 pool-1-thread-7 700
4 pool-1-thread-5 500
7 pool-1-thread-3 800
8 pool-1-thread-5 900
9 pool-1-thread-7 1000

输出不是期望的，并且从一次运行到下一次运行会有所不同。问题是所有的任务都试图写入val的单个实例，并且他们正在踩着彼此的脚趾。这样的类就不是线程安全的。
看SingleThreadExecutor表现怎样：

输出结果：

0 pool-1-thread-1 100
1 pool-1-thread-1 200
2 pool-1-thread-1 300
3 pool-1-thread-1 400
4 pool-1-thread-1 500
5 pool-1-thread-1 600
6 pool-1-thread-1 700
7 pool-1-thread-1 800
8 pool-1-thread-1 900
9 pool-1-thread-1 1000

每次都得到一致结果，虽然InterferingTask缺乏线程安全性。这是SingleThreadExecutor的主要好处 - 因为它一次运行一个任务，这些任务不会相互干扰，等于强加了线程安全性。这种现象称为线程限制，因为在单线程上运行任务限制了它们的影响。【线程限制】限制了加速，但能节省很多困难的调试和重写。

产生结果

因为InterferingTask是Runnable，无返回值，因此只能使用副作用产生结果 - 操纵缓冲值而不是返回结果。副作用是并发编程中的主要问题之一，因为我们看到了CachedThreadPool2.java。InterferingTask中的val被称为可变共享状态，这就是问题：多个任务同时修改同一个变量会产生竞争。结果取决于首先在终点线上执行哪个任务，并修改变量（以及其他可能性的各种变化）。

避免竞争条件的最好方法是避免可变的共享状态，可称为自私的孩子原则：什么都不分享。
使用InterferingTask，最好删除副作用并返回任务结果。为此，我们创建Callable而非Runnable：

call()完全独立于所有其他CountingTasks生成其结果，这意味着没有可变的共享状态。

输出结果：

0 pool-1-thread-1 100
2 pool-1-thread-3 100
1 pool-1-thread-2 100
3 pool-1-thread-4 100
4 pool-1-thread-5 100
5 pool-1-thread-6 100
6 pool-1-thread-7 100
7 pool-1-thread-5 100
8 pool-1-thread-7 100
9 pool-1-thread-6 100
sum = 1000

所有任务完成后，invokeAll()才会返回一个Future列表，每个任务一个Future。Future是Java 5中引入的机制，允许提交任务而无需等待它完成。

结果：

99 pool-1-thread-1 100
100

但这意味着，在CachedThreadPool3.java中，Future似乎是多余的，因为invokeAll()在所有任务完成前都不会返回。但此处的Future并非用于延迟结果，而是捕获任何可能的异常。

在CachedThreadPool3.java.get()抛异常，因此extractResult()在Stream中执行此提取。因为调用get()时，Future会阻塞，所以它只能解决【等待任务完成】的问题。最终，Futures被认为是一种无效解决方案，现在不鼓励，支持Java 8的CompletableFuture，将在后面探讨。当然，你仍会在遗留库中遇到Futures。

可使用并行Stream，更简单优雅解决该问题：

输出结果：

4 ForkJoinPool.commonPool-worker-15 100
1 ForkJoinPool.commonPool-worker-11 100
5 ForkJoinPool.commonPool-worker-1 100
2 ForkJoinPool.commonPool-worker-9 100
0 ForkJoinPool.commonPool-worker-6 100
3 ForkJoinPool.commonPool-worker-8 100
9 ForkJoinPool.commonPool-worker-13 100
6 main 100
8 ForkJoinPool.commonPool-worker-2 100
7 ForkJoinPool.commonPool-worker-4 100
1000

这更容易理解，需要做的就是将parallel()插入到其他顺序操作中，然后一切都在同时运行。

Lambda和方法引用作为任务

使用lambdas和方法引用，你不仅限于使用Runnables和Callables。因为Java 8通过匹配签名来支持lambda和方法引用（即支持结构一致性），所以我们可以将不是Runnables或Callables的参数传递给ExecutorService：

输出结果：

Lambda1
NotRunnable
Lambda2
NotCallable

这里，前两个submit()调用可以改为调用execute()。所有submit()调用都返回Futures，你可以在后两次调用的情况下提取结果。