java异步多线程精讲和多种写法

2024-10-31 07:05:49

// 1、处理mita订单轨迹数据
        List<ExpressOrderInfo> mitaExpressOrderInfo = expressOrderInfos.stream().filter(t -> Constants.MITA.equalsIgnoreCase(t.getCourierCode())).collect(Collectors.toList());
        if (CollectionUtils.isNotEmpty(mitaExpressOrderInfo)) {
            CompletableFuture.runAsync(() -> {
                try {
                    iExpressDisposeTaskService.disposeMiTaExpressOrderRouter(mitaExpressOrderInfo);
                } catch (Exception e) {
                    MessageWarningUtils.sendMessage("处理订单物流信息数据异常=" + e, dingDingConfig.getToken(), dingDingConfig.getSecretKey());
                    log.error("处理订单物流信息数据异常=", e);
                }
            });
        }

解析下代码

这段代码是用来处理特定快递公司（MITA）的订单轨迹数据的Java代码。下面是对这段代码的详细解析：

  1. 筛选MITA订单:

    List<ExpressOrderInfo> mitaExpressOrderInfo = expressOrderInfos.stream()
    .filter(t -> Constants.MITA.equalsIgnoreCase(t.getCourierCode()))
    .collect(Collectors.toList());
    • 这行代码使用Java 8的流（Stream）API从一个已有的订单信息列表expressOrderInfos中筛选出所有属于MITA快递公司的订单。
    • .filter(t -> Constants.MITA.equalsIgnoreCase(t.getCourierCode()))是一个过滤操作，它检查每个订单的快递公司代码是否等于常量Constants.MITA（不区分大小写）。
    • .collect(Collectors.toList())将过滤后的结果收集到一个新的列表mitaExpressOrderInfo中。

  2. 检查MITA订单列表是否不为空:

    if (CollectionUtils.isNotEmpty(mitaExpressOrderInfo)) {
    • 这行代码检查筛选后的MITA订单列表是否不为空，即是否有MITA的订单需要处理。

  3. 异步执行任务:

    CompletableFuture.runAsync(() -> {
    try {
        iExpressDisposeTaskService.disposeMiTaExpressOrderRouter(mitaExpressOrderInfo);
    } catch (Exception e) {
        MessageWarningUtils.sendMessage("处理订单物流信息数据异常=" + e, dingDingConfig.getToken(), dingDingConfig.getSecretKey());
        log.error("处理订单物流信息数据异常=", e);
    }
});
    • 如果有MITA的订单，代码将使用CompletableFuture的runAsync方法异步执行一个任务。
    • try { ... } catch (Exception e) { ... }是一个try-catch块，用于处理在执行异步任务时可能发生的异常。

  4. 处理MITA订单:

    iExpressDisposeTaskService.disposeMiTaExpressOrderRouter(mitaExpressOrderInfo);
    • 这是在try块中调用的一个方法，可能是处理MITA订单轨迹数据的业务逻辑。
    • iExpressDisposeTaskService可能是一个服务类的实例，disposeMiTaExpressOrderRouter是该服务的一个方法。
    • mitaExpressOrderInfo是传递给这个方法的参数，包含了所有MITA快递的订单信息。

  5. 异常处理和通知:

    MessageWarningUtils.sendMessage("处理订单物流信息数据异常=" + e, dingDingConfig.getToken(), dingDingConfig.getSecretKey());
log.error("处理订单物流信息数据异常=", e);
    • 如果在处理过程中发生异常，将捕获这些异常并执行以下操作：
      • MessageWarningUtils.sendMessage方法发送警告消息，可能是通过钉钉或其他消息平台。
      • "处理订单物流信息数据异常=" + e是消息的内容，包含了异常信息。
      • dingDingConfig.getToken()和dingDingConfig.getSecretKey()是用于发送消息的认证信息。
    • log.error是日志记录器的error级别方法，用于记录错误信息。

总结来说，这段代码异步处理MITA快递的订单轨迹数据，并在发生异常时发送警告消息和记录日志，以确保程序的健壮性和问题的可追踪性。

简短的代码里面有大内容

???? 能否详细解释一下CompletableFuture的线程池是如何工作的？

CompletableFuture是Java 8引入的一个强大的异步编程工具，它允许你以非阻塞的方式执行任务，并在任务完成后处理结果。CompletableFuture的线程池管理是通过ForkJoinPool来实现的。以下是对CompletableFuture线程池工作原理的详细解释：

1. ForkJoinPool的概念

  • ForkJoinPool是Java并发包中的一个线程池实现，专门用于处理大量小任务的并行执行。它通过“分而治之”的策略来提高任务的执行效率。
  • ForkJoinPool的核心思想是将大任务拆分成多个小任务，并将这些小任务分配给可用的线程执行。完成后，结果会被合并。

2. 默认线程池

  • 当你使用CompletableFuture.runAsync()或CompletableFuture.supplyAsync()等方法时，如果没有指定自定义的线程池，CompletableFuture会使用一个共享的ForkJoinPool.commonPool()。
  • 这个公共线程池的大小通常是可用处理器核心数的两倍，但可以通过系统属性java.util.concurrent.ForkJoinPool.common.parallelism进行调整。

3. 任务提交

  • 当你调用runAsync或supplyAsync方法时，任务会被提交到ForkJoinPool中。ForkJoinPool会将任务分配给空闲的工作线程。
  • 每个工作线程维护一个双端队列（Deque），用于存放待执行的任务。任务可以从队列的两端添加和移除，允许更高效的任务调度。

4. 任务分解与合并

  • ForkJoinPool支持任务的递归分解。大任务可以被拆分成多个小任务，这些小任务可以并行执行。
  • 当小任务完成后，结果会被合并回大任务中。这个过程是通过ForkJoinTask的fork()和join()方法实现的。

5. 线程管理

  • ForkJoinPool的线程是工作窃取（Work Stealing）模型的实现。即如果某个线程的任务队列为空，它可以从其他线程的队列中“窃取”任务来执行。这种方式可以提高CPU的利用率，减少线程的空闲时间。
  • 线程的生命周期是由ForkJoinPool管理的，线程会在没有任务可执行时被挂起，而不是一直占用系统资源。

6. 异常处理

  • 在CompletableFuture中，如果任务执行过程中抛出异常，异常会被捕获并存储在CompletableFuture对象中。你可以通过handle()、exceptionally()等方法来处理这些异常。

7. 组合与链式调用

  • CompletableFuture支持链式调用，你可以将多个异步操作组合在一起。每个操作都可以在前一个操作完成后执行，形成一个异步任务的执行链。

8. 自定义线程池

  • 如果需要更细粒度的控制，CompletableFuture也允许你指定自定义的线程池。你可以通过Executor参数来实现：
    ExecutorService executor = Executors.newFixedThreadPool(10);
CompletableFuture.runAsync(() -> {
    // 任务逻辑
}, executor);

总结

CompletableFuture通过ForkJoinPool实现了高效的异步任务处理，利用工作窃取模型和任务分解策略来提高并行处理能力。它的设计使得开发者能够以非阻塞的方式编写异步代码，同时提供了丰富的API来处理任务结果和异常。通过合理使用CompletableFuture，程序员可以显著提高应用程序的响应性和性能。

???? 能否解释一下CompletableFuture与Future的区别？

CompletableFuture和Future都是Java中用于异步编程的接口，但它们之间存在一些关键的区别：

Future

  • Future是一个代表异步计算结果的接口。它提供了检查计算是否完成的方法，以及获取计算结果的方法。Future的主要方法是get()，它会阻塞当前线程直到计算完成并返回结果。
  • Future可以取消任务、查询任务是否已经完成，以及等待任务完成。
  • Future本身不提供任何方法来完成或取消任务，这些操作需要由返回Future对象的执行器来控制。

CompletableFuture

  • CompletableFuture是Java 8中引入的，扩展了Future接口，提供了更多的方法来处理异步计算的结果。
  • 除了Future的功能外，CompletableFuture允许你附加对结果的回调，这些回调会在异步计算完成时被执行。这意味着你可以将一个任务的结果作为另一个任务的输入，实现任务的链式调用。
  • CompletableFuture提供了complete()、completeExceptionally()等方法，允许你显式地完成或异常完成一个异步操作。
  • 它支持组合多个CompletableFuture对象，例如通过thenApply()、thenAccept()、thenCombine()等方法，可以轻松地实现任务的组合和流程控制。
  • CompletableFuture还提供了orTimeout()、exceptionally()等方法，用于处理超时和异常情况。
  • CompletableFuture可以与Executor一起使用，允许你指定异步计算应该在哪个线程池中执行。

区别总结

  • 功能性：CompletableFuture提供了比Future更丰富的API，支持非阻塞获取结果、链式回调、组合任务等。
  • 灵活性：CompletableFuture允许更灵活的异步编程模型，可以轻松地处理复杂的异步逻辑。
  • 控制力：CompletableFuture允许你有更多的控制力，包括取消任务、完成任务、处理异常等。
  • 线程池：CompletableFuture可以直接与Executor配合使用，而Future则需要依赖于具体的实现来控制任务的执行。

简而言之，CompletableFuture是Future的增强版，提供了更多功能和灵活性，使得异步编程更加强大和方便。在实际开发中，CompletableFuture通常是首选，除非你有特定的理由需要使用Future。

???? 能否举例说明CompletableFuture在实际开发中的应用场景？

CompletableFuture在实际开发中有着广泛的应用，以下是一些常见的应用场景示例：

1. 异步处理数据库操作

在处理数据库查询或更新操作时，可以使用CompletableFuture来异步执行，提高应用的响应性能。

CompletableFuture.supplyAsync(() -> {
    // 模拟数据库查询操作
    return databaseQuery();
}).thenAccept(result -> {
    // 处理查询结果
    processResult(result);
}).exceptionally(throwable -> {
    // 处理异常情况
    handleException(throwable);
    return null;
});

2. 并行处理多个任务

当需要并行执行多个独立任务，并且在所有任务完成后进行一些汇总操作时，可以使用CompletableFuture。

CompletableFuture<Void> future1 = CompletableFuture.runAsync(() -> {
    // 执行任务1
    performTask1();
});

CompletableFuture<Void> future2 = CompletableFuture.runAsync(() -> {
    // 执行任务2
    performTask2();
});

CompletableFuture<Void> combinedFuture = CompletableFuture.allOf(future1, future2);
combinedFuture.thenRun(() -> {
    // 两个任务都完成后执行
    bothTasksCompleted();
});

3. 异步Web请求

在进行HTTP请求时，可以使用CompletableFuture来异步获取响应，而不阻塞主线程。

CompletableFuture<Response> future = CompletableFuture.supplyAsync(() -> {
    // 异步发送HTTP请求
    return httpClient.sendAsyncRequest();
}).thenApply(response -> {
    // 处理响应
    return processResponse(response);
});

4. 文件上传或下载

对于文件上传或下载操作，可以利用CompletableFuture来提高效率，尤其是在处理大文件或网络延迟较大时。

CompletableFuture<Void> uploadFuture = CompletableFuture.runAsync(() -> {
    // 异步上传文件
    uploadFileAsync(file);
});

uploadFuture.thenRun(() -> {
    // 上传完成后的处理
    fileUploaded();
}).exceptionally(throwable -> {
    // 处理上传异常
    handleUploadException(throwable);
    return null;
});

5. 异步处理业务逻辑

在复杂的业务逻辑处理中，可以将不同的步骤拆分成异步任务，提高处理速度。

CompletableFuture<String> step1Future = CompletableFuture.supplyAsync(() -> {
    // 第一步处理
    return step1();
});

CompletableFuture<String> step2Future = step1Future.thenApply(result1 -> {
    // 第二步处理，依赖第一步的结果
    return step2(result1);
});

step2Future.thenAccept(result2 -> {
    // 处理第二步的结果
    processResult2(result2);
});

6. 错误处理和补偿操作

在异步任务中，可以利用exceptionally来处理异常，并执行一些补偿操作。

CompletableFuture.supplyAsync(() -> {
    // 可能抛出异常的操作
    return riskyOperation();
}).exceptionally(throwable -> {
    // 异常处理
    logError(throwable);
    return defaultValue;
});

这些场景展示了CompletableFuture如何帮助开发者编写非阻塞的异步代码，提高程序的性能和响应速度，同时提供了强大的错误处理和任务组合能力。

????️ 在Java中，除了CompletableFuture，还有哪些异步编程的API可以使用？

在Java中，除了CompletableFuture，还有以下几种异步编程的API可以使用：

  1. Java原生线程（Thread）：

    • 最基础的异步执行方式，通过创建新的线程来执行任务。
    Thread thread = new Thread(() -> {
    // 任务代码
});
thread.start();

  2. ExecutorService：

    • java.util.concurrent包中的ExecutorService接口提供了一个管理线程池的API，允许你异步执行任务。
    ExecutorService executor = Executors.newFixedThreadPool(10);
executor.submit(() -> {
    // 任务代码
});

  3. Future和Callable：

    • Future接口和Callable接口与ExecutorService一起使用，允许你异步执行任务并可能获取返回值。
    Future<String> future = executor.submit(() -> {
    return "Result";
});

  4. 同步器（Synchronizers）：

    • java.util.concurrent包中的同步器，如CountDownLatch、CyclicBarrier、Semaphore和Phaser，可以用来控制并发任务的执行。

  5. Fork/Join框架：

    • java.util.concurrent.forkjoin.ForkJoinPool是一个专门用于分治任务的框架，它通过递归地将任务分割成更小的任务来实现并行处理。
    ForkJoinPool pool = new ForkJoinPool();
pool.invoke(new ForkJoinTask<Void>() {
    protected Void compute() {
        // 任务代码
        return null;
    }
});

  6. 反应式编程库（Reactive Libraries）：

    • 如Project Reactor和RxJava，它们提供了反应式编程模型，允许你以声明式的方式处理异步数据流。
    Flux.just("a", "b", "c")
    .map(this::uppercase)
    .subscribe(System.out::println);

  7. CompletableFuture的补充：

    • CompletableFuture的补充API，如CompletableFuture（用于表示一个单一的异步操作，没有结果）和BiConsumer、BiFunction等函数式接口，可以用于更复杂的异步编程场景。

  8. Web客户端（WebClient）：

    • 在Java 11及以上版本中，java.net.http.HttpClient提供了一个异步的HTTP客户端，用于发送请求和接收响应。
    HttpClient client = HttpClient.newHttpClient();
client.sendAsync(HttpRequest.newBuilder()
    .uri(URI.create("http://example.com"))
    .build(), BodyHandlers.ofString())
    .thenApply(HttpResponse::body)
    .thenAccept(System.out::println)
    .join();

  9. Quasar：

    • Quasar是一个面向服务的多语言编程框架，它提供了轻量级的代理和异步处理能力。

这些API和库提供了不同的异步编程模型和工具，开发者可以根据具体的应用场景和需求选择合适的工具。

???? 在异步编程中，如何优雅地处理异常？

在异步编程中，优雅地处理异常是非常重要的，因为异步操作通常不会直接抛出异常，而是需要通过回调、Promises/Futures等方式来处理。以下是一些处理异步异常的策略：

  1. 使用回调函数：
    在异步操作中，通常会提供一个回调函数来处理结果或异常。

    asyncOperation(param, (result) -> {
    if (result.isSuccess()) {
        // 处理结果
    } else {
        // 处理异常
        handleException(result.getError());
    }
});

  2. 使用CompletableFuture的异常处理方法：
    CompletableFuture提供了exceptionally和handle方法来处理异步操作中的异常。

    CompletableFuture<Void> future = CompletableFuture.runAsync(() -> {
    // 可能抛出异常的操作
}).exceptionally(throwable -> {
    // 处理异常
    handleException(throwable);
    return null;
});

  3. 组合异常处理：
    当有多个异步操作需要组合时，可以在每个操作中单独处理异常，或者在最终的结果处理中统一处理。

    CompletableFuture.allOf(future1, future2).thenRun(() -> {
    // 所有操作完成后的处理
}).exceptionally(throwable -> {
    // 任何一个操作抛出异常时的处理
    handleException(throwable);
    return null;
});

  4. 使用响应式编程框架：
    如RxJava或Project Reactor，它们提供了丰富的操作符来处理异常，例如onErrorResume、onErrorReturn等。

    observable
    .onErrorResume(e -> {
        // 处理异常并返回一个替代的值或操作
        return Observable.just(defaultValue);
    })
    .subscribe(value -> {
        // 正常处理
    }, error -> {
        // 处理所有异常
    });

  5. 记录日志：
    在异步操作中，记录详细的日志可以帮助你追踪异常发生的原因和上下文。

    CompletableFuture.supplyAsync(() -> {
    try {
        return riskyOperation();
    } catch (Exception e) {
        log.error("Error during async operation", e);
        throw e;
    }
});

  6. 使用try-catch块：
    在某些情况下，你可能需要在异步代码中直接使用try-catch块来捕获和处理异常。

    Thread thread = new Thread(() -> {
    try {
        // 执行可能抛出异常的操作
    } catch (Exception e) {
        // 处理异常
    }
});
thread.start();

  7. 错误传播：
    在异步操作链中，可以将异常传播到链的末端统一处理，或者在每个步骤中处理。

    CompletableFuture.supplyAsync(() -> {
    // 第一步操作
}).thenCompose(result -> {
    // 第二步操作
}).exceptionally(throwable -> {
    // 处理整个链中的异常
    return CompletableFuture.completedFuture(null);
});

  8. 超时和取消：
    在异步操作中，处理超时和取消也是异常处理的一部分。可以通过设置超时或监听取消信号来优雅地处理这些情况。

    CompletableFuture.supplyAsync(() -> {
    // 可能超时的操作
}).orTimeout(1, TimeUnit.MINUTES).thenApply(result -> {
    // 处理结果
}).exceptionally(throwable -> {
    if (throwable instanceof TimeoutException) {
        // 处理超时异常
    }
    // 处理其他异常
    return null;
});

优雅地处理异步异常需要根据具体的应用场景和需求来选择合适的策略，并且确保异常信息能够被正确记录和处理，以便进行问题排查和系统恢复。

???? 异步编程中异常处理的经典案例有哪些？

异步编程中的异常处理是确保程序健壮性的关键部分。以下是一些经典案例，展示了在不同场景下如何优雅地处理异步异常：

1. 网络请求异常处理

在进行HTTP请求时，异步处理可以提高性能，同时需要妥善处理网络异常、超时等问题。

CompletableFuture.supplyAsync(() -> httpClient.getAsync("http://api.example.com/data"))
    .thenApply(response -> response.body())
    .exceptionally(throwable -> {
        // 处理网络请求异常，可能是超时、连接失败等
        logger.error("Network request failed", throwable);
        return null; // 或者返回默认值
    });

2. 数据库操作异常处理

异步执行数据库操作时，需要处理SQL异常、连接问题等。

CompletableFuture.supplyAsync(() -> {
    Connection connection = dataSource.getConnection();
    return executeQuery(connection);
}).thenApply(result -> {
    // 处理查询结果
    return processResult(result);
}).exceptionally(throwable -> {
    // 处理数据库异常
    logger.error("Database query failed", throwable);
    return null; // 或者返回默认值
});

3. 文件I/O操作异常处理

在异步读写文件时，需要处理文件不存在、权限问题等I/O异常。

CompletableFuture.runAsync(() -> {
    try {
上一篇：solidity的struct对象，web3j java解析输出参数

下一篇：STL——list的介绍和使用-3. 总结