文章目录
- 1.线程的创建方式
- 1.1继承Thread类，重写run方法
- 1.2实现Runnable接口，重写run方法。
- 1.3实现Callable接口，重新call方法
- 1.4以上三种总结
- 1.5使用线程池创建线程
- 1.5.1线程池创建线程的方式
- 1.5.2线程池的七大参数含义
- 1.5.3线程池的工作流程
- 1.5.4一个线程池core：7，max：20，queue：50。100个并发进来，怎么分配。
- 2.CompletableFuture异步编排
- 2.1创建异步对象方式
- 2.2计算完成时回调方法
- 2.1.1方法完成时的感知（方法一）
- 2.1.2方法完成时的处理（方法二）
- 2.3线程的串行化的方法
- 2.3.1不能接收值且没有返回值
- 2.3.2可以接收值但是没有返回值
- 2.3.3可以接收值也可以返回值
- 2.4两任务组合-一个完成即可
- 2.5两任务组合-两个都要完成
- 2.6多任务组合
1.线程的创建方式
1.1继承Thread类，重写run方法
package com.atguigu.gmall.product.thread;
import java.math.BigDecimal;
public class ThreadTest {
public static void main(String[] args) {
/**
* 线程的创建方式
* 1.继承Thread类
*/
//开启线程
System.out.println("主线程开始");
Thread thread = new Thread01();
thread.start();
System.out.println("主线程完毕");
}
public static class Thread01 extends Thread{
//创建线程方法一
//通过继承Thread类重写run()方法,在run()方法中编写业务类
@Override
public void run() {
System.out.println("通过继承Thread类，重写run()方法，创建线程"+Thread.currentThread().getId());
BigDecimal bigDecimal = new BigDecimal(10);
BigDecimal bigDecimal1 = new BigDecimal(3);
BigDecimal divide = bigDecimal1.divide(bigDecimal);
System.out.println("divide = " + divide);
}
}
}
结果
1.2实现Runnable接口，重写run方法。
package com.atguigu.gmall.product.thread;
import java.math.BigDecimal;
public class RunableTest {
public static void main(String[] args) {
/**
* 创建线程的方法二：
* 通过实现Runable接口，重新run方法，创建线程。
*/
//开启线程
System.out.println("主线程开始");
Runable01 runable01 = new Runable01();
Thread thread = new Thread(runable01);
thread.start();
System.out.println("主线程完毕");
}
public static class Runable01 implements Runnable{
@Override
public void run() {
System.out.println("通过实现Runnable接口，重写run()方法，创建线程"+Thread.currentThread().getId());
BigDecimal bigDecimal = new BigDecimal(10);
BigDecimal bigDecimal1 = new BigDecimal(3);
BigDecimal divide = bigDecimal1.divide(bigDecimal);
System.out.println("divide = " + divide);
}
}
}
1.3实现Callable接口，重新call方法
package com.atguigu.gmall.product.thread;
import java.math.BigDecimal;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;
import java.util.concurrent.FutureTask;
public class CallableTest {
public static void main(String[] args) throws ExecutionException, InterruptedException {
/**
* 创建线程的方法三
* 通过实现Callable<>接口，重写call方法，创建线程。可以获取到线程的返回值
*/
System.out.println("主线程开始");
FutureTask<String> futureTask = new FutureTask<String>(new Callable01());
//开启线程
new Thread(futureTask).start();
//获取线程的返回值，会阻塞主线程
System.out.println("主线程阻塞。。。。。。");
String s = futureTask.get();
System.out.println("线程的返回值s = " + s);
System.out.println("主线程结束");
}
public static class Callable01 implements Callable<String>{
@Override
public String call() throws Exception {
System.out.println("通过实现Callable<>接口，重写call方法，创建线程。可以获取到线程的返回值"+Thread.currentThread().getId());
BigDecimal bigDecimal = new BigDecimal(10);
BigDecimal bigDecimal1 = new BigDecimal(3);
BigDecimal divide = bigDecimal1.divide(bigDecimal);
System.out.println("divide = " + divide);
return divide.toString();
}
}
}
1.4以上三种总结
1.开启线程的方式，Thread对象调用start方法。
2.以上三种只有第三种可以接收线程的返回值。
1.5使用线程池创建线程
1.5.1线程池创建线程的方式
/**
* 使用线程池创建线程
*/
ThreadPoolExecutor executor = new ThreadPoolExecutor(
10,
20,
10,
TimeUnit.SECONDS,
new ArrayBlockingQueue<>(1000),
Executors.defaultThreadFactory(),
new ThreadPoolExecutor.AbortPolicy()
);
1.5.2线程池的七大参数含义
/**
* Creates a new {@code ThreadPoolExecutor} with the given initial
* parameters.
*
* @param corePoolSize the number of threads to keep in the pool, even
* if they are idle, unless {@code allowCoreThreadTimeOut} is set
* @param maximumPoolSize the maximum number of threads to allow in the
* pool
* @param keepAliveTime when the number of threads is greater than
* the core, this is the maximum time that excess idle threads
* will wait for new tasks before terminating.
* @param unit the time unit for the {@code keepAliveTime} argument
* @param workQueue the queue to use for holding tasks before they are
* executed. This queue will hold only the {@code Runnable}
* tasks submitted by the {@code execute} method.
* @param threadFactory the factory to use when the executor
* creates a new thread
* @param handler the handler to use when execution is blocked
* because the thread bounds and queue capacities are reached
*/
public ThreadPoolExecutor(int corePoolSize,
int maximumPoolSize,
long keepAliveTime,
TimeUnit unit,
BlockingQueue<Runnable> workQueue,
ThreadFactory threadFactory,
RejectedExecutionHandler handler)
- corePoolSize:核心的线程池数。也就是线程池一创建就有的。
- maximumPoolSize:最大的线程池数。这个线程池可以创建的最大的线程池数。
- keepAliveTime:当线程池中的线程数大于核心的线程池时，这些线程池执行完任务保持存活的时间。
- unit:时间单位
- workQueue:阻塞队列，当任务大于核心线程数时，任务就会放在阻塞队列中。
- threadFactory：创建工厂。指定线程名。
- handler:拒绝策略。当线程池中所有的线程都在执行任务，而且阻塞队列已经满了。那么来了任务就需要执行拒绝策略了。
1.5.3线程池的工作流程
1、创建线程池，会创建core线程。
2、当任务来了，core线程进行处理，若core不够，那么就会将任务放在workQueue中，当核心线程空闲下来，去workQueue阻塞队列中去任务。
3、若阻塞队列满了，线程池就去开启新的线程，直至线程池中的线程数达到maximumPoolSize最大线程池数。若新的线程空闲下来，过了过期时间，就会自动销毁。
4、若线程池中的线程池数达到了最大线程池数，而且还来了任务，那么就会使用拒绝策略进行处理。
5、所有的线程都是由指定的factory工厂创建的。
1.5.4一个线程池core：7，max：20，queue：50。100个并发进来，怎么分配。
首先：7个线程直接进行处理。
然后：进入队列50个。
再次：开启13个线程进行处理。
最后：70个被安排，30个交给阻塞队列。
2.CompletableFuture异步编排
2.1创建异步对象方式
//方法一：
public static CompletableFuture<Void> runAsync(Runnable runnable) {
return asyncRunStage(asyncPool, runnable);
}
//方法二
public static CompletableFuture<Void> runAsync(Runnable runnable,Executor executor) {
return asyncRunStage(screenExecutor(executor), runnable);
}
//方法三
public static <U> CompletableFuture<U> supplyAsync(Supplier<U>supplier) {
return asyncSupplyStage(asyncPool, supplier);
}
//方法四
public static <U> CompletableFuture<U> supplyAsync(Supplier<U> supplier,Executor executor) {
return asyncSupplyStage(screenExecutor(executor), supplier);
}
1.runXxx方法没有返回值，supplyXxx方法有返回值。
2.可以传入自定义的线程池，否则默认的线程池。
3.都不会接收返回值。
代码
package com.atguigu.gmall.product.completableFuture;
import rx.Completable;
import java.math.BigDecimal;
import java.util.concurrent.*;
public class Test {
public static ExecutorService executors = Executors.newFixedThreadPool(10);
public static void main(String[] args) throws ExecutionException, InterruptedException {
/**
* 1.创建异步对象
*/
//CompletableFuture类中的静态方法
long startMain = System.currentTimeMillis();
System.out.println("主线程--开始");
CompletableFuture<Void> future01 = CompletableFuture.runAsync(new Runnable01());
CompletableFuture<Void> future02 = CompletableFuture.runAsync(() -> {
long start02 = System.currentTimeMillis();
System.out.println("id============================");
long id = Thread.currentThread().getId();
System.out.println("当前线程的id = " + id);
BigDecimal a = new BigDecimal(10);
BigDecimal b = new BigDecimal(2);
BigDecimal divide = a.divide(b);
System.out.println("运行结果divide = " + divide+"02-"+(System.currentTimeMillis() - start02));
}, executors);
CompletableFuture<String> future03 = CompletableFuture.supplyAsync(() -> {
long start03 = System.currentTimeMillis();
long id = Thread.currentThread().getId();
System.out.println("id============================");
System.out.println("当前线程的id = " + id);
BigDecimal a = new BigDecimal(10);
BigDecimal b = new BigDecimal(2);
BigDecimal divide = a.divide(b);
System.out.println("运行结果divide = " + divide+"03-"+(System.currentTimeMillis() - start03));
return divide.toString();
});
System.out.println("获取返回结果future03.get() = " + future03.get());
CompletableFuture<String> future04 = CompletableFuture.supplyAsync(() -> {
long start04 = System.currentTimeMillis();
long id = Thread.currentThread().getId();
System.out.println("id============================");
System.out.println("当前线程的id = " + id);
BigDecimal a = new BigDecimal(10);
BigDecimal b = new BigDecimal(2);
BigDecimal divide = a.divide(b);
System.out.println("运行结果divide = " + divide+"04-"+(System.currentTimeMillis() - start04));
return divide.toString();
},executors);
System.out.println("获取返回结果future04 = " + future04.get());
System.out.println("主线程--结束"+"Main用时"+(System.currentTimeMillis() - startMain));
}
public static class Runnable01 implements Runnable{
@Override
public void run() {
long start01 = System.currentTimeMillis();
System.out.println("id============================");
long id = Thread.currentThread().getId();
System.out.println("当前线程的id = " + id);
BigDecimal a = new BigDecimal(10);
BigDecimal b = new BigDecimal(2);
BigDecimal divide = a.divide(b);
System.out.println("运行结果divide = " + divide+"01-"+(System.currentTimeMillis() - start01));
}
}
public static class Callable01 implements Callable<String> {
@Override
public String call() {
System.out.println("id============================");
long id = Thread.currentThread().getId();
System.out.println("当前线程的id = " + id);
BigDecimal a = new BigDecimal(10);
BigDecimal b = new BigDecimal(2);
BigDecimal divide = a.divide(b);
System.out.println("运行结果divide = " + divide);
return divide.toString();
}
}
}
2.2计算完成时回调方法
2.1.1方法完成时的感知（方法一）
public CompletableFuture<T> whenComplete(
BiConsumer<? super T, ? super Throwable> action) {
return uniWhenCompleteStage(null, action);
}
public CompletableFuture<T> whenCompleteAsync(
BiConsumer<? super T, ? super Throwable> action) {
return uniWhenCompleteStage(asyncPool, action);
}
public CompletableFuture<T> whenCompleteAsync(
BiConsumer<? super T, ? super Throwable> action, Executor executor) {
return uniWhenCompleteStage(screenExecutor(executor), action);
}
public CompletableFuture<T> exceptionally(
Function<Throwable, ? extends T> fn) {
return uniExceptionallyStage(fn);
}
whenComplete 可以处理正常结果但是不能返回结果、感知异常但是不能处理异常。这个方法不可以进行返回值
exceptionally可以感知异常并且修改返回值进行返回。
whenComplete 和 whenCompleteAsync 的区别：
whenComplete：是执行当前任务的线程执行继续执行 whenComplete 的任务。
whenCompleteAsync：是执行把 whenCompleteAsync 这个任务继续提交给线程池来进行执行。
方法不以 Async 结尾，意味着 Action 使用相同的线程执行，而 Async 可能会使用其他线程执行（如果是使用相同的线程池，也可能会被同一个线程选中执行）
代码示例
package com.atguigu.gmall.product.completableFuture;
import java.math.BigDecimal;
import java.util.concurrent.*;
public class Test02 {
public static ExecutorService executors = Executors.newFixedThreadPool(10);
public static void main(String[] args) throws ExecutionException, InterruptedException {
CompletableFuture<String> exceptionally = CompletableFuture.supplyAsync(() -> {
int i = 10/0;
return "a";
}).whenCompleteAsync((res, exception) -> {
//尽可以感到异常，不可以修改返回结果
System.out.println("输出返回结果" + res);
}, executors).exceptionally((exception -> {
//可以感到异常，并且修改返回结果
return "b";
}));
System.out.println("获取返回结果：" + exceptionally.get());
}
public static class Runnable01 implements Runnable{
@Override
public void run() {
long start01 = System.currentTimeMillis();
System.out.println("id============================");
long id = Thread.currentThread().getId();
System.out.println("当前线程的id = " + id);
BigDecimal a = new BigDecimal(10);
BigDecimal b = new BigDecimal(2);
BigDecimal divide = a.divide(b);
System.out.println("运行结果divide = " + divide+"01-"+(System.currentTimeMillis() - start01));
}
}
public static class Callable01 implements Callable<String> {
@Override
public String call() {
System.out.println("id============================");
long id = Thread.currentThread().getId();
System.out.println("当前线程的id = " + id);
BigDecimal a = new BigDecimal(10);
BigDecimal b = new BigDecimal(2);
BigDecimal divide = a.divide(b);
System.out.println("运行结果divide = " + divide);
return divide.toString();
}
}
}
2.1.2方法完成时的处理（方法二）
public <U> CompletableFuture<U> handle(
BiFunction<? super T, Throwable, ? extends U> fn) {
return uniHandleStage(null, fn);
}
public <U> CompletableFuture<U> handleAsync(
BiFunction<? super T, Throwable, ? extends U> fn) {
return uniHandleStage(asyncPool, fn);
}
public <U> CompletableFuture<U> handleAsync(
BiFunction<? super T, Throwable, ? extends U> fn, Executor executor) {
return uniHandleStage(screenExecutor(executor), fn);
}
不仅可以处理正常结果而且可以处理异常
不仅可以接收值，而且可以返回处理结果
代码实例
package com.atguigu.gmall.product.completableFuture;
import java.math.BigDecimal;
import java.util.concurrent.*;
public class Test02 {
public static ExecutorService executors = Executors.newFixedThreadPool(10);
public static void main(String[] args) throws ExecutionException, InterruptedException {
CompletableFuture<String> exceptionally = CompletableFuture.supplyAsync(() -> {
int i = 10/0;
return "a";
}).handleAsync((res,exception) -> {
//不仅可以接收参数，而且可以返回结果
if (res != null){
return "值"+res;
}
if (exception != null){
return "异常"+exception.getMessage();
}
return "0";
},executors);
System.out.println("获取返回结果：" + exceptionally.get());
}
public static class Runnable01 implements Runnable{
@Override
public void run() {
long start01 = System.currentTimeMillis();
System.out.println("id============================");
long id = Thread.currentThread().getId();
System.out.println("当前线程的id = " + id);
BigDecimal a = new BigDecimal(10);
BigDecimal b = new BigDecimal(2);
BigDecimal divide = a.divide(b);
System.out.println("运行结果divide = " + divide+"01-"+(System.currentTimeMillis() - start01));
}
}
public static class Callable01 implements Callable<String> {
@Override
public String call() {
System.out.println("id============================");
long id = Thread.currentThread().getId();
System.out.println("当前线程的id = " + id);
BigDecimal a = new BigDecimal(10);
BigDecimal b = new BigDecimal(2);
BigDecimal divide = a.divide(b);
System.out.println("运行结果divide = " + divide);
return divide.toString();
}
}
}
2.3线程的串行化的方法
2.3.1不能接收值且没有返回值
thenRun方法：只要上面的任务执行完成，就开始执行thenRun，只是处理完任务后，执行 thenRun的后续操作
public CompletableFuture<Void> thenRun(Runnable action) {
return uniRunStage(null, action);
}
public CompletableFuture<Void> thenRunAsync(Runnable action) {
return uniRunStage(asyncPool, action);
}
public CompletableFuture<Void> thenRunAsync(Runnable action,
Executor executor) {
return uniRunStage(screenExecutor(executor), action);
}
代码示例
package com.atguigu.gmall.product.completableFuture;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
public class Test03 {
public static ExecutorService excutor =Executors.newFixedThreadPool(10);
public static void main(String[] args) throws ExecutionException, InterruptedException {
CompletableFuture<Void> future01 = CompletableFuture.supplyAsync(() -> {
int i = 0;
System.out.println("i = " + i);
return i;
}).thenRunAsync(() -> {
int j = 0;
System.out.println("j = " + j);
});
Void unused = future01.get();
System.out.println("unused = " + unused);
}
}
2.3.2可以接收值但是没有返回值
thenAccept方法：消费处理结果。接收任务的处理结果，并消费处理，无返回结果。
public CompletableFuture<Void> thenAccept(Consumer<? super T> action) {
return uniAcceptStage(null, action);
}
public CompletableFuture<Void> thenAcceptAsync(Consumer<? super T> action) {
return uniAcceptStage(asyncPool, action);
}
public CompletableFuture<Void> thenAcceptAsync(Consumer<? super T> action,
Executor executor) {
return uniAcceptStage(screenExecutor(executor), action);
}
2.3.3可以接收值也可以返回值
thenApply 方法：当一个线程依赖另一个线程时，获取上一个任务返回的结果，并返回当前任务的返回值。
public <U> CompletableFuture<U> thenApply(
Function<? super T,? extends U> fn) {
return uniApplyStage(null, fn);
}
public <U> CompletableFuture<U> thenApplyAsync(
Function<? super T,? extends U> fn) {
return uniApplyStage(asyncPool, fn);
}
public <U> CompletableFuture<U> thenApplyAsync(
Function<? super T,? extends U> fn, Executor executor) {
return uniApplyStage(screenExecutor(executor), fn);
}
代码示例
package com.atguigu.gmall.product.completableFuture;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
public class Test03 {
public static ExecutorService excutor =Executors.newFixedThreadPool(10);
public static void main(String[] args) throws ExecutionException, InterruptedException {
CompletableFuture<Integer> future03 = CompletableFuture.supplyAsync(() -> {
int i = 0;
System.out.println("i = " + i);
return i;
}).thenApplyAsync((res) -> {
res++;
return res;
});
Integer integer = future03.get();
System.out.println("integer = " + integer);
}
}
带有Async默认是异步执行的。这里所谓的异步指的是不在当前线程内执行。
Function<? super T,? extends U>
T：上一个任务返回结果的类型
U：当前任务的返回值类型
![[hddm]python模块hddm安装后测试代码](https://img-blog.csdnimg.cn/direct/94855141845b40529578290714fbeeb4.png)
