Java实现异步的几种方式

红太狼 2023-10-14 11:31 18阅读 0赞

**一、前言**

异步执行对于开发者来说并不陌生，在实际的开发过程中，很多场景多会使用到异步，相比同步执行，异步可以大大缩短请求链路耗时时间，比如：「发送短信、邮件、异步更新等」，这些都是典型的可以通过异步实现的场景。

**二、异步的五种实现方式**

1.  线程Thread
2.  Future
3.  异步框架CompletableFuture
4.  Spring注解@Async
5.  ThreadUtil异步工具类

**三、异步编程**

**3.1 线程异步**

public class AsyncThread extends Thread {
     
        @Override
        public void run() {
            System.out.println("Current thread name:" + Thread.currentThread().getName() + " Send email success!");
        }
     
        public static void main(String[] args) {
            AsyncThread asyncThread = new AsyncThread();
            asyncThread.start();
        }
    }

当然如果每次都创建一个Thread线程，频繁的创建、销毁，浪费系统资源，我们可以采用线程池：

private ExecutorService executorService = Executors.newCachedThreadPool();
     
    public void fun() {
        executorService.submit(new Runnable() {
            @Override
            public void run() {
                log.info("执行业务逻辑...");
            }
        });
    }

可以将业务逻辑封装到Runnable或Callable中，交由线程池来执行。

### **3.2 Future异步** ###

@Slf4j
    public class FutureManager {
     
        public String execute() throws Exception {
     
            ExecutorService executor = Executors.newFixedThreadPool(1);
            Future<String> future = executor.submit(new Callable<String>() {
                @Override
                public String call() throws Exception {
     
                    System.out.println(" --- task start --- ");
                    Thread.sleep(3000);
                    System.out.println(" --- task finish ---");
                    return "this is future execute final result!!!";
                }
            });
     
            //这里需要返回值时会阻塞主线程
            String result = future.get();
            log.info("Future get result: {}", result);
            return result;
        }
     
        @SneakyThrows
        public static void main(String[] args) {
            FutureManager manager = new FutureManager();
            manager.execute();
        }
    }

**3.2.1 Future的不足之处**  
Future的不足之处的包括以下几点：

1️⃣ 无法被动接收异步任务的计算结果：虽然我们可以主动将异步任务提交给线程池中的线程来执行，但是待异步任务执行结束之后，主线程无法得到任务完成与否的通知，它需要通过get方法主动获取任务执行的结果。

2️⃣ Future件彼此孤立：有时某一个耗时很长的异步任务执行结束之后，你想利用它返回的结果再做进一步的运算，该运算也会是一个异步任务，两者之间的关系需要程序开发人员手动进行绑定赋予，Future并不能将其形成一个任务流（pipeline），每一个Future都是彼此之间都是孤立的，所以才有了后面的CompletableFuture，CompletableFuture就可以将多个Future串联起来形成任务流。

3️⃣ Futrue没有很好的错误处理机制：截止目前，如果某个异步任务在执行发的过程中发生了异常，调用者无法被动感知，必须通过捕获get方法的异常才知晓异步任务执行是否出现了错误，从而在做进一步的判断处理。

### **3.3 CompletableFuture实现异步** ###

> CompletableFuture的用法主要可概括为以下几点
> 
> 简单用法 get() 与 complete()
> 
> 提交任务 runAsync() 与 supplyAsync()
> 
> 链式处理 theRun()、thenAccept() 和 thenApply()
> 
> 组合处理 thenCompose() 与 thenCombine() 、allOf与anyOf()

runAsync 异步执行，无返回值  
supplyAsync 异步执行，有返回值

CompletableFuture allOf 和 anyOf

anyOf 的含义是只要有任意一个 CompletableFuture 结束，就可以做 接下来的事情，而无须像 AllOf 那样，等待所有的 CompletableFuture 结束。

//自定义线程池
    private static final ThreadPoolExecutor EXECUTOR = new ThreadPoolExecutor(20, 20,
                0L, TimeUnit.MILLISECONDS,
                new LinkedBlockingQueue<>(100),
                new ThreadFactoryBuilder().setNameFormat("export-pay-record-info-%d").build(),
                new ThreadPoolExecutor.CallerRunsPolicy());
    //获取所有异步线程的返回值之后结束
    CompletableFuture.allOf(
                    // 任务1
                    CompletableFuture.runAsync(() -> resultMap.put("Test1", “。。。”), EXECUTOR),
                    // 任务2
                    CompletableFuture.runAsync(() -> resultMap.put("Test2", “。。。”), EXECUTOR),
                    // 任务3
                    CompletableFuture.runAsync(() -> resultMap.put("Test3", “。。。”), EXECUTOR),
                    // 任务4
                    CompletableFuture.runAsync(() -> resultMap.put("Test4", “。。。”), EXECUTOR)
            ).join();

public class CompletableFutureCompose {
     
        /**
         * thenAccept子任务和父任务公用同一个线程
         */
        @SneakyThrows
        public static void thenRunAsync() {
            CompletableFuture<Integer> cf1 = CompletableFuture.supplyAsync(() -> {
                System.out.println(Thread.currentThread() + " cf1 do something....");
                return 1;
            });
            CompletableFuture<Void> cf2 = cf1.thenRunAsync(() -> {
                System.out.println(Thread.currentThread() + " cf2 do something...");
            });
            //等待任务1执行完成
            System.out.println("cf1结果->" + cf1.get());
            //等待任务2执行完成
            System.out.println("cf2结果->" + cf2.get());
        }
     
        public static void main(String[] args) {
            thenRunAsync();
        }
    }

我们不需要显式使用ExecutorService，CompletableFuture 内部使用了ForkJoinPool来处理异步任务，如果在某些业务场景我们想自定义自己的异步线程池也是可以的。

### **3.4 Spring的@Async异步** ###

#### **3.4.1 自定义异步线程池** ####

/**
     * 线程池参数配置，多个线程池实现线程池隔离，@Async注解，默认使用系统自定义线程池，可在项目中设置多个线程池，在异步调用的时候，指明需要调用的线程池名称，比如：@Async("taskName")
    @EnableAsync
    @Configuration
    public class TaskPoolConfig {
        /**
         * 自定义线程池
         *
         **/
        @Bean("taskExecutor")
        public Executor taskExecutor() {
            //返回可用处理器的Java虚拟机的数量 12
            int i = Runtime.getRuntime().availableProcessors();
            System.out.println("系统最大线程数  ：" + i);
            ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
            //核心线程池大小
            executor.setCorePoolSize(16);
            //最大线程数
            executor.setMaxPoolSize(20);
            //配置队列容量，默认值为Integer.MAX_VALUE
            executor.setQueueCapacity(99999);
            //活跃时间
            executor.setKeepAliveSeconds(60);
            //线程名字前缀
            executor.setThreadNamePrefix("asyncServiceExecutor -");
            //设置此执行程序应该在关闭时阻止的最大秒数，以便在容器的其余部分继续关闭之前等待剩余的任务完成他们的执行
            executor.setAwaitTerminationSeconds(60);
            //等待所有的任务结束后再关闭线程池
            executor.setWaitForTasksToCompleteOnShutdown(true);
            return executor;
        }
    }

#### **3.4.2 AsyncService** ####

public interface AsyncService {
     
        MessageResult sendSms(String callPrefix, String mobile, String actionType, String content);
     
        MessageResult sendEmail(String email, String subject, String content);
    }
     
    @Slf4j
    @Service
    public class AsyncServiceImpl implements AsyncService {
     
        @Autowired
        private IMessageHandler mesageHandler;
     
        @Override
        @Async("taskExecutor")
        public MessageResult sendSms(String callPrefix, String mobile, String actionType, String content) {
            try {
     
                Thread.sleep(1000);
                mesageHandler.sendSms(callPrefix, mobile, actionType, content);
     
            } catch (Exception e) {
                log.error("发送短信异常 -> ", e)
            }
        }
        
        @Override
        @Async("taskExecutor")
        public sendEmail(String email, String subject, String content) {
            try {
     
                Thread.sleep(1000);
                mesageHandler.sendsendEmail(email, subject, content);
     
            } catch (Exception e) {
                log.error("发送email异常 -> ", e)
            }
        }
    }

在实际项目中， 使用@Async调用线程池，推荐等方式是是使用自定义线程池的模式，不推荐直接使用@Async直接实现异步。

### **3.5 ThreadUtil异步工具类** ###

@Slf4j
    public class ThreadUtils {
     
        public static void main(String[] args) {
            for (int i = 0; i < 3; i++) {
                ThreadUtil.execAsync(() -> {
                    ThreadLocalRandom threadLocalRandom = ThreadLocalRandom.current();
                    int number = threadLocalRandom.nextInt(20) + 1;
                    System.out.println(number);
                });
                log.info("当前第：" + i + "个线程");
            }
     
            log.info("task finish!");
        }
    }