异步编程必须要知道的九种实现方式！

大家好，欢迎来到IT知识分享网。

异步编程

      异步编程是一种程序设计方法，它允许多个任务在不阻塞主线程的情况下并行执行。异步编程的核心在于提高程序的并发性能和响应能力，在实际的开发过程中，很多场景多都会使用到异步编程，相比同步执行，异步编程通过将耗时的操作放在后台运行，避免了主线程的长时间等待，从而不会冻结用户界面或阻塞后续操作。

优点：

缺点：

知道了具体的优缺点，那我们来看下异步编程具体可以怎么实现呢？

Java 具体实现方式：

一、Java原生线程Thread、ExecutorService和相关的ThreadPoolExecutor

1）原生线程Thread：

创建并启动新的Thread对象执行异步任务。

示例：

// 原生线程示例 new Thread(() -> { // 异步逻辑代码 }).start();

2）ExecutorService和相关的ThreadPoolExecutor

使用java.util.concurrent.ExecutorService和相关的ThreadPoolExecutor创建线程池来管理异步任务。

简单示例：

// 简单示例 ExecutorService executor = Executors.newFixedThreadPool(5); executor.submit(() -> { // 异步任务逻辑 }); // 记得在某个时候关闭ExecutorService // executorService.shutdown();

示例：

 import java.util.concurrent.*; public class ThreadPoolExample { public static void main(String[] args) { // 创建一个固定大小的线程池 int corePoolSize = 5; int maximumPoolSize = 10; long keepAliveTime = 60L; TimeUnit unit = TimeUnit.SECONDS; BlockingQueue<Runnable> workQueue = new LinkedBlockingQueue<>(); ThreadFactory threadFactory = Executors.defaultThreadFactory(); RejectedExecutionHandler handler = new ThreadPoolExecutor.AbortPolicy(); ExecutorService executorService = new ThreadPoolExecutor( corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, threadFactory, handler); // 提交任务到线程池 for (int i = 0; i < 20; i++) { final int taskId = i; executorService.submit(new Runnable() { @Override public void run() { System.out.println("Task ID: " + taskId + " is running by " + Thread.currentThread().getName()); // 异步任务逻辑 try { Thread.sleep(1000); } catch (InterruptedException e) { e.printStackTrace(); } } }); } // 关闭线程池 executorService.shutdown(); // 等待所有任务完成 while (!executorService.isTerminated()) { // 循环检查线程池是否已经终止,通常情况下， // 在实际应用中可能不需要这样阻塞等待，因为关闭线程池并不会立即停止正在执行的任务， // 而是不再接受新的任务并将等待队列中的任务处理完为止。 } System.out.println("所有任务已完成，线程池已关闭"); } }

二、Future

Future与Callable

java.util.concurrent.Future接口用于获取异步计算的结果，配合Callable接口可以返回一个结果给调用者。

简单示例：

 // 简单示例 Future<String> future = executor.submit(new Callable<String>() { @Override public String call() throws Exception { // 异步结果处理 return "Async Result"; } }); // 获取结果，可能会阻塞直到结果可用 String result = future.get();

示例：

import java.util.concurrent.*; public class FutureExample { public static void main(String[] args) { ExecutorService executorService = Executors.newFixedThreadPool(5); // 提交任务并获取Future对象 Future<Integer> future = executorService.submit(new Callable<Integer>() { @Override public Integer call() throws Exception { int result = someTimeConsumingOperation(); return result; } private int someTimeConsumingOperation() { Thread.sleep(2000); // 模拟异步代码逻辑耗时操作 return 42; // 返回结果 } }); // 主线程继续执行其他逻辑 System.out.println("主线程继续执行其他任务..."); try { // 获取Future的结果，如果任务还没完成，这将阻塞直到任务完成 Integer result = future.get(); System.out.println("Future任务的结果是: " + result); } catch (InterruptedException | ExecutionException e) { e.printStackTrace(); } // 关闭线程池 executorService.shutdown(); } } / 在这个例子中，我们创建了一个固定大小的线程池，并提交了一个实现了Callable接口的任务。 Callable与Runnable类似，但可以有返回值，其call()方法的返回值会被包装到Future对象中。 通过调用future.get()方法，我们可以获取任务的执行结果。如果任务还未完成，此方法会阻塞直到任务完成。 如果在任务执行过程中发生异常，get()方法会抛出ExecutionException。 */

三、CompletableFuture（Java 8及以上版本）

CompletableFuture（Java 8及以上版本）

java.util.concurrent.CompletableFuture是Java 8引入的一个强大的异步编程工具，它提供了对Future的扩展，支持链式调用、回调函数和组合多个Future等高级特性。

简单示例：

 // 简单示例 CompletableFuture.runAsync(() -> { // 异步执行的代码 System.out.println("异步任务执行中..."); }); // 返回结果的简单示例 CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> { // 模拟异步逻辑实现 try { Thread.sleep(1000); } catch (InterruptedException e) { throw new IllegalStateException(e); } return "异步任务结果"; }); // 获取异步任务的结果（如果还没完成会阻塞） String result = future.join(); System.out.println(result); 

示例：

import java.util.concurrent.CompletableFuture; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; public class CompletableFutureExample { public static void main(String[] args) { ExecutorService executor = Executors.newFixedThreadPool(5); // 创建一个CompletableFuture CompletableFuture<String> future1 = CompletableFuture.supplyAsync(() -> { try { // 模拟异步逻辑耗时操作 Thread.sleep(1000); } catch (InterruptedException e) { Thread.currentThread().interrupt(); } return "Hello, "; }, executor); CompletableFuture<String> future2 = CompletableFuture.supplyAsync(() -> { try { // 模拟异步逻辑耗时操作 Thread.sleep(1500); } catch (InterruptedException e) { Thread.currentThread().interrupt(); } return "World!"; }, executor); // 使用thenCombine将两个异步任务的结果合并 CompletableFuture<String> combinedFuture = future1.thenCombine(future2, (greeting, world) -> greeting + world); // 获取最终结果 combinedFuture.whenComplete((result, exception) -> { if (exception == null) { System.out.println("异步计算的结果是: " + result); } else { exception.printStackTrace(); } // 关闭线程池（实际应用中应确保所有任务完成后关闭） executor.shutdown(); }); } } / 在这个例子中，我们创建了两个CompletableFuture对象，每个都代表一个独立的异步任务。 然后，我们使用thenCombine方法将这两个任务的结果合并在一起。 当所有的异步任务完成时，我们通过whenComplete方法来处理最终结果或任何可能发生的异常。 请注意，为了保证资源的正确释放，在实际应用中应在所有异步任务执行完毕后再关闭ExecutorService。 在本例中为了简洁，直接在结果处理回调中关闭了线程池， 但在实际开发中，可能需要更复杂的逻辑来确保这一点。 */

四、Spring框架@Async注解

在Spring中，通过在方法上添加@Async注解，并配置一个合适的TaskExecutor，可以使该方法异步执行。

示例：

1）首先需要在配置类中启用异步支持

 @Configuration @EnableAsync public class AsyncConfig { // 可以自定义任务执行器 @Bean public TaskExecutor taskExecutor() { // 可以自定义任务执行器，此处示例略。。。 return new ConcurrentTaskExecutor(); } }

2）在服务类中使用@Async注解

 @Service public class AsyncService { @Async public void asyncMethod() { // 异步执行的任务 } }

五、Spring WebFlux与Reactor响应式编程

Spring Framework 5引入了WebFlux模块，支持非阻塞和反应式编程模型。它依赖于Project Reactor库，通过Mono和Flux类处理异步数据流。

Spring WebFlux是Spring框架中的一个模块，它提供了基于Reactor库的非阻塞、反应式编程模型，用于构建异步和高性能的应用程序。在WebFlux中，可以使用Mono和Flux来表示0-N个值（单值流和多值流）的异步处理。

简单示例：

 Mono.fromCallable(() -> { // 异步操作 return "Reactive Result"; }) .subscribe(result -> { System.out.println("Reactive Received: " + result); });

示例：

 import org.springframework.web.bind.annotation.GetMapping; import org.springframework.web.bind.annotation.RestController; import reactor.core.publisher.Mono; import java.time.LocalDateTime; import java.util.concurrent.CompletableFuture; @RestController public class AsyncController { @GetMapping("/async") public Mono<String> asyncCall() { // 模拟一个异步任务 return Mono.fromFuture(CompletableFuture.supplyAsync(() -> { // 模拟异步逻辑耗时操作 try { Thread.sleep(1000); } catch (InterruptedException e) { Thread.currentThread().interrupt(); } return "async response ：" + LocalDateTime.now(); })); } } / 在这个例子中，我们创建了一个WebFlux控制器，并定义了一个名为asyncCall的方法。 该方法返回一个Mono对象，这个Mono通过fromFuture方法从CompletableFuture转换而来。 CompletableFuture.supplyAsync方法用来执行一个模拟耗时的异步任务。 当客户端发起对该/async路径的GET请求时，服务器不会阻塞等待该异步任务完成， 而是立即返回响应头，然后在后台继续执行异步任务。一旦任务完成， 其结果将作为HTTP响应体的一部分发送给客户端。 */

此外，WebFlux还可以直接与Reactor结合，不借助CompletableFuture，如下所示：

 import org.springframework.web.bind.annotation.GetMapping; import org.springframework.web.bind.annotation.RestController; import reactor.core.publisher.Mono; import java.time.LocalDateTime; @RestController public class AsyncController { @GetMapping("/async") public Mono<String> asyncCall() { return Mono.fromCallable(() -> { // 模拟异步逻辑耗时操作 Thread.sleep(1000); return "async response ：" + LocalDateTime.now(); }); } } // 此例使用了Mono.fromCallable方法，它接受一个Callable， // 将其转换为Mono，从而实现了更原生的反应式编程风格。

六、第三方库

1）Guava的ListenableFuture：Google Guava库提供了一个可监听的Future扩展，增加了回调机制。当Future的结果可用时，你可以注册一个监听器（Listener）来处理结果或者异常。

示例：

 import com.google.common.util.concurrent.*; import java.util.concurrent.*; public class ListenableFutureExample { public static void main(String[] args) throws ExecutionException, InterruptedException { ListeningExecutorService executor = MoreExecutors.listeningDecorator(Executors.newFixedThreadPool(5)); // 创建一个ListenableFuture ListenableFuture<String> future = executor.submit(() -> { // 异步逻辑实现 Thread.sleep(1000); return "Hello, World!"; }); // 添加一个成功回调(添加一个FutureCallback监听器) Futures.addCallback(future, new FutureCallback<String>() { @Override public void onSuccess(String result) { System.out.println("任务完成，结果是: " + result); } @Override public void onFailure(Throwable t) { System.out.println("任务执行失败: " + t.getMessage()); } }); // 关闭线程池 executor.shutdown(); // 等待所有任务完成(在实际开发中，应确保所有任务完成后再关闭线程池以防止丢失数据或中断正在进行的任务) executor.awaitTermination(1, TimeUnit.MINUTES); } } / 通过Futures.addCallback方法，我们可以添加一个FutureCallback监听器。 当ListenableFuture的任务完成后，无论成功还是失败，都会调用相应的回调方法。 */

2）RxJava：基于Observable的反应式编程库，提供了丰富的操作符处理异步事件序列。它使用扩展的观察者模式来创建可观察序列（Observable），并允许订阅者（Observer）对这些序列中的数据变化做出反应。

示例：

 import io.reactivex.rxjava3.core.Observable; import io.reactivex.rxjava3.schedulers.Schedulers; public class RxJavaAsyncExample { public static void main(String[] args) { // 创建一个发出异步操作结果的Observable Observable<String> observable = Observable.fromCallable(() -> { // 异步逻辑实现 Thread.sleep(1000); return "Hello, Async World!"; }); // 将这个Observable调度到一个新的线程上执行(此处实际的任务将在后台IO线程池中异步执行) Observable<String> asyncObservable = observable.subscribeOn(Schedulers.io()); // 订阅Observable，并定义如何处理发射的数据或错误 // 当Observable完成异步计算并发射结果时，相应的回调会被调用 asyncObservable.subscribe( result -> System.out.println("Result asynchronously: " + result), error -> System.out.println("Error occurred: " + error.getMessage()), () -> System.out.println("Asynchronous task") ); } }

3）Hutool工具包中的异步工具类

例如ThreadUtil提供了便捷的异步执行工具方法。

示例：

import cn.hutool.core.thread.ThreadUtil; public class HutoolAsyncExample { public static void main(String[] args) { // 使用默认全局线程池执行异步任务 ThreadUtil.execAsync(() -> { try { // 模拟异步逻辑实现 Thread.sleep(1000); } catch (InterruptedException e) { Thread.currentThread().interrupt(); } System.out.println("异步任务执行完成，当前线程：" + Thread.currentThread().getName()); }); // 如果需要自定义线程池，则可以先创建一个线程池，然后使用它来执行任务 ExecutorService customExecutor = ThreadUtil.newExecutor(5, "custom-pool"); ThreadUtil.execute(customExecutor, () -> { try { Thread.sleep(1000); } catch (InterruptedException e) { Thread.currentThread().interrupt(); } System.out.println("使用自定义线程池执行的任务完成，当前线程：" + Thread.currentThread().getName()); }); // 关闭自定义线程池（在实际开发中，确保所有任务完成后关闭） ThreadUtil.shutdown(customExecutor); } }

七、消息队列（如RabbitMQ、 Apache Kafka）

消息队列虽然不是严格意义上的编程实现方式，但消息队列常用于分布式系统的异步通信，发送消息到队列后，消费者在后台异步处理这些消息。

简单示例：（RabbitMQ）

    基于RabbitMQ的Spring框架支持来实现异步处理

1）添加依赖：

 <!-- Spring Boot 中 --> <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-amqp</artifactId> </dependency> <!-- 或者非Spring Boot项目中 --> <dependency> <groupId>org.springframework.amqp</groupId> <artifactId>spring-rabbit</artifactId> <version>使用的版本号</version> </dependency>

2）编写配置类

 @Configuration public class RabbitMQConfig { @Value("${spring.rabbitmq.host}") private String host; @Value("${spring.rabbitmq.port}") private int port; @Value("${spring.rabbitmq.username}") private String username; @Value("${spring.rabbitmq.password}") private String password; @Bean public ConnectionFactory connectionFactory() { CachingConnectionFactory factory = new CachingConnectionFactory(host, port); factory.setUsername(username); factory.setPassword(password); return factory; } @Bean public Queue myQueue() { return new Queue("myAsyncQueue", true); } @Bean public TopicExchange myExchange() { return new TopicExchange("myExchange"); } @Bean public Binding binding(Queue queue, TopicExchange exchange) { return BindingBuilder.bind(queue).to(exchange).with("routingKey.*"); } }

3）创建一个消息生产者，用于发送异步任务

@Component public class AsyncTaskProducer { @Autowired private RabbitTemplate rabbitTemplate; public void sendMessage(String message) { CorrelationData correlationData = new CorrelationData(UUID.randomUUID().toString()); rabbitTemplate.convertAndSend("myExchange", "routingKey.task", message, correlationData); System.out.println("Send asyn message: " + message); } }

4）创建一个消息消费者，处理异步任务

 @Service @RabbitListener(queues = "myAsyncQueue") public class AsyncTaskConsumer { @RabbitHandler public void receiveMessage(String message) { // 异步业务逻辑 System.out.println("asyn task: " + message); } }

八、Spring ApplicationEvent事件实现异步

在Spring框架中，事件驱动模型（Event-Driven Model）也可以用来实现异步编程。通过发布和监听事件，可以在不同的组件之间解耦，并且可以将耗时的任务放在事件监听器中异步执行。

示例：

1）定义事件（创建一个事件类，它通常是一个简单的Java对象，用于携带必要的信息。）

 import org.springframework.context.ApplicationEvent; public class MyCustomEvent extends ApplicationEvent { private String eventData; public MyCustomEvent(Object source, String eventData) { super(source); this.eventData = eventData; } public String getEventData() { return eventData; } }

2）发布事件

 @Service public class EventPublisherService { @Autowired private ApplicationEventPublisher eventPublisher; public void publishAsyncEvent(String data) { // 发布自定义事件 eventPublisher.publishEvent(new MyCustomEvent(this, data)); } }

3)创建事件监听器

 @Component public class MyCustomEventListener implements ApplicationListener<MyCustomEvent> { @Override public void onApplicationEvent(MyCustomEvent event) { // 异步处理事件 String eventData = event.getEventData(); // 这里执行耗时任务 executeAsyncTask(eventData); } private void executeAsyncTask(String data) { // 模拟异步逻辑执行 try { Thread.sleep(3000); } catch (InterruptedException e) { e.printStackTrace(); } System.out.println("异步事件: " + data); } }

当调用publishAsyncEvent方法发布事件时，Spring会自动查找并调用所有注册的、能处理MyCustomEvent类型的ApplicationListener，即MyCustomEventListener。

在这个监听器中的executeAsyncTask方法会在发布事件后的某个时刻异步执行，从而实现了异步编程。

九、NIO (Non-blocking I/O)

Java NIO（New I/O）提供了对非阻塞 I/O 操作的支持，使得开发人员可以实现高效的 I/O 操作，特别是在处理大量并发连接时。虽然 NIO 本身并不直接等同于异步编程，但它为实现异步 I/O 提供了基础。

在 Java NIO 中，有几个关键组件支持异步编程：

1. Selector（选择器）：Selector 是 Java NIO 中能够实现非阻塞 I/O 的核心对象。它可以检测多个注册在其上的 Channel（通道）的状态变化，如读就绪、写就绪等。这样，单个线程可以管理多个 Channel，提高了系统的可伸缩性。
2. Channel（通道）：Channel 是 Java NIO 中用于数据传输的对象。与传统的 I/O 操作使用 InputStream 和 OutputStream 不同，NIO 使用 Channel 进行读写操作。Channel 可以是异步的，也可以是同步的，但通常与 Selector 配合使用以实现非阻塞 I/O。
3. Buffer（缓冲区）：Buffer 是用于在 Channel 和应用程序之间传输数据的容器。数据从 Channel 读取到 Buffer 中，或者从 Buffer 写入到 Channel 中。Buffer 的使用减少了不必要的内存分配和拷贝，提高了数据传输的效率。

虽然 Java NIO 提供了非阻塞 I/O 的基础，但它本身并不直接支持真正的异步编程模型，如回调或 Promises/Futures。然而，通过使用 Selector 和非阻塞 Channel，我们是可以构建出类似异步的行为。

import java.io.IOException; import java.net.InetSocketAddress; import java.nio.ByteBuffer; import java.nio.channels.SelectionKey; import java.nio.channels.Selector; import java.nio.channels.ServerSocketChannel; import java.nio.channels.SocketChannel; import java.util.Iterator; import java.util.Set; public class NIOServer { public static void main(String[] args) throws IOException { // 创建一个 ServerSocketChannel ServerSocketChannel serverSocketChannel = ServerSocketChannel.open(); // 设置为非阻塞模式 serverSocketChannel.configureBlocking(false); // 绑定监听端口 serverSocketChannel.bind(new InetSocketAddress(8080)); // 创建一个 Selector Selector selector = Selector.open(); // 注册 ServerSocketChannel，关心 ACCEPT 事件（新的连接） serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT); while (true) { // 阻塞等待就绪的 Channel 数量 int readyChannels = selector.select(); if (readyChannels == 0) continue; // 获取可用 Channel 的集合 Set<SelectionKey> selectedKeys = selector.selectedKeys(); Iterator<SelectionKey> keyIterator = selectedKeys.iterator(); // 遍历 SelectionKey while (keyIterator.hasNext()) { SelectionKey key = keyIterator.next(); // 检查是否有新的连接 if (key.isAcceptable()) { ServerSocketChannel server = (ServerSocketChannel) key.channel(); // 非阻塞，立即返回 SocketChannel client = server.accept(); // 设置为非阻塞模式 client.configureBlocking(false); // 注册读事件 client.register(selector, SelectionKey.OP_READ); System.out.println("Accepted new connection from " + client); } // 检查是否有数据可读 else if (key.isReadable()) { SocketChannel client = (SocketChannel) key.channel(); ByteBuffer buffer = ByteBuffer.allocate(1024); // 非阻塞读操作 int bytesRead = client.read(buffer); if (bytesRead > 0) { buffer.flip(); // 处理读取到的数据... System.out.println("Received: " + new String(buffer.array()).trim()); } else if (bytesRead < 0) { // 连接关闭 client.close(); } } // 处理完后，必须从集合中移除 SelectionKey keyIterator.remove(); } } } }

注意：

      虽然上述代码使用了非阻塞的 Channel 和 Selector，但它并没有实现真正的异步回调机制。在处理大量并发连接时，这种基于事件驱动的非阻塞模型可以显著提高性能，因为它允许单个线程管理多个连接而无需为每个连接分配一个单独的线程。然而，对于复杂的异步编程场景，可能需要考虑使用更高级的库，如 CompletableFuture 或 Reactor。

如果对你有帮助的话，点个赞吧~^-^~