AI大模型编写多线程并发框架（六十二）：限流和并发度优化

系列文章目录

文章目录

前言

在这个充满技术创新的时代，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/sMmcTaskExecutor<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]Taskssubmitted. Total tasks:100[mmcTaskExample]Taskstarted. Completed tasks:14, remaining tasks:86[mmcTaskExample]Taskstarted. Completed tasks:14, remaining tasks:86[mmcTaskExample]Taskstarted. Completed tasks:20, remaining tasks:80[mmcTaskExample]Taskstarted. Completed tasks:26, remaining tasks:74[mmcTaskExample]Taskstarted. Completed tasks:33, remaining tasks:67[mmcTaskExample]Taskstarted. Completed tasks:40, remaining tasks:60[mmcTaskExample]Taskstarted. Completed tasks:46, remaining tasks:54[mmcTaskExample]Taskstarted. Completed tasks:52, remaining tasks:48[mmcTaskExample]Taskstarted. Completed tasks:58, remaining tasks:42[mmcTaskExample]Taskstarted. Completed tasks:64, remaining tasks:36[mmcTaskExample]Taskstarted. Completed tasks:70, remaining tasks:30[mmcTaskExample]Taskstarted. Completed tasks:76, remaining tasks:24[mmcTaskExample]Taskstarted. Completed tasks:82, remaining tasks:18[mmcTaskExample]Taskstarted. Completed tasks:88, remaining tasks:12[mmcTaskExample]Taskstarted. Completed tasks:94, remaining tasks:6[mmcTaskExample]Taskstarted. Completed tasks:100, remaining tasks:0[mmcTaskExample]Taskscompleted. Elapsed time:1107 ms 
result:5050

二、第三轮对话-补充异步执行代码

1、由于提交任务有可能是同步的也有可能是异步的，我们需要这个能力，所以让AI补充一下。

2、下面是AI给的回复。

• 【根据您的要求，我们将在MmcTaskExecutor类中添加commit方法以支持异步执行任务。同时，我们将创建一个回调接口MmcTaskCallback，用于在异步执行完成时获取结果。】
• 【首先，我们创建一个MmcTaskCallback接口。】

publicinterfaceMmcTaskCallback<R>{voidonComplete(R result);}

• 【接下来，我们修改MmcTaskExecutor类以添加commit方法和支持回调。】

importjava.util.List;importjava.util.concurrent.ForkJoinPool;publicclassMmcTaskExecutor<T,R>{privateList<T> taskSource;privateMmcTaskProcessor<T,R> taskProcessor;privateMmcTaskMerger<R> taskMerger;privateint threshold;privateForkJoinPool forkJoinPool;privateMmcTaskExecutor(Builder<T,R> builder){this.taskSource = builder.taskSource;this.taskProcessor = builder.taskProcessor;this.taskMerger = builder.taskMerger;this.threshold = builder.threshold;this.forkJoinPool = builder.forkJoinPool;}publicstatic<T,R>Builder<T,R>builder(){returnnewBuilder<>();}// 同步执行并返回结果publicRexecute(){MmcTask<T,R> mmcTask =newMmcTask<>(taskSource, taskProcessor, taskMerger, threshold,0, taskSource.size());return forkJoinPool.invoke(mmcTask);}// 异步执行publicvoidcommit(){commit(null);}// 异步执行并获取结果publicvoidcommit(MmcTaskCallback<R> callback){MmcTask<T,R> mmcTask =newMmcTask<>(taskSource, taskProcessor, taskMerger, threshold,0, taskSource.size());
        forkJoinPool.submit(()->{R result = mmcTask.invoke();if(callback !=null){
                callback.onComplete(result);}});}// ...省略Builder类代码}

• 【现在您可以使用commit方法异步执行任务，并通过回调函数获取结果。以下是一个示例，演示如何异步计算从1加到100的结果并在完成时输出结果。】

importjava.util.ArrayList;importjava.util.List;publicclassMmcTaskExample{publicstaticvoidmain(String[] args){List<Integer> taskSource =newArrayList<>();for(int i =1; i <=100; i++){
            taskSource.add(i);}MmcTaskProcessor<Integer,Integer> taskProcessor =newMmcTaskProcessor<Integer,Integer>(){@OverridepublicIntegerprocess(List<Integer> taskSource){int sum =0;for(Integer num : taskSource){
                    sum += num;}return sum;}};MmcTaskMerger<Integer> taskMerger =newMmcTaskMerger<Integer>(){@OverridepublicIntegermerge(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类。】

publicclassTokenBucket{privatefinallong capacity;privatefinallong tokensPerSecond;privatelong tokens;privatelong lastRefillTimestamp;publicTokenBucket(long capacity,long tokensPerSecond){this.capacity = capacity;this.tokensPerSecond = tokensPerSecond;this.tokens = capacity;this.lastRefillTimestamp =System.nanoTime();}publicsynchronizedbooleantryConsume(){refill();if(tokens >0){
            tokens--;returntrue;}returnfalse;}privatevoidrefill(){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类以支持速率控制。】

importjava.util.List;importjava.util.concurrent.ForkJoinPool;publicclassMmcTaskExecutor<T,R>{privateList<T> taskSource;privateMmcTaskProcessor<T,R> taskProcessor;privateMmcTaskMerger<R> taskMerger;privateint threshold;privateForkJoinPool forkJoinPool;privateTokenBucket tokenBucket;// 添加令牌桶成员变量privateMmcTaskExecutor(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;}publicstatic<T,R>Builder<T,R>builder(){returnnewBuilder<>();}// ...省略其他方法publicstaticclassBuilder<T,R>{privateList<T> taskSource;privateMmcTaskProcessor<T,R> taskProcessor;privateMmcTaskMerger<R> taskMerger;privateint threshold =10;privateForkJoinPool forkJoinPool =newForkJoinPool();privateTokenBucket tokenBucket;// 添加令牌桶成员变量publicBuilder<T,R>taskSource(List<T> taskSource){this.taskSource = taskSource;returnthis;}// ...省略其他方法publicBuilder<T,R>rateLimiter(long capacity,long tokensPerSecond){this.tokenBucket =newTokenBucket(capacity, tokensPerSecond);returnthis;}publicMmcTaskExecutor<T,R>build(){returnnewMmcTaskExecutor<>(this);}}}

• 【然后，我们需要修改MmcTask类以在处理任务之前尝试消耗令牌。】

importjava.util.List;importjava.util.concurrent.RecursiveTask;publicclassMmcTask<T,R>extendsRecursiveTask<R>{// ...省略其他成员变量privateTokenBucket tokenBucket;publicMmcTask(List<T> taskSource,MmcTaskProcessor<T,R> taskProcessor,MmcTaskMerger<R> taskMerger,int threshold,int start,int end,TokenBucket tokenBucket){// ...省略其他初始化代码this.tokenBucket = tokenBucket;}@OverrideprotectedRcompute(){// 在处理任务之前尝试消耗令牌while(!tokenBucket.tryConsume()){try{Thread.sleep(10);}catch(InterruptedException e){
                e.printStackTrace();}}// ...省略其他代码}}

• 【现在，您可以使用rateLimiter方法设置容器的速率限制。以下是一个示例，演示如何使用速率限制计算从1加到100的结果。】

importjava.util.ArrayList;importjava.util.List;publicclassMmcTaskExample{publicstaticvoidmain(String[] args){List<Integer> taskSource =newArrayList<>();for(int i =1; i <=100; i++){
            taskSource.add(i);}MmcTaskProcessor<Integer,Integer> taskProcessor =newMmcTaskProcessor<Integer,Integer>(){@OverridepublicIntegerprocess(List<Integer> taskSource){int sum =0;for(Integer num : taskSource){
                    sum += num;}return sum;}};MmcTaskMerger<Integer> taskMerger =newMmcTaskMerger<Integer>(){@OverridepublicIntegermerge(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接口。】

publicinterfaceRateLimiter{booleantryConsume();}

• 【接下来，我们修改TokenBucket类以实现RateLimiter接口。】

publicclassTokenBucketimplementsRateLimiter{// ...省略其他代码@OverridepublicbooleantryConsume(){// ...省略其他代码}// ...省略其他代码}

• 【然后，我们修改MmcTaskExecutor类以使用RateLimiter接口。】

importjava.util.List;importjava.util.concurrent.ForkJoinPool;publicclassMmcTaskExecutor<T,R>{// ...省略其他成员变量privateRateLimiter rateLimiter;// 使用RateLimiter接口privateMmcTaskExecutor(Builder<T,R> builder){// ...省略其他初始化代码this.rateLimiter = builder.rateLimiter;}// ...省略其他方法publicstaticclassBuilder<T,R>{// ...省略其他成员变量privateRateLimiter rateLimiter =newTokenBucket(10,1);// 默认使用TokenBucket// ...省略其他方法publicBuilder<T,R>rateLimiter(RateLimiter rateLimiter){this.rateLimiter = rateLimiter;returnthis;}// ...省略其他方法}}

这次AI大模型的回复也很完美，是不是很棒？我们继续优化。

五、第六轮对话-修改并发度

1、由于ForkJoinPool默认是线程池的数量核心数-1，但有时候我们需要自定义去定义线程数量来控制并发度，也就是任务处理速度。

2、以下是AI的回复。

• 【为了支持修改容器ForkJoinPool的默认并发度，我们可以在MmcTaskExecutor.Builder类中添加一个方法，允许用户设置并发度。修改MmcTaskExecutor类的Builder以添加设置并发度的方法：】

importjava.util.List;importjava.util.concurrent.ForkJoinPool;publicclassMmcTaskExecutor<T,R>{// ...省略其他成员变量和方法publicstaticclassBuilder<T,R>{// ...省略其他成员变量和方法publicBuilder<T,R>forkJoinPoolConcurrency(int concurrency){this.forkJoinPool =newForkJoinPool(concurrency);returnthis;}// ...省略其他方法}}

• 【现在，您可以使用forkJoinPoolConcurrency方法设置ForkJoinPool的并发度。以下是一个示例，演示如何设置并发度为4。】

importjava.util.ArrayList;importjava.util.List;publicclassMmcTaskExample{publicstaticvoidmain(String[] args){List<Integer> taskSource =newArrayList<>();for(int i =1; i <=100; i++){
            taskSource.add(i);}MmcTaskProcessor<Integer,Integer> taskProcessor =newMmcTaskProcessor<Integer,Integer>(){@OverridepublicIntegerprocess(List<Integer> taskSource){int sum =0;for(Integer num : taskSource){
                    sum += num;}return sum;}};MmcTaskMerger<Integer> taskMerger =newMmcTaskMerger<Integer>(){@OverridepublicIntegermerge(Integer leftResult,Integer rightResult){return leftResult + rightResult;}};MmcTaskExecutor<Integer,Integer> mmcTaskExecutor =MmcTaskExecutor.<Integer,Integer>builder().taskSource(taskSource).taskProcessor(taskProcessor).taskMerger(taskMerger).threshold(10).rateLimiter(newTokenBucket(10,20))// 设置速率限制.forkJoinPoolConcurrency(4)// 设置ForkJoinPool的并发度为4.build();Integer result = mmcTaskExecutor.execute();System.out.println(result);// 输出5050}}

下一篇，继续优化这个多线程框架。。。

三、参考文章

• 《AI大模型编写多线程并发框架（六十一）：从零开始搭建框架》
• 《AI大模型编写多线程并发框架（六十二）：限流和并发度优化》
• 《AI大模型编写多线程并发框架（六十三）：监听器优化·上》
• 《AI大模型编写多线程并发框架（六十四）：监听器优化·下》
• 《AI大模型编写多线程并发框架（六十五）：发布和应用》

加我加群（备注csdn）一起交流学习！更多干货下载、项目源码和大厂内推等着你