系列文章目录
文章目录
- 系列文章目录
- 前言
- 一、项目背景
- 二、第三轮对话-补充异步执行代码
- 三、第四轮对话-增加限流器
- 四、第五轮对话-抽取限流器接口
- 五、第六轮对话-修改并发度
- 三、参考文章
前言
在这个充满技术创新的时代，AI大模型正成为开发者们的新宠。它们可以帮助我们完成从简单的问答到复杂的编程任务，所以AI编程将会是未来的主流方向，利用AI大模型的能力，本文将介绍从零到一用AI大模型编写一个多线程并发框架。
一、项目背景
由于博主每次批量处理数据任务时，都需要去考虑线程池各种参数，例如考虑任务执行速率、任务监控、任务回调、异常处理等。由于每个需求的实际诉求不一样，所以数据处理任务的代替代码都一样，只有部分有差异，从而导致代码仓库里各种复制粘贴，
这不是博主想要的，于是，萌发了一个封装一个多线程组件框架的想法，只需要传入自定义任务等参数，其它东西都不需要关心。
使用示例如下：源码地址
1、引入依赖。
<dependency>
<groupId>io.github.vipjoey</groupId>
<artifactId>mmc-juc</artifactId>
<version>1.0</version>
</dependency>
2、使用示例。
// 创建一个MmcTaskExecutor实例，用于执行单次长耗时任务
// 下面是创建一个计算从1加到100的任务，总共100个任务，采用fork分治算法，阈值为10，总共任务为100 / 10 * 2 = 20个大任务，执行速率约为10/s
MmcTaskExecutor<Integer, Integer> mmcTaskExecutor = MmcTaskExecutor.<Integer, Integer>builder()
.taskSource(IntStream.rangeClosed(1, 100).boxed().collect(Collectors.toList())) // 设置任务源
.taskProcessor(x -> x.stream().reduce(0, Integer::sum)) // 设置任务处理方法
.taskMerger(Integer::sum) // 设置结果处理方法（可选）
.threshold(10) // 设置任务处理阈值（可选）
.taskName("mmcTaskExample") // 设置任务名称
.rateLimiter(10, 20) // 设置速率限制，容量为10，每秒产生令牌为20，休眠时间为10ms
.forkJoinPoolConcurrency(4) // 设置ForkJoinPool的并发度为4
.build();
// 同步执行并打印结果
System.out.println("result: " + mmcTaskExecutor.execute());
// 任务执行过程监控
[mmcTaskExample] Tasks submitted. Total tasks: 100
[mmcTaskExample] Task started. Completed tasks: 14, remaining tasks: 86
[mmcTaskExample] Task started. Completed tasks: 14, remaining tasks: 86
[mmcTaskExample] Task started. Completed tasks: 20, remaining tasks: 80
[mmcTaskExample] Task started. Completed tasks: 26, remaining tasks: 74
[mmcTaskExample] Task started. Completed tasks: 33, remaining tasks: 67
[mmcTaskExample] Task started. Completed tasks: 40, remaining tasks: 60
[mmcTaskExample] Task started. Completed tasks: 46, remaining tasks: 54
[mmcTaskExample] Task started. Completed tasks: 52, remaining tasks: 48
[mmcTaskExample] Task started. Completed tasks: 58, remaining tasks: 42
[mmcTaskExample] Task started. Completed tasks: 64, remaining tasks: 36
[mmcTaskExample] Task started. Completed tasks: 70, remaining tasks: 30
[mmcTaskExample] Task started. Completed tasks: 76, remaining tasks: 24
[mmcTaskExample] Task started. Completed tasks: 82, remaining tasks: 18
[mmcTaskExample] Task started. Completed tasks: 88, remaining tasks: 12
[mmcTaskExample] Task started. Completed tasks: 94, remaining tasks: 6
[mmcTaskExample] Task started. Completed tasks: 100, remaining tasks: 0
[mmcTaskExample] Tasks completed. Elapsed time: 1107 ms
result: 5050
二、第三轮对话-补充异步执行代码
1、由于提交任务有可能是同步的也有可能是异步的，我们需要这个能力，所以让AI补充一下。
2、下面是AI给的回复。
-
【根据您的要求，我们将在MmcTaskExecutor类中添加commit方法以支持异步执行任务。同时，我们将创建一个回调接口MmcTaskCallback，用于在异步执行完成时获取结果。】
-
【首先，我们创建一个MmcTaskCallback接口。】
public interface MmcTaskCallback<R> {
void onComplete(R result);
}
- 【接下来，我们修改MmcTaskExecutor类以添加commit方法和支持回调。】
import java.util.List;
import java.util.concurrent.ForkJoinPool;
public class MmcTaskExecutor<T, R> {
private List<T> taskSource;
private MmcTaskProcessor<T, R> taskProcessor;
private MmcTaskMerger<R> taskMerger;
private int threshold;
private ForkJoinPool forkJoinPool;
private MmcTaskExecutor(Builder<T, R> builder) {
this.taskSource = builder.taskSource;
this.taskProcessor = builder.taskProcessor;
this.taskMerger = builder.taskMerger;
this.threshold = builder.threshold;
this.forkJoinPool = builder.forkJoinPool;
}
public static <T, R> Builder<T, R> builder() {
return new Builder<>();
}
// 同步执行并返回结果
public R execute() {
MmcTask<T, R> mmcTask = new MmcTask<>(taskSource, taskProcessor, taskMerger, threshold, 0, taskSource.size());
return forkJoinPool.invoke(mmcTask);
}
// 异步执行
public void commit() {
commit(null);
}
// 异步执行并获取结果
public void commit(MmcTaskCallback<R> callback) {
MmcTask<T, R> mmcTask = new MmcTask<>(taskSource, taskProcessor, taskMerger, threshold, 0, taskSource.size());
forkJoinPool.submit(() -> {
R result = mmcTask.invoke();
if (callback != null) {
callback.onComplete(result);
}
});
}
// ...省略Builder类代码
}
- 【现在您可以使用commit方法异步执行任务，并通过回调函数获取结果。以下是一个示例，演示如何异步计算从1加到100的结果并在完成时输出结果。】
import java.util.ArrayList;
import java.util.List;
public class MmcTaskExample {
public static void main(String[] args) {
List<Integer> taskSource = new ArrayList<>();
for (int i = 1; i <= 100; i++) {
taskSource.add(i);
}
MmcTaskProcessor<Integer, Integer> taskProcessor = new MmcTaskProcessor<Integer, Integer>() {
@Override
public Integer process(List<Integer> taskSource) {
int sum = 0;
for (Integer num : taskSource) {
sum += num;
}
return sum;
}
};
MmcTaskMerger<Integer> taskMerger = new MmcTaskMerger<Integer>() {
@Override
public Integer merge(Integer leftResult, Integer rightResult) {
return leftResult + rightResult;
}
};
MmcTaskExecutor<Integer, Integer> mmcTaskExecutor = MmcTaskExecutor.<Integer, Integer>builder()
.taskSource(taskSource)
.taskProcessor(taskProcessor)
.taskMerger(taskMerger)
.threshold(10)
.build();
mmcTaskExecutor.commit(result -> System.out.println("异步执行结果：" + result)); // 输出5050
// 等待异步任务完成，防止主线程提前退出
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
这次诉求改动不大，AI回答也很完美，我们继续优化！
三、第四轮对话-增加限流器
1、因为有些任务不能执行过快，例如刷DB数据或者远程接口调用等，需要补充限流功能。
2、下面是AI大模型给的回复。
- 【为了支持速率控制，我们将使用令牌桶算法实现速率限制。首先，我们需要创建一个令牌桶类TokenBucket，然后在MmcTaskExecutor类中使用它来控制任务的执行速度。首先，我们创建一个TokenBucket类。】
public class TokenBucket {
private final long capacity;
private final long tokensPerSecond;
private long tokens;
private long lastRefillTimestamp;
public TokenBucket(long capacity, long tokensPerSecond) {
this.capacity = capacity;
this.tokensPerSecond = tokensPerSecond;
this.tokens = capacity;
this.lastRefillTimestamp = System.nanoTime();
}
public synchronized boolean tryConsume() {
refill();
if (tokens > 0) {
tokens--;
return true;
}
return false;
}
private void refill() {
long now = System.nanoTime();
long elapsedTime = now - lastRefillTimestamp;
long tokensToAdd = (elapsedTime * tokensPerSecond) / 1_000_000_000;
if (tokensToAdd > 0) {
tokens = Math.min(capacity, tokens + tokensToAdd);
lastRefillTimestamp = now;
}
}
}
- 【接下来，我们修改MmcTaskExecutor类以支持速率控制。】
import java.util.List;
import java.util.concurrent.ForkJoinPool;
public class MmcTaskExecutor<T, R> {
private List<T> taskSource;
private MmcTaskProcessor<T, R> taskProcessor;
private MmcTaskMerger<R> taskMerger;
private int threshold;
private ForkJoinPool forkJoinPool;
private TokenBucket tokenBucket; // 添加令牌桶成员变量
private MmcTaskExecutor(Builder<T, R> builder) {
this.taskSource = builder.taskSource;
this.taskProcessor = builder.taskProcessor;
this.taskMerger = builder.taskMerger;
this.threshold = builder.threshold;
this.forkJoinPool = builder.forkJoinPool;
this.tokenBucket = builder.tokenBucket;
}
public static <T, R> Builder<T, R> builder() {
return new Builder<>();
}
// ...省略其他方法
public static class Builder<T, R> {
private List<T> taskSource;
private MmcTaskProcessor<T, R> taskProcessor;
private MmcTaskMerger<R> taskMerger;
private int threshold = 10;
private ForkJoinPool forkJoinPool = new ForkJoinPool();
private TokenBucket tokenBucket; // 添加令牌桶成员变量
public Builder<T, R> taskSource(List<T> taskSource) {
this.taskSource = taskSource;
return this;
}
// ...省略其他方法
public Builder<T, R> rateLimiter(long capacity, long tokensPerSecond) {
this.tokenBucket = new TokenBucket(capacity, tokensPerSecond);
return this;
}
public MmcTaskExecutor<T, R> build() {
return new MmcTaskExecutor<>(this);
}
}
}
- 【然后，我们需要修改MmcTask类以在处理任务之前尝试消耗令牌。】
import java.util.List;
import java.util.concurrent.RecursiveTask;
public class MmcTask<T, R> extends RecursiveTask<R> {
// ...省略其他成员变量
private TokenBucket tokenBucket;
public MmcTask(List<T> taskSource, MmcTaskProcessor<T, R> taskProcessor, MmcTaskMerger<R> taskMerger, int threshold, int start, int end, TokenBucket tokenBucket) {
// ...省略其他初始化代码
this.tokenBucket = tokenBucket;
}
@Override
protected R compute() {
// 在处理任务之前尝试消耗令牌
while (!tokenBucket.tryConsume()) {
try {
Thread.sleep(10);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
// ...省略其他代码
}
}
- 【现在，您可以使用rateLimiter方法设置容器的速率限制。以下是一个示例，演示如何使用速率限制计算从1加到100的结果。】
import java.util.ArrayList;
import java.util.List;
public class MmcTaskExample {
public static void main(String[] args) {
List<Integer> taskSource = new ArrayList<>();
for (int i = 1; i <= 100; i++) {
taskSource.add(i);
}
MmcTaskProcessor<Integer, Integer> taskProcessor = new MmcTaskProcessor<Integer, Integer>() {
@Override
public Integer process(List<Integer> taskSource) {
int sum = 0;
for (Integer num : taskSource) {
sum += num;
}
return sum;
}
};
MmcTaskMerger<Integer> taskMerger = new MmcTaskMerger<Integer>() {
@Override
public Integer merge(Integer leftResult, Integer rightResult) {
return leftResult + rightResult;
}
};
MmcTaskExecutor<Integer, Integer> mmcTaskExecutor = MmcTaskExecutor.<Integer, Integer>builder()
.taskSource(taskSource)
.taskProcessor(taskProcessor)
.taskMerger(taskMerger)
.threshold(10)
.rateLimiter(10, 1) // 设置速率限制
.build();
Integer result = mmcTaskExecutor.execute();
System.out.println(result); // 输出5050
}
}
- 【在这个示例中，我们使用rateLimiter方法设置了速率限制，每秒只能处理1个任务。然后我们使用execute方法执行任务并输出结果。因为我们设置了速率限制，所以执行任务的速度会受到限制。】
四、第五轮对话-抽取限流器接口
1、从AI大模型给出的代码来看，基本符合我们的诉求，不过也存在一下问题：
我希望任务限流器可以使用第三方例如Guava RateLimiter这些工具去限流，在工具里写死是不太合适的，我们继续让它修改。
2、下面是AI的回复。
- 【为了支持第三方速率限制器，我们需要创建一个速率限制器接口RateLimiter，并将TokenBucket类实现此接口。然后，我们在MmcTaskExecutor类中使用RateLimiter接口，而不是直接使用TokenBucket类。首先，我们创建一个RateLimiter接口。】
public interface RateLimiter {
boolean tryConsume();
}
- 【接下来，我们修改TokenBucket类以实现RateLimiter接口。】
public class TokenBucket implements RateLimiter {
// ...省略其他代码
@Override
public boolean tryConsume() {
// ...省略其他代码
}
// ...省略其他代码
}
- 【然后，我们修改MmcTaskExecutor类以使用RateLimiter接口。】
import java.util.List;
import java.util.concurrent.ForkJoinPool;
public class MmcTaskExecutor<T, R> {
// ...省略其他成员变量
private RateLimiter rateLimiter; // 使用RateLimiter接口
private MmcTaskExecutor(Builder<T, R> builder) {
// ...省略其他初始化代码
this.rateLimiter = builder.rateLimiter;
}
// ...省略其他方法
public static class Builder<T, R> {
// ...省略其他成员变量
private RateLimiter rateLimiter = new TokenBucket(10, 1); // 默认使用TokenBucket
// ...省略其他方法
public Builder<T, R> rateLimiter(RateLimiter rateLimiter) {
this.rateLimiter = rateLimiter;
return this;
}
// ...省略其他方法
}
}
这次AI大模型的回复也很完美，是不是很棒？我们继续优化。
五、第六轮对话-修改并发度
1、由于ForkJoinPool默认是线程池的数量核心数-1，但有时候我们需要自定义去定义线程数量来控制并发度，也就是任务处理速度。
2、以下是AI的回复。
- 【为了支持修改容器ForkJoinPool的默认并发度，我们可以在MmcTaskExecutor.Builder类中添加一个方法，允许用户设置并发度。修改MmcTaskExecutor类的Builder以添加设置并发度的方法：】
import java.util.List;
import java.util.concurrent.ForkJoinPool;
public class MmcTaskExecutor<T, R> {
// ...省略其他成员变量和方法
public static class Builder<T, R> {
// ...省略其他成员变量和方法
public Builder<T, R> forkJoinPoolConcurrency(int concurrency) {
this.forkJoinPool = new ForkJoinPool(concurrency);
return this;
}
// ...省略其他方法
}
}
- 【现在，您可以使用forkJoinPoolConcurrency方法设置ForkJoinPool的并发度。以下是一个示例，演示如何设置并发度为4。】
import java.util.ArrayList;
import java.util.List;
public class MmcTaskExample {
public static void main(String[] args) {
List<Integer> taskSource = new ArrayList<>();
for (int i = 1; i <= 100; i++) {
taskSource.add(i);
}
MmcTaskProcessor<Integer, Integer> taskProcessor = new MmcTaskProcessor<Integer, Integer>() {
@Override
public Integer process(List<Integer> taskSource) {
int sum = 0;
for (Integer num : taskSource) {
sum += num;
}
return sum;
}
};
MmcTaskMerger<Integer> taskMerger = new MmcTaskMerger<Integer>() {
@Override
public Integer merge(Integer leftResult, Integer rightResult) {
return leftResult + rightResult;
}
};
MmcTaskExecutor<Integer, Integer> mmcTaskExecutor = MmcTaskExecutor.<Integer, Integer>builder()
.taskSource(taskSource)
.taskProcessor(taskProcessor)
.taskMerger(taskMerger)
.threshold(10)
.rateLimiter(new TokenBucket(10,20)) // 设置速率限制
.forkJoinPoolConcurrency(4) // 设置ForkJoinPool的并发度为4
.build();
Integer result = mmcTaskExecutor.execute();
System.out.println(result); // 输出5050
}
}
下一篇，继续优化这个多线程框架。。。
三、参考文章
- 《AI大模型编写多线程并发框架（六十一）：从零开始搭建框架》
- 《AI大模型编写多线程并发框架（六十二）：限流和并发度优化》
- 《AI大模型编写多线程并发框架（六十三）：监听器优化》
- 《AI大模型编写多线程并发框架（六十四）：发布和应用》
加我加群（备注csdn）一起交流学习！更多干货下载、项目源码和大厂内推等着你
 |  |
