当前位置： 代码网 > it编程>编程语言>Java > Java 中多线程异步处理最佳实践

Java 中多线程异步处理最佳实践

2025年12月10日 • Java •我要评论
第一部分: completablefuture 级联调用以下代码展示了 supplyasync（异步执行有返回值任务）和 thenapplyasync（异步处理前序任务结果）的级联调用，包含完整的异常

第一部分: completablefuture 级联调用

以下代码展示了 supplyasync（异步执行有返回值任务）和 thenapplyasync（异步处理前序任务结果）的级联调用，包含完整的异常处理、资源管理和 java 17+ 语法特性。

import lombok.extern.slf4j.slf4j;
import org.springframework.stereotype.service;
import java.util.concurrent.completablefuture;
import java.util.concurrent.executorservice;
import java.util.concurrent.executors;
import java.util.concurrent.timeunit;
/**
 * 异步任务级联调用示例服务
 * 使用 java 17+ 特性 + spring 注解 + 完整异常处理
 */
@slf4j
@service
public class asynccascadeservice {
    // 自定义线程池（避免使用默认 forkjoinpool，提升可控性）
    private static final executorservice custom_executor = executors.newfixedthreadpool(5);
    /**
     * 级联异步任务示例：
     * 1. supplyasync：异步执行任务1（生成订单id）
     * 2. thenapplyasync：异步处理任务1结果（根据订单id查询订单详情）
     * 3. thenapplyasync：异步处理任务2结果（计算订单金额）
     */
    public completablefuture<orderamountresult> processorderasync(long userid) {
        // 第一个异步任务：生成订单id（supplyasync 执行有返回值的异步任务）
        completablefuture<string> generateorderidfuture = completablefuture.supplyasync(() -> {
            log.info("异步任务1：生成订单id，线程：{}", thread.currentthread().getname());
            // 模拟业务耗时
            simulatedelay(100);
            return "order_" + userid + "_" + system.currenttimemillis();
        }, custom_executor) // 指定自定义线程池（推荐，避免默认池耗尽）
        .exceptionally(ex -> {
            log.error("任务1执行失败：", ex);
            throw new runtimeexception("生成订单id失败", ex);
        });
        // 第二个异步任务：处理订单id，查询订单详情（thenapplyasync 异步处理前序结果）
        completablefuture<orderdetail> queryorderdetailfuture = generateorderidfuture
                .thenapplyasync(orderid -> {
                    log.info("异步任务2：查询订单详情，订单id：{}，线程：{}", orderid, thread.currentthread().getname());
                    simulatedelay(150);
                    // 模拟查询结果
                    return new orderdetail(orderid, userid, "商品a", 2, 99.9);
                }, custom_executor)
                .exceptionally(ex -> {
                    log.error("任务2执行失败：", ex);
                    throw new runtimeexception("查询订单详情失败", ex);
                });
        // 第三个异步任务：计算订单总金额（级联处理前序结果）
        return queryorderdetailfuture
                .thenapplyasync(orderdetail -> {
                    log.info("异步任务3：计算订单金额，订单详情：{}，线程：{}", orderdetail, thread.currentthread().getname());
                    simulatedelay(100);
                    double totalamount = orderdetail.price() * orderdetail.quantity();
                    // 使用 java 17 record 简化数据返回
                    return new orderamountresult(orderdetail.orderid(), totalamount, totalamount * 0.08);
                }, custom_executor)
                .exceptionally(ex -> {
                    log.error("任务3执行失败：", ex);
                    throw new runtimeexception("计算订单金额失败", ex);
                });
    }
    // 模拟业务耗时
    private void simulatedelay(long millis) {
        try {
            timeunit.milliseconds.sleep(millis);
        } catch (interruptedexception e) {
            thread.currentthread().interrupt();
            throw new runtimeexception("任务被中断", e);
        }
    }
    // java 17 record：订单详情
    public record orderdetail(string orderid, long userid, string productname, int quantity, double price) {}
    // java 17 record：订单金额计算结果
    public record orderamountresult(string orderid, double totalamount, double tax) {}
    // 资源关闭：spring 销毁时关闭线程池（try-with-resources 思想）
    @override
    protected void finalize() throws throwable {
        try (executorservice executor = custom_executor) {
            executor.shutdown();
            if (!executor.awaittermination(5, timeunit.seconds)) {
                executor.shutdownnow();
            }
            log.info("异步线程池已关闭");
        } finally {
            super.finalize();
        }
    }
    // 测试入口
    public static void main(string[] args) {
        asynccascadeservice service = new asynccascadeservice();
        // 执行级联异步任务
        completablefuture<orderamountresult> resultfuture = service.processorderasync(1001l);
        // 阻塞获取结果（实际业务中建议使用 thenaccept/thenrun 非阻塞处理）
        resultfuture.whencomplete((result, ex) -> {
            if (ex != null) {
                log.error("级联任务执行失败", ex);
            } else {
                log.info("级联任务执行完成：{}", result);
            }
            // 关闭线程池
            custom_executor.shutdown();
        }).join();
    }
}

核心知识点解释

  • supplyasync 作用
    • 异步执行有返回值的任务，参数是 supplier<t>（无入参、有返回值）
    • 第二个参数指定自定义线程池（推荐，避免使用 jvm 默认的 forkjoinpool.commonpool()，防止核心线程被耗尽）
  • thenapplyasync 作用
    • 异步处理前序 completablefuture 的结果，参数是 function<t, r>（接收前序结果、返回新结果）
    • 与 thenapply 的区别：thenapply 使用前序任务的线程执行，thenapplyasync 提交到线程池异步执行，真正实现“级联异步”
  • java 17+ 特性使用
    • record：简化数据载体（orderdetail/orderamountresult），自动生成构造器、getter、equals/hashcode 等
    • 异常处理：exceptionally 捕获前序任务异常并兜底，whencomplete 统一处理最终结果/异常
  • 资源管理
    • 线程池使用 try-with-resources 思想关闭（finalize 方法中通过 try-with-resources 语法确保线程池关闭）
    • 中断处理：捕获 interruptedexception 后恢复线程中断状态，避免线程状态丢失
  • 级联调用逻辑
  • 生成订单id（supplyasync）→ 查询订单详情（thenapplyasync）→ 计算金额（thenapplyasync）
    • 每个步骤都异步执行，且后序任务依赖前序任务的结果，实现“流水线式”异步处理。

最佳实践

  • 避免在 thenapplyasync 中执行耗时过长的任务，建议拆分多个小任务
  • 优先使用自定义线程池，通过 @configuration 配置线程池参数（核心线程数、最大线程数、队列等）
  • 异常处理：每个异步步骤都建议添加 exceptionally 或 handle，避免异常穿透导致整个链路失败
  • 非阻塞处理结果：使用 thenaccept（消费结果）/thenrun（无结果处理）替代 join() 阻塞调用

第二部分： 非阻塞处理结果

以下示例移除阻塞的 join()，完全基于 thenaccept（消费结果）、thenrun（无结果收尾）实现非阻塞处理，并补充 whencomplete 统一异常处理，同时优化 spring 规范的资源关闭（替换不推荐的 finalize() 为 @predestroy）。

import lombok.extern.slf4j.slf4j;
import org.springframework.stereotype.service;
import org.springframework.beans.factory.annotation.predestroy;
import java.util.concurrent.completablefuture;
import java.util.concurrent.executorservice;
import java.util.concurrent.executors;
import java.util.concurrent.timeunit;
/**
 * 非阻塞处理 completablefuture 结果示例
 * 核心：thenaccept（消费结果）/thenrun（收尾动作）替代 join() 阻塞调用
 */
@slf4j
@service
public class nonblockingasyncservice {
    // 自定义线程池（spring 中建议通过 @bean 配置，此处简化）
    private final executorservice customexecutor = executors.newfixedthreadpool(5);
    // 订单状态枚举（java 17+ 增强枚举，可结合 switch 表达式）
    public enum orderstatus { success, failure }
    /**
     * 级联异步任务（非阻塞核心逻辑）
     */
    public void processordernonblocking(long userid) {
        // 步骤1：异步生成订单id（supplyasync）
        completablefuture<string> orderidfuture = completablefuture.supplyasync(() -> {
            log.info("【异步任务1】生成订单id | 线程：{}", thread.currentthread().getname());
            simulatedelay(100);
            return "order_" + userid + "_" + system.currenttimemillis();
        }, customexecutor)
        .exceptionally(ex -> {
            log.error("【任务1失败】生成订单id异常", ex);
            throw new runtimeexception("生成订单id失败", ex);
        });
        // 步骤2：异步查询订单详情（thenapplyasync）
        completablefuture<orderdetail> orderdetailfuture = orderidfuture
                .thenapplyasync(orderid -> {
                    log.info("【异步任务2】查询订单详情 | 订单id：{} | 线程：{}", orderid, thread.currentthread().getname());
                    simulatedelay(150);
                    return new orderdetail(orderid, userid, "java编程实战", 2, 89.9);
                }, customexecutor)
                .exceptionally(ex -> {
                    log.error("【任务2失败】查询订单详情异常", ex);
                    throw new runtimeexception("查询订单详情失败", ex);
                });
        // 步骤3：非阻塞处理最终结果（核心：thenaccept + thenrun）
        orderdetailfuture
                // 1. whencomplete：统一处理结果/异常（非阻塞）
                .whencomplete((detail, ex) -> {
                    if (ex != null) {
                        log.error("【级联任务失败】订单处理异常", ex);
                        // 非阻塞处理失败逻辑（如通知、落库）
                        handleorderfailure(userid, ex.getmessage());
                    } else {
                        log.info("【级联任务成功】订单详情：{}", detail);
                    }
                })
                // 2. thenaccept：消费结果（有返回值时用，非阻塞）
                .thenaccept(detail -> {
                    log.info("【非阻塞消费】计算订单金额 | 订单id：{}", detail.orderid());
                    double total = detail.quantity() * detail.price();
                    // 模拟：非阻塞更新订单金额到数据库
                    updateorderamount(detail.orderid(), total);
                })
                // 3. thenrun：无结果收尾动作（非阻塞）
                .thenrun(() -> {
                    log.info("【非阻塞收尾】订单处理流程结束 | 清理临时资源/发送通知");
                    // 模拟：非阻塞发送处理完成通知
                    sendordernotify(userid, orderstatus.success);
                });
        // 关键：当前线程（如主线程/controller线程）不会阻塞，直接继续执行后续逻辑
        log.info("【主线程不阻塞】订单处理请求已提交，当前线程继续执行其他任务...");
    }
    // ------------------- 模拟业务方法 -------------------
    private void simulatedelay(long millis) {
        try {
            timeunit.milliseconds.sleep(millis);
        } catch (interruptedexception e) {
            thread.currentthread().interrupt();
            throw new runtimeexception("任务被中断", e);
        }
    }
    private void handleorderfailure(long userid, string errormsg) {
        log.warn("【失败处理】用户{}订单处理失败：{}", userid, errormsg);
        // 实际场景：非阻塞发送失败通知、记录异常日志等
    }
    private void updateorderamount(string orderid, double total) {
        log.info("【更新金额】订单{}总金额：{}元", orderid, total);
        // 实际场景：非阻塞调用dao更新数据库
    }
    private void sendordernotify(long userid, orderstatus status) {
        // java 17+ switch表达式（简化分支逻辑）
        string msg = switch (status) {
            case success -> "用户" + userid + "订单处理完成";
            case failure -> "用户" + userid + "订单处理失败";
        };
        log.info("【发送通知】{}", msg);
    }
    // ------------------- java 17+ 特性 -------------------
    // record：简化数据载体（自动生成构造器、getter、equals等）
    public record orderdetail(string orderid, long userid, string productname, int quantity, double price) {}
    // ------------------- 资源管理（spring规范） -------------------
    // spring 销毁bean时关闭线程池（try-with-resources 确保资源释放）
    @predestroy
    public void destroy() {
        log.info("【关闭线程池】开始释放异步线程池资源...");
        // try-with-resources 自动关闭 executorservice（java 9+ 支持）
        try (executorservice executor = customexecutor) {
            executor.shutdown();
            if (!executor.awaittermination(3, timeunit.seconds)) {
                executor.shutdownnow();
                log.warn("【强制关闭】线程池未正常终止，强制关闭");
            }
        } catch (interruptedexception e) {
            customexecutor.shutdownnow();
            thread.currentthread().interrupt();
            log.error("【关闭异常】线程池关闭被中断", e);
        }
        log.info("【关闭完成】异步线程池已释放");
    }
    // ------------------- 测试入口（非阻塞效果演示） -------------------
    public static void main(string[] args) {
        nonblockingasyncservice service = new nonblockingasyncservice();
        // 提交异步任务（非阻塞）
        service.processordernonblocking(1001l);
        // 主线程继续执行（不会等待异步任务完成）
        log.info("【主线程】已提交订单处理任务，开始执行其他业务逻辑...");
        simulatemainthreadwork();
        // 等待异步任务完成（仅测试用，实际业务无需此步骤）
        try {
            timeunit.seconds.sleep(2);
        } catch (interruptedexception e) {
            thread.currentthread().interrupt();
        }
    }
    private static void simulatemainthreadwork() {
        log.info("【主线程】执行其他任务：检查库存、验证用户权限...");
        try {
            timeunit.milliseconds.sleep(500);
        } catch (interruptedexception e) {
            thread.currentthread().interrupt();
        }
    }
}

核心知识点：非阻塞处理的关键区别

方法作用是否阻塞适用场景
join()阻塞当前线程，等待异步任务返回结果是测试/必须同步获取结果的场景
thenaccept()异步消费结果（入参为结果，无返回值）否有结果需要处理（如更新/通知）
thenrun()异步执行收尾动作（无入参、无返回值）否结果处理完成后的清理/日志
whencomplete异步处理结果+异常（入参为结果+异常）否统一捕获异常+分支处理

非阻塞处理的核心优势

  1. 提升系统吞吐量：提交异步任务的线程（如 controller 线程）无需等待，可立即处理下一个请求，避免线程阻塞导致的资源浪费；
  2. 避免线程死锁/饥饿：阻塞式 join() 可能导致线程池耗尽，非阻塞回调更符合异步编程的设计初衷；
  3. 逻辑解耦：结果消费、异常处理、收尾动作拆分到不同回调中，代码更清晰（符合单一职责）。

关键注意事项

  1. 异常处理：非阻塞场景下，必须通过 whencomplete/exceptionally 捕获异常，否则异常会被“吞掉”（无感知失败）；
  2. 线程池选择：thenaccept/thenrun 默认使用前序任务的线程池，建议统一指定自定义线程池（避免默认池耗尽）；
  3. 资源生命周期：spring 中通过 @predestroy 关闭线程池（替代 finalize()，finalize() 已被标记为过时），结合 try-with-resources 确保资源释放；
  4. java 17+ 增强：示例中使用 switch 表达式（替代传统 switch）、record（简化数据类），符合最新语法规范。

输出效果（非阻塞特征）

运行代码后，控制台会先打印主线程不阻塞的日志，再异步打印各任务的日志，体现“提交任务后主线程立即继续执行”的非阻塞特性：

【主线程不阻塞】订单处理请求已提交，当前线程继续执行其他任务...
【主线程】执行其他任务：检查库存、验证用户权限...
【异步任务1】生成订单id | 线程：pool-1-thread-1
【异步任务2】查询订单详情 | 订单id：order_1001_17338xxxxxx | 线程：pool-1-thread-2
【级联任务成功】订单详情：orderdetail[orderid=order_1001_17338xxxxxx, userid=1001, productname=java编程实战, quantity=2, price=89.9]
【非阻塞消费】计算订单金额 | 订单id：order_1001_17338xxxxxx
【更新金额】订单order_1001_17338xxxxxx总金额：179.8元
【非阻塞收尾】订单处理流程结束 | 清理临时资源/发送通知
【发送通知】用户1001订单处理完成

第三部分 其它异步编程方式全解析

java 异步编程体系从基础线程模型到现代响应式编程、虚拟线程逐步演进，以下是主流方式的代码示例+特性解析，均基于 java 17+ 语法（部分如虚拟线程需 java 21+ 正式支持，17+ 可通过预览特性启用），并包含异常处理、资源管理规范。

一、基础核心：线程池 + future（java 5+）

executorservice 结合 future 是 completablefuture 之前的核心异步方式，通过提交 callable/runnable 实现异步任务，缺点是结果获取阻塞、无链式调用，但胜在简单可控。

代码示例（java 17+）

import java.util.concurrent.*;
import java.util.concurrent.atomic.atomiclong;
/**
 * 线程池 + future 异步示例
 * 核心：提交callable获取future，通过get()阻塞获取结果（可设置超时）
 */
public class futureasyncdemo {
    // 自定义线程池（java 17+ 推荐使用executors静态工厂+try-with-resources关闭）
    private static final executorservice executor = executors.newvirtualthreadpertaskexecutor(); // java 21+ 虚拟线程池（17+需预览）
    // java 17 record：异步任务结果
    public record taskresult(long taskid, string content, long costtimems) {}
    // 异步任务：生成业务数据
    private static callable<taskresult> generatedatatask(long taskid) {
        return () -> {
            long start = system.currenttimemillis();
            log("任务{}开始执行，线程：{}", taskid, thread.currentthread().getname());
            // 模拟耗时
            timeunit.milliseconds.sleep(200);
            long cost = system.currenttimemillis() - start;
            return new taskresult(taskid, "异步生成数据_" + taskid, cost);
        };
    }
    public static void main(string[] args) {
        // 提交异步任务
        future<taskresult> future1 = executor.submit(generatedatatask(1l));
        future<taskresult> future2 = executor.submit(generatedatatask(2l));
        // 非阻塞提交后，主线程可执行其他逻辑
        log("主线程执行其他任务...");
        // 阻塞获取结果（可设置超时，避免永久阻塞）
        try {
            taskresult result1 = future1.get(1, timeunit.seconds); // 超时时间1秒
            taskresult result2 = future2.get(1, timeunit.seconds);
            log("任务1结果：{}", result1);
            log("任务2结果：{}", result2);
        } catch (interruptedexception e) {
            thread.currentthread().interrupt();
            log("任务被中断：{}", e.getmessage());
        } catch (executionexception e) {
            log("任务执行失败：{}", e.getcause().getmessage());
        } catch (timeoutexception e) {
            log("任务超时：{}", e.getmessage());
            future1.cancel(true); // 超时取消任务
            future2.cancel(true);
        } finally {
            // try-with-resources 关闭线程池（java 9+ 支持executorservice自动关闭）
            try (executorservice executor = executor) {
                executor.shutdown();
                if (!executor.awaittermination(3, timeunit.seconds)) {
                    executor.shutdownnow();
                    log("线程池强制关闭");
                }
            } catch (interruptedexception e) {
                executor.shutdownnow();
                thread.currentthread().interrupt();
            }
        }
    }
    private static void log(string msg, object... args) {
        system.out.printf("[%s] %s%n", thread.currentthread().getname(), string.format(msg, args));
    }
}

核心特性

  • 优点：基础通用，兼容所有java版本，线程池可控；
  • 缺点：future.get() 阻塞线程，无链式调用、无异步异常处理，无法组合多个任务；
  • 适用场景：简单异步任务，无需结果联动的场景。

二、现代轻量：虚拟线程（virtualthread，java 21+ 正式/17+ 预览）

project loom 推出的虚拟线程（用户态线程）是 java 异步编程的重大升级，相比传统平台线程，虚拟线程轻量（百万级创建）、无上下文切换开销，天然适配异步场景。

代码示例（java 21+）

import java.time.duration;
import java.util.concurrent.executors;
import java.util.concurrent.timeunit;
/**
 * 虚拟线程（virtualthread）异步示例
 * java 21+ 正式支持，17+ 需添加 --enable-preview 启动参数
 */
public class virtualthreadasyncdemo {
    // java 17 record：用户数据
    public record userdata(long userid, string username, string email) {}
    // 异步任务：查询用户信息
    private static userdata queryuser(long userid) {
        try {
            // 模拟io阻塞（虚拟线程阻塞时会释放载体线程，无性能损耗）
            timeunit.milliseconds.sleep(300);
            log("查询用户{}完成，线程：{}", userid, thread.currentthread().getname());
            return new userdata(userid, "user_" + userid, userid + "@example.com");
        } catch (interruptedexception e) {
            thread.currentthread().interrupt();
            throw new runtimeexception("查询用户中断", e);
        }
    }
    public static void main(string[] args) throws interruptedexception {
        // 虚拟线程池（java 21+ 推荐）
        try (var executor = executors.newvirtualthreadpertaskexecutor()) {
            // 提交1000个虚拟线程任务（轻量，无性能压力）
            for (long i = 1; i <= 1000; i++) {
                long userid = i;
                executor.submit(() -> queryuser(userid));
            }
            // 主线程等待任务完成（仅测试用）
            executor.awaittermination(duration.ofseconds(5));
            log("所有虚拟线程任务执行完成");
        } // try-with-resources 自动关闭线程池
    }
    private static void log(string msg, object... args) {
        system.out.printf("[%s] %s%n", thread.currentthread().getname(), string.format(msg, args));
    }
}

核心特性

  • 优点：
    • 轻量：单个jvm可创建百万级虚拟线程，远超平台线程（千级）；
    • 无阻塞损耗：虚拟线程遇到io阻塞时，会释放底层载体线程（os线程），避免资源浪费；
    • 编程模型简单：无需回调地狱，同步写法实现异步效果；
  • 缺点：java 21+ 正式支持，低版本需升级；
  • 适用场景：io密集型异步任务（如数据库、网络调用），替代传统线程池+回调。

三、异步io：nio.2 异步通道（java 7+）

java nio.2 提供 asynchronousfilechannel/asynchronoussocketchannel 等异步io通道，基于事件驱动实现文件/网络io的异步操作，无需线程阻塞等待io完成。

代码示例（异步文件读写，java 17+）

import java.nio.bytebuffer;
import java.nio.channels.asynchronousfilechannel;
import java.nio.channels.completionhandler;
import java.nio.file.path;
import java.nio.file.standardopenoption;
import java.nio.charset.standardcharsets;
/**
 * nio.2 异步文件io示例
 * 核心：completionhandler 回调处理异步结果，无线程阻塞
 */
public class asyncfileiodemo {
    private static final path file_path = path.of("async_demo.txt");
    public static void main(string[] args) throws interruptedexception {
        // try-with-resources 管理异步文件通道（自动关闭资源）
        try (asynchronousfilechannel filechannel = asynchronousfilechannel.open(
                file_path,
                standardopenoption.read,
                standardopenoption.write,
                standardopenoption.create
        )) {
            // 步骤1：异步写入文件
            string content = "java 17 异步io示例：" + system.currenttimemillis();
            bytebuffer writebuffer = bytebuffer.wrap(content.getbytes(standardcharsets.utf_8));
            filechannel.write(writebuffer, 0, null, new completionhandler<integer, void>() {
                @override
                public void completed(integer byteswritten, void attachment) {
                    log("异步写入完成，写入字节数：{}", byteswritten);
                    // 写入完成后异步读取文件
                    readfileasync(filechannel);
                }
                @override
                public void failed(throwable exc, void attachment) {
                    log("异步写入失败：{}", exc.getmessage());
                }
            });
            // 主线程等待异步操作完成（仅测试用）
            thread.sleep(2000);
        } catch (exception e) {
            log("文件通道操作异常：{}", e.getmessage());
        }
    }
    // 异步读取文件
    private static void readfileasync(asynchronousfilechannel filechannel) {
        bytebuffer readbuffer = bytebuffer.allocate(1024);
        filechannel.read(readbuffer, 0, null, new completionhandler<integer, void>() {
            @override
            public void completed(integer bytesread, void attachment) {
                if (bytesread == -1) {
                    log("文件读取完成（无数据）");
                    return;
                }
                readbuffer.flip();
                string content = standardcharsets.utf_8.decode(readbuffer).tostring();
                log("异步读取结果：{}", content);
            }
            @override
            public void failed(throwable exc, void attachment) {
                log("异步读取失败：{}", exc.getmessage());
            }
        });
    }
    private static void log(string msg, object... args) {
        system.out.printf("[%s] %s%n", thread.currentthread().getname(), string.format(msg, args));
    }
}

核心特性

  • 优点：底层基于操作系统异步io（aio），无用户态线程阻塞，io效率极高；
  • 缺点：编程模型繁琐（依赖 completionhandler 回调），仅适用于io场景；
  • 适用场景：高并发文件/网络io（如文件服务器、网络通信）。

四、响应式编程：reactor（spring webflux，java 8+）

reactor 是 java 响应式编程的主流框架（实现 reactive streams 规范），基于 mono（单值异步）/flux（多值异步）实现非阻塞、背压可控的异步编程，是 spring webflux 的核心。

代码示例（spring boot 3.x + java 17+）

import lombok.extern.slf4j.slf4j;
import org.springframework.stereotype.service;
import reactor.core.publisher.mono;
import reactor.core.scheduler.schedulers;
import java.time.duration;
/**
 * reactor 响应式异步示例（spring boot 3.x + java 17+）
 */
@slf4j
@service
public class reactorasyncdemo {
    // java 17 record：商品数据
    public record product(long id, string name, double price) {}
    // 异步查询商品（mono 表示单值异步结果）
    public mono<product> queryproductasync(long productid) {
        return mono.fromsupplier(() -> {
                    log.info("查询商品{}，线程：{}", productid, thread.currentthread().getname());
                    // 模拟io耗时
                    try {
                        thread.sleep(200);
                    } catch (interruptedexception e) {
                        thread.currentthread().interrupt();
                        throw new runtimeexception("查询中断", e);
                    }
                    return new product(productid, "商品_" + productid, 99.9 + productid);
                })
                .subscribeon(schedulers.boundedelastic()) // 指定异步线程池
                .timeout(duration.ofseconds(1)) // 超时控制
                .onerrorresume(ex -> { // 异常兜底
                    log.error("查询商品失败：{}", ex.getmessage());
                    return mono.just(new product(productid, "默认商品", 0.0));
                });
    }
    // 测试入口（spring boot 环境）
    public static void main(string[] args) {
        reactorasyncdemo service = new reactorasyncdemo();
        // 非阻塞消费结果
        service.queryproductasync(1001l)
                .doonnext(product -> log.info("查询结果：{}", product)) // 消费结果
                .doonterminate(() -> log.info("异步任务结束")) // 收尾动作
                .subscribe(); // 订阅（触发异步执行）
        // 主线程等待（仅测试用）
        try {
            thread.sleep(1000);
        } catch (interruptedexception e) {
            thread.currentthread().interrupt();
        }
    }
}

核心特性

  • 优点：
    • 非阻塞+背压：可控制数据生产速率，避免消费者过载；
    • 丰富的操作符：map/flatmap/filter 等实现复杂异步逻辑；
    • 与 spring 生态深度集成：spring webflux 支持全响应式web开发；
  • 缺点：学习曲线陡峭，调试难度高；
  • 适用场景：高并发微服务、流式数据处理、实时响应系统。

五、补充：swingworker（gui 异步场景）

swingworker 是 java 专门为 swing gui 设计的异步工具，用于在后台线程执行耗时任务，避免阻塞ui线程（edt），属于小众但专用的异步方式。

核心特性

  • 优点：专为gui异步设计，内置进度回调、结果发布；
  • 缺点：仅适用于swing/awt场景，通用性差；
  • 适用场景：桌面gui应用的后台任务（如数据加载、文件导出）。

六、各异步方式对比与选型建议

方式核心优势核心劣势适用场景
executorservice+future基础通用，兼容性好阻塞获取结果，无链式调用简单异步任务，无结果联动
virtualthread轻量高效，同步写法异步java 21+ 依赖io密集型异步任务（数据库/网络）
nio.2 异步通道底层aio，io效率极高编程繁琐，仅适用于io高并发文件/网络io
reactor（webflux）非阻塞+背压，生态完善学习成本高，调试难高并发微服务、流式数据处理
swingworker专为gui设计，适配edt通用性差swing/awt桌面应用
completablefuture链式调用，任务组合线程池需手动管理通用异步任务，结果联动/组合

选型核心原则

  1. io密集型任务：优先选 虚拟线程（java 21+），次选 completablefuture；
  2. 高并发流式处理：选 reactor（webflux）；
  3. 文件/网络io：选 nio.2 异步通道；
  4. 简单异步无联动：选 executorservice+future；
  5. gui应用：选 swingworker。

所有方式均需遵循：

  • 资源管理：使用 try-with-resources 关闭线程池/io通道；
  • 异常处理：捕获中断异常并恢复线程中断状态，避免状态丢失；
  • 线程池规范：自定义线程池，避免使用默认池（如 forkjoinpool.commonpool()）。

到此这篇关于java 中多线程异步处理的文章就介绍到这了,更多相关java多线程异步内容请搜索代码网以前的文章或继续浏览下面的相关文章希望大家以后多多支持代码网！

赞 (0)

相关文章:

版权声明：本文内容由互联网用户贡献，该文观点仅代表作者本人。本站仅提供信息存储服务，不拥有所有权，不承担相关法律责任。 如发现本站有涉嫌抄袭侵权/违法违规的内容， 请发送邮件至 2386932994@qq.com 举报，一经查实将立刻删除。

发表评论

验证码：
最近更新的文章
推荐阅读
大家感兴趣的文章
Copyright © 2017-2026  代码网 保留所有权利. 粤ICP备2024248653号
站长QQ:2386932994 | 联系邮箱：2386932994@qq.com