MRCODE-BOOK

# 本地调用异步化

# 同步 VS 异步

  • 同步：程序按照定义的顺序执行，每一行都必须等待上一行完成后才能执行
  • 异步：程序执行时，无需等待异步调用的语句返回，即可执行后面的操作

# 异步适用场景

耗时操作中，将不影响主流程的操作异步执行，降低响应时间。 下面是一个例子：该例子模拟了一个非常耗时的操作

package cn.mrcode.study;

import java.util.Date;
import java.util.concurrent.TimeUnit;

/**
 * @author mrcode
 * @date 2023/2/21 21:25
 */
public class SyncDemo {
    private void subBiz1() throws InterruptedException {
        Thread.sleep(1000L);
        System.out.println(new Date() + " subBiz1");
    }

    private void subBiz2() throws InterruptedException {
        Thread.sleep(1000L);
        System.out.println(new Date() + " subBiz2");
    }

    private void saveOpLog() throws InterruptedException {
        Thread.sleep(1000L);
        System.out.println(new Date() + " 插入操作");
    }

    private void biz() throws InterruptedException {
        this.subBiz1();
        this.saveOpLog();
        this.subBiz2();
        System.out.println(new Date() + " 执行结束");
    }

    public static void main(String[] args) throws InterruptedException {
        new SyncDemo().biz();
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37

运行输出结果如下

Tue Feb 21 21:28:17 CST 2023 subBiz1
Tue Feb 21 21:28:18 CST 2023 插入操作
Tue Feb 21 21:28:19 CST 2023 subBiz2
Tue Feb 21 21:28:19 CST 2023 执行结束
1
2
3
4

可以看到，需要耗费 3 秒才能返回。 假设 biz 中的三个方法，其中一个方法的返回结果不会影响整个流程的花，就可以把它改造成异步的方式

# 异步方式

# 线程执行

比如上面的 saveOpLog()方法是保存操作日志，这里的结果不会影响主流程（日志是否保存成功无关紧要） 下面是改造后的例子

package cn.mrcode.study;

import java.util.Date;
import java.util.concurrent.TimeUnit;

/**
 * @author mrcode
 * @date 2023/2/21 21:25
 */
public class SyncDemo {
    private void subBiz1() throws InterruptedException {
        Thread.sleep(1000L);
        System.out.println(new Date() + " subBiz1");
    }

    private void subBiz2() throws InterruptedException {
        Thread.sleep(1000L);
        System.out.println(new Date() + " subBiz2");
    }

    private void saveOpLog() throws InterruptedException {
        Thread.sleep(1000L);
        System.out.println(new Date() + " 插入操作");
    }

    private void biz() throws InterruptedException {
        this.subBiz1();
        new Thread(()-> {
            try {
                this.saveOpLog();
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
        }).start();
//        this.saveOpLog();
        this.subBiz2();
        System.out.println(new Date() + " 执行结束");
    }

    public static void main(String[] args) throws InterruptedException {
        new SyncDemo().biz();
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44

运行输出结果

Tue Feb 21 21:33:16 CST 2023 subBiz1
Tue Feb 21 21:33:17 CST 2023 插入操作
Tue Feb 21 21:33:17 CST 2023 subBiz2
Tue Feb 21 21:33:17 CST 2023 执行结束
1
2
3
4

可以看到，现在只耗费了 2 秒

WARNING

但是这样的手动创建线程的方式，其实不是很优雅。 因为它没有线程重用、控制并发等能力

# 线程池

package cn.mrcode.study;

import java.util.Date;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingDeque;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * @author mrcode
 * @date 2023/2/21 21:25
 */
public class SyncDemoThreadPool {
    ThreadPoolExecutor executor = new ThreadPoolExecutor(5, 10,
            10L, TimeUnit.SECONDS,
            new LinkedBlockingDeque<>(100),
            Executors.defaultThreadFactory(),
            new ThreadPoolExecutor.AbortPolicy());

    private void subBiz1() throws InterruptedException {
        Thread.sleep(1000L);
        System.out.println(new Date() + " subBiz1");
    }

    private void subBiz2() throws InterruptedException {
        Thread.sleep(1000L);
        System.out.println(new Date() + " subBiz2");
    }

    private void saveOpLog() throws InterruptedException {
        Thread.sleep(1000L);
        System.out.println(new Date() + " 插入操作");
    }

    private void biz() throws InterruptedException {
        this.subBiz1();
        executor.submit(()->{
            try {
                this.saveOpLog();
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
        });
        this.subBiz2();
        System.out.println(new Date() + " 执行结束");
    }

    public static void main(String[] args) throws InterruptedException {
        new SyncDemoThreadPool().biz();
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52

添加一个线程池，然后将操作放到线程池里面去，运行结果如下

Tue Feb 21 21:40:24 CST 2023 subBiz1
Tue Feb 21 21:40:26 CST 2023 subBiz2
Tue Feb 21 21:40:26 CST 2023 插入操作
Tue Feb 21 21:40:26 CST 2023 执行结束
1
2
3
4

# @Async 注解

此方式是 SpringBoot 框架下的注解功能，本质上是对线程池方式的一种简化。

package cn.mrcode.demo.boodadmin.jvm;

import org.springframework.context.annotation.Configuration;
import org.springframework.scheduling.annotation.EnableAsync;

/**
 * @author mrcode
 * @date 2023/2/21 21:43
 */
@Configuration
@EnableAsync // 开启异步
public class AsyncConfig {
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14

然后将需要异步的方法提取到另外一个类中（同类中其实也可以，就是调用方式需要改变，否则 AOP 不会生效）

package cn.mrcode.demo.boodadmin.jvm;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.stereotype.Component;

import java.util.Date;

/**
 * @author mrcode
 * @date 2023/2/21 21:25
 */
@Component
public class SyncDemoAsyncAnnotation {
    @Autowired
    private AsyncJob asyncJob;

    private void subBiz1() throws InterruptedException {
        Thread.sleep(1000L);
        System.out.println(new Date() + " subBiz1");
    }

    private void subBiz2() throws InterruptedException {
        Thread.sleep(1000L);
        System.out.println(new Date() + " subBiz2");
    }


    public void biz() throws InterruptedException {
        this.subBiz1();
        asyncJob.saveOpLog();
        this.subBiz2();
        System.out.println(new Date() + " 执行结束");
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36

这是提取出来的方法

package cn.mrcode.demo.boodadmin.jvm;

import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Component;

import java.util.Date;

/**
 * @author mrcode
 * @date 2023/2/21 21:46
 */
@Component
public class AsyncJob {
    @Async
    public void saveOpLog() throws InterruptedException {
        Thread.sleep(1000L);
        System.out.println(new Date() + " 插入操作");
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20

然后再写一个测试入口来触发

package cn.mrcode.demo.boodadmin.jvm;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;

/**
 * @author mrcode
 * @date 2023/2/21 21:47
 */
@RestController
public class AsyncTestController {
    @Autowired
    private SyncDemoAsyncAnnotation syncDemoAsyncAnnotation;

    @GetMapping("/async-test")
    public String asyncTest() throws InterruptedException {
        syncDemoAsyncAnnotation.biz();
        return "success";
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22

程序启动后访问：http://127.0.0.1:8090/async-test (opens new window) 控制太输出如下

Tue Feb 21 22:07:53 CST 2023 subBiz1
Tue Feb 21 22:07:54 CST 2023 subBiz2
Tue Feb 21 22:07:54 CST 2023 执行结束
Tue Feb 21 22:07:54 CST 2023 插入操作
1
2
3
4

可以看到也是 2 秒就完成了

# 如果你想要执行结果

package cn.mrcode.demo.boodadmin.jvm;

import org.springframework.scheduling.annotation.Async;
import org.springframework.scheduling.annotation.AsyncResult;
import org.springframework.stereotype.Component;

import java.util.Date;
import java.util.concurrent.Future;

/**
* @author mrcode
* @date 2023/2/21 21:46
*/
@Component
    public class AsyncJob {
        @Async
        public Future<String> saveOpLog() throws InterruptedException {
            Thread.sleep(1000L);
            String result = new Date() + " 插入操作";
            // System.out.println(result);
            return new AsyncResult<>(result);
        }
    }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24

调用处做修改

package cn.mrcode.demo.boodadmin.jvm;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.stereotype.Component;

import java.util.Date;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;

/**
 * @author mrcode
 * @date 2023/2/21 21:25
 */
@Component
public class SyncDemoAsyncAnnotation {
    @Autowired
    private AsyncJob asyncJob;

    private void subBiz1() throws InterruptedException {
        Thread.sleep(1000L);
        System.out.println(new Date() + " subBiz1");
    }

    private void subBiz2() throws InterruptedException {
        Thread.sleep(1000L);
        System.out.println(new Date() + " subBiz2");
    }


    public void biz() throws InterruptedException, ExecutionException {
        this.subBiz1();
        Future<String> f = asyncJob.saveOpLog();
        this.subBiz2();
        // 在这里获取结果
        // 因为 subBiz2 会休眠 1 秒，saveOpLog 调用后，1 秒后才会返回结果
        // 所以在这里调用的话，刚好能拿到结果，而不阻塞（如果结果没有返回，这里调用 get 方法就会阻塞）
        System.out.println(f.get());
        System.out.println(new Date() + " 执行结束");
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42

测试输出

Tue Feb 21 22:16:22 CST 2023 subBiz1
Tue Feb 21 22:16:23 CST 2023 subBiz2
Tue Feb 21 22:16:23 CST 2023 插入操作
Tue Feb 21 22:16:23 CST 2023 执行结束
1
2
3
4

可以看到结果同样是只耗费 2 秒

# @Async 注意事项

@Async 使用很简单，但是有几个注意事项：

  • @Async注解标注的方法必须返回 void 或 Future
  • 建议将 @Async方法放置到独立的类中，否则出现 this 调用时，异步调用就会失效
  • 建议自定义 BlockingQueue 的大小

该注解的本质是简化线程池的调用，所以底层最终也是放到线程池中运行的，这一点在 @EnableAsync� 注解源码上就有相关的说明

// org.springframework.boot.autoconfigure.task.TaskExecutionAutoConfiguration

@Bean
	@ConditionalOnMissingBean
	public TaskExecutorBuilder taskExecutorBuilder(TaskExecutionProperties properties,
			ObjectProvider<TaskExecutorCustomizer> taskExecutorCustomizers,
			ObjectProvider<TaskDecorator> taskDecorator) {
		TaskExecutionProperties.Pool pool = properties.getPool();
		TaskExecutorBuilder builder = new TaskExecutorBuilder();
        // 这里默认的容量是 Integer.MAX_VALUE;
		builder = builder.queueCapacity(pool.getQueueCapacity());
		builder = builder.corePoolSize(pool.getCoreSize());
		builder = builder.maxPoolSize(pool.getMaxSize());
		builder = builder.allowCoreThreadTimeOut(pool.isAllowCoreThreadTimeout());
		builder = builder.keepAlive(pool.getKeepAlive());
		Shutdown shutdown = properties.getShutdown();
		builder = builder.awaitTermination(shutdown.isAwaitTermination());
		builder = builder.awaitTerminationPeriod(shutdown.getAwaitTerminationPeriod());
		builder = builder.threadNamePrefix(properties.getThreadNamePrefix());
		builder = builder.customizers(taskExecutorCustomizers.orderedStream()::iterator);
		builder = builder.taskDecorator(taskDecorator.getIfUnique());
		return builder;
	}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24

有几种方式设置，这些也在 @EnableAsync 注解源码上有说明，最简单的方式就是修改配置文件

spring:
  task:
    execution:
      pool:
        max-size: 200
1
2
3
4
5

其次就是你自己创建一个 ThreadPoolTaskExecutor�，如下所示：

 @Configuration
  @EnableAsync
  public class AppConfig implements AsyncConfigurer {
 
      @Override
      public Executor getAsyncExecutor() {
          ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
          executor.setCorePoolSize(7);
          executor.setMaxPoolSize(42);
          executor.setQueueCapacity(11);
          executor.setThreadNamePrefix("MyExecutor-");
          executor.initialize();
          return executor;
      }
 
      @Override
      public AsyncUncaughtExceptionHandler getAsyncUncaughtExceptionHandler() {
          return new MyAsyncUncaughtExceptionHandler();
      }
  }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20

# 生产者消费者模式

将想要异步的工作放入队列中，由另外的线程去消费该队列。 这种方式也是异步化的一种。

← ReentrantReadWriteLock →