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

2024-08-27 03:44
文章标签 ai 优化 编写 模型 并发 多线程 框架 限流 六十二

本文主要是介绍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/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>() {@Overridepublic Integer process(List<Integer> taskSource) {int sum = 0;for (Integer num : taskSource) {sum += num;}return sum;}};MmcTaskMerger<Integer> taskMerger = new MmcTaskMerger<Integer>() {@Overridepublic 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;}@Overrideprotected 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>() {@Overridepublic Integer process(List<Integer> taskSource) {int sum = 0;for (Integer num : taskSource) {sum += num;}return sum;}};MmcTaskMerger<Integer> taskMerger = new MmcTaskMerger<Integer>() {@Overridepublic 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 {// ...省略其他代码@Overridepublic 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>() {@Overridepublic Integer process(List<Integer> taskSource) {int sum = 0;for (Integer num : taskSource) {sum += num;}return sum;}};MmcTaskMerger<Integer> taskMerger = new MmcTaskMerger<Integer>() {@Overridepublic 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）一起交流学习！更多干货下载、项目源码和大厂内推等着你

这篇关于AI大模型编写多线程并发框架（六十二）：限流和并发度优化的文章就介绍到这儿，希望我们推荐的文章对编程师们有所帮助！


http://www.chinasem.cn/article/1110534。

相关文章

C#使用HttpClient进行Post请求出现超时问题的解决及优化

C#使用HttpClient进行Post请求出现超时问题的解决及优化

《C#使用HttpClient进行Post请求出现超时问题的解决及优化》最近我的控制台程序发现有时候总是出现请求超时等问题,通常好几分钟最多只有3-4个请求,在使用apipost发现并发10个5分钟也... 目录优化结论单例HttpClient连接池耗尽和并发并发异步最终优化后优化结论我直接上优化结论吧，
阅读更多...
Java内存泄漏问题的排查、优化与最佳实践

Java内存泄漏问题的排查、优化与最佳实践

《Java内存泄漏问题的排查、优化与最佳实践》在Java开发中,内存泄漏是一个常见且令人头疼的问题,内存泄漏指的是程序在运行过程中,已经不再使用的对象没有被及时释放,从而导致内存占用不断增加,最终... 目录引言1. 什么是内存泄漏？常见的内存泄漏情况2. 如何排查 Java 中的内存泄漏？2.1 使用 J
阅读更多...
Golang的CSP模型简介(最新推荐)

Golang的CSP模型简介(最新推荐)

《Golang的CSP模型简介(最新推荐)》Golang采用了CSP（CommunicatingSequentialProcesses,通信顺序进程）并发模型,通过goroutine和channe... 目录前言一、介绍1. 什么是 CSP 模型2. Goroutine3. Channel4. Channe
阅读更多...
利用Python编写一个简单的聊天机器人

利用Python编写一个简单的聊天机器人

《利用Python编写一个简单的聊天机器人》这篇文章主要为大家详细介绍了如何利用Python编写一个简单的聊天机器人,文中的示例代码讲解详细,感兴趣的小伙伴可以跟随小编一起学习一下... 使用 python 编写一个简单的聊天机器人可以从最基础的逻辑开始，然后逐步加入更复杂的功能。这里我们将先实现一个简单的
阅读更多...
Ubuntu系统怎么安装Warp? 新一代AI 终端神器安装使用方法

Ubuntu系统怎么安装Warp? 新一代AI 终端神器安装使用方法

《Ubuntu系统怎么安装Warp?新一代AI终端神器安装使用方法》Warp是一款使用Rust开发的现代化AI终端工具，该怎么再Ubuntu系统中安装使用呢？下面我们就来看看详细教程... Warp Terminal 是一款使用 Rust 开发的现代化「AI 终端」工具。最初它只支持 MACOS，但在 20
阅读更多...
MySQL不使用子查询的原因及优化案例

MySQL不使用子查询的原因及优化案例

《MySQL不使用子查询的原因及优化案例》对于mysql,不推荐使用子查询,效率太差,执行子查询时,MYSQL需要创建临时表,查询完毕后再删除这些临时表,所以,子查询的速度会受到一定的影响,本文给大家... 目录不推荐使用子查询和JOIN的原因解决方案优化案例案例1：查询所有有库存的商品信息案例2：使用EX
阅读更多...
使用PyQt5编写一个简单的取色器

使用PyQt5编写一个简单的取色器

《使用PyQt5编写一个简单的取色器》：本文主要介绍PyQt5搭建的一个取色器,一共写了两款应用,一款使用快捷键捕获鼠标附近图像的RGB和16进制颜色编码,一款跟随鼠标刷新图像的RGB和16... 目录取色器1取色器2PyQt5搭建的一个取色器，一共写了两款应用，一款使用快捷键捕获鼠标附近图像的RGB和16
阅读更多...
MySQL中my.ini文件的基础配置和优化配置方式

MySQL中my.ini文件的基础配置和优化配置方式

《MySQL中my.ini文件的基础配置和优化配置方式》文章讨论了数据库异步同步的优化思路,包括三个主要方面：幂等性、时序和延迟,作者还分享了MySQL配置文件的优化经验,并鼓励读者提供支持... 目录mysql my.ini文件的配置和优化配置优化思路MySQL配置文件优化总结MySQL my.ini文件
阅读更多...
Python基于火山引擎豆包大模型搭建QQ机器人详细教程(2024年最新)

Python基于火山引擎豆包大模型搭建QQ机器人详细教程(2024年最新)

《Python基于火山引擎豆包大模型搭建QQ机器人详细教程(2024年最新)》：本文主要介绍Python基于火山引擎豆包大模型搭建QQ机器人详细的相关资料,包括开通模型、配置APIKEY鉴权和SD... 目录豆包大模型概述开通模型付费安装 SDK 环境配置 API KEY 鉴权Ark 模型接口Prompt
阅读更多...
正则表达式高级应用与性能优化记录

正则表达式高级应用与性能优化记录

《正则表达式高级应用与性能优化记录》本文介绍了正则表达式的高级应用和性能优化技巧,包括文本拆分、合并、XML/HTML解析、数据分析、以及性能优化方法,通过这些技巧,可以更高效地利用正则表达式进行复杂... 目录第6章：正则表达式的高级应用6.1 模式匹配与文本处理6.1.1 文本拆分6.1.2 文本合并6
阅读更多...

Copyright China编程 23002807@qq.com

沪ICP备2023033072号-2