AI大模型编写多线程并发框架（六十一）：从零开始搭建框架

2024-08-27 03:20
文章标签 ai 编写 模型 并发 多线程 搭建 框架 从零开始 六十一

本文主要是介绍AI大模型编写多线程并发框架（六十一）：从零开始搭建框架，希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！

系列文章目录

文章目录

  • 系列文章目录
  • 前言
    • 一、项目背景
    • 二、第一轮对话-让AI大模型理解我们的诉求
    • 二、第二轮对话-优化任务处理方法和结果处理方法
    • 三、参考文章

前言

在这个充满技术创新的时代，AI大模型正成为开发者们的新宠。它们可以帮助我们完成从简单的问答到复杂的编程任务，所以AI编程将会是未来的主流方向，利用AI大模型的能力，本文将介绍从零到一用AI大模型编写一个多线程并发框架。

一、项目背景

由于博主每次批量处理数据任务时，都需要去考虑线程池各种参数，例如考虑任务执行速率、任务监控、任务回调、异常处理等。由于每个需求的实际诉求不一样，所以数据处理任务的代替代码都一样，只有部分有差异，从而导致代码仓库里各种复制粘贴，
这不是博主想要的，于是，萌发了一个封装一个多线程组件框架的想法，只需要传入自定义任务等参数，其它东西都不需要关心。

使用示例如下：源码地址


// 创建一个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

二、第一轮对话-让AI大模型理解我们的诉求

1、首先告诉AI大模型我的想法和目的，让它充当编程专家，给出初始化代码。
在这里插入图片描述

2、下面是AI给的回复。

  • 【首先，我们需要创建一个核心类MmcTask，它将实现RecursiveTask接口，并实现compute方法，用于执行真正的任务处理方法。】
import java.util.List;
import java.util.concurrent.RecursiveTask;public class MmcTask<T, R> extends RecursiveTask<R> {private List<T> taskSource;private MmcTaskHandler<T, R> taskHandler;private int threshold;private int start;private int end;public MmcTask(List<T> taskSource, MmcTaskHandler<T, R> taskHandler, int threshold, int start, int end) {this.taskSource = taskSource;this.taskHandler = taskHandler;this.threshold = threshold;this.start = start;this.end = end;}@Overrideprotected R compute() {// 当任务数量小于等于阈值时，执行真正的任务处理方法if (end - start <= threshold) {return taskHandler.handle(taskSource.subList(start, end));}// 将任务拆分为两部分，并递归执行int middle = (start + end) / 2;MmcTask<T, R> leftTask = new MmcTask<>(taskSource, taskHandler, threshold, start, middle);MmcTask<T, R> rightTask = new MmcTask<>(taskSource, taskHandler, threshold, middle, end);leftTask.fork();R rightResult = rightTask.compute();R leftResult = leftTask.join();// 合并两个任务的结果return taskHandler.merge(leftResult, rightResult);}
}
  • 【接下来，我们需要创建一个MmcTaskHandler接口，用于处理任务并合并任务结果。】
public interface MmcTaskHandler<T, R> {R handle(List<T> taskSource);R merge(R leftResult, R rightResult);
}
  • 【然后，我们创建一个MmcTaskExecutor类，用于构建任务容器并执行任务。】
import java.util.List;
import java.util.concurrent.ForkJoinPool;public class MmcTaskExecutor<T, R> {private List<T> taskSource;private MmcTaskHandler<T, R> taskHandler;private int threshold;private ForkJoinPool forkJoinPool;private MmcTaskExecutor(Builder<T, R> builder) {this.taskSource = builder.taskSource;this.taskHandler = builder.taskHandler;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, taskHandler, threshold, 0, taskSource.size());return forkJoinPool.invoke(mmcTask);}public static class Builder<T, R> {private List<T> taskSource;private MmcTaskHandler<T, R> taskHandler;private int threshold = 10;private ForkJoinPool forkJoinPool = new ForkJoinPool();public Builder<T, R> taskSource(List<T> taskSource) {this.taskSource = taskSource;return this;}public Builder<T, R> taskHandler(MmcTaskHandler<T, R> taskHandler) {this.taskHandler = taskHandler;return this;}public Builder<T, R> threshold(int threshold) {this.threshold = threshold;return this;}public Builder<T, R> forkJoinPool(ForkJoinPool forkJoinPool) {this.forkJoinPool = forkJoinPool;return this;}public MmcTaskExecutor<T, R> build() {return new MmcTaskExecutor<>(this);}}
}
  • 【最后，我们编写一个示例，计算从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);}MmcTaskHandler<Integer, Integer> taskHandler = new MmcTaskHandler<Integer, Integer>() {@Overridepublic Integer handle(List<Integer> taskSource) {int sum = 0;for (Integer num : taskSource) {sum += num;}return sum;}@Overridepublic Integer merge(Integer leftResult, Integer rightResult) {return leftResult + rightResult;}};MmcTaskExecutor<Integer, Integer> mmcTaskExecutor = MmcTaskExecutor.<Integer, Integer>builder().taskSource(taskSource).taskHandler(taskHandler).threshold(10).build();Integer result = mmcTaskExecutor.execute();System.out.println(result);  // 输出5050}
}

从AI大模型给出的代码来看，它基本能理解我们的意图，不过也存在一下问题：

  • 我希望任务处理方法和结果处理方法可以分开传入，而AI给出的MmcTaskHandler是合并一起的；
  • 我要求代码需要有注释，但AI实际上并没有给太多注释；

不过问题不大，我们可以通过多轮对话去优化。

二、第二轮对话-优化任务处理方法和结果处理方法

1、MmcTaskHandler应该要拆分，也就是MmcTaskExecutor.builder要支持传入真正处理任务的方法，也要支持传入merge结果的处理方法，因为有些任务可能是没有计算结果的，所以merge结果处理方法可以为空。在这里插入图片描述

2、下面是AI大模型给的回复。

  • 【根据您的要求，我们可以将MmcTaskHandler拆分为两个接口：MmcTaskProcessor用于处理任务，MmcTaskMerger用于合并任务结果。同时，我们修改MmcTaskExecutor.Builder类以支持传入这两个接口。首先，我们创建一个MmcTaskProcessor接口。】
public interface MmcTaskProcessor<T, R> {R process(List<T> taskSource);
}
  • 【接下来，我们创建一个MmcTaskMerger接口。】
public interface MmcTaskMerger<R> {R merge(R leftResult, R rightResult);
}
  • 【然后，我们修改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 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 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();public Builder<T, R> taskSource(List<T> taskSource) {this.taskSource = taskSource;return this;}public Builder<T, R> taskProcessor(MmcTaskProcessor<T, R> taskProcessor) {this.taskProcessor = taskProcessor;return this;}public Builder<T, R> taskMerger(MmcTaskMerger<R> taskMerger) {this.taskMerger = taskMerger;return this;}public Builder<T, R> threshold(int threshold) {this.threshold = threshold;return this;}public Builder<T, R> forkJoinPool(ForkJoinPool forkJoinPool) {this.forkJoinPool = forkJoinPool;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 List<T> taskSource;private MmcTaskProcessor<T, R> taskProcessor;private MmcTaskMerger<R> taskMerger;private int threshold;private int start;private int end;public MmcTask(List<T> taskSource, MmcTaskProcessor<T, R> taskProcessor, MmcTaskMerger<R> taskMerger, int threshold, int start, int end) {this.taskSource = taskSource;this.taskProcessor = taskProcessor;this.taskMerger = taskMerger;this.threshold = threshold;this.start = start;this.end = end;}@Overrideprotected R compute() {if (end - start <= threshold) {return taskProcessor.process(taskSource.subList(start, end));}int middle = (start + end) / 2;MmcTask<T, R> leftTask = new MmcTask<>(taskSource, taskProcessor, taskMerger, threshold, start, middle);MmcTask<T, R> rightTask = new MmcTask<>(taskSource, taskProcessor, taskMerger, threshold, middle, end);leftTask.fork();R rightResult = rightTask.compute();R leftResult = leftTask.join();if (taskMerger != null) {return taskMerger.merge(leftResult, rightResult);} else {return null;}}
}
  • 【计算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();Integer result = mmcTaskExecutor.execute();System.out.println(result);  // 输出5050}
}

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

三、参考文章

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

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

这篇关于AI大模型编写多线程并发框架（六十一）：从零开始搭建框架的文章就介绍到这儿，希望我们推荐的文章对编程师们有所帮助！


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

相关文章

Ilya-AI分享的他在OpenAI学习到的15个提示工程技巧

Ilya-AI分享的他在OpenAI学习到的15个提示工程技巧

Ilya（不是本人，claude AI）在社交媒体上分享了他在OpenAI学习到的15个Prompt撰写技巧。 以下是详细的内容： 提示精确化:在编写提示时,力求表达清晰准确。清楚地阐述任务需求和概念定义至关重要。例:不用"分析文本",而用"判断这段话的情感倾向:积极、消极还是中性"。 快速迭代:善于快速连续调整提示。熟练的提示工程师能够灵活地进行多轮优化。例:从"总结文章"到"用
阅读更多...
AI绘图怎么变现？想做点副业的小白必看！

AI绘图怎么变现？想做点副业的小白必看！

在科技飞速发展的今天，AI绘图作为一种新兴技术，不仅改变了艺术创作的方式，也为创作者提供了多种变现途径。本文将详细探讨几种常见的AI绘图变现方式，帮助创作者更好地利用这一技术实现经济收益。 更多实操教程和AI绘画工具，可以扫描下方，免费获取 定制服务：个性化的创意商机 个性化定制 AI绘图技术能够根据用户需求生成个性化的头像、壁纸、插画等作品。例如，姓氏头像在电商平台上非常受欢迎，
阅读更多...
大模型研发全揭秘：客服工单数据标注的完整攻略

大模型研发全揭秘：客服工单数据标注的完整攻略

在人工智能（AI）领域，数据标注是模型训练过程中至关重要的一步。无论你是新手还是有经验的从业者，掌握数据标注的技术细节和常见问题的解决方案都能为你的AI项目增添不少价值。在电信运营商的客服系统中，工单数据是客户问题和解决方案的重要记录。通过对这些工单数据进行有效标注，不仅能够帮助提升客服自动化系统的智能化水平，还能优化客户服务流程，提高客户满意度。本文将详细介绍如何在电信运营商客服工单的背景下进行
阅读更多...
从去中心化到智能化：Web3如何与AI共同塑造数字生态

从去中心化到智能化：Web3如何与AI共同塑造数字生态

在数字时代的演进中，Web3和人工智能（AI）正成为塑造未来互联网的两大核心力量。Web3的去中心化理念与AI的智能化技术，正相互交织，共同推动数字生态的变革。本文将探讨Web3与AI的融合如何改变数字世界，并展望这一新兴组合如何重塑我们的在线体验。 Web3的去中心化愿景 Web3代表了互联网的第三代发展，它基于去中心化的区块链技术，旨在创建一个开放、透明且用户主导的数字生态。不同于传统
阅读更多...
AI一键生成 PPT

AI一键生成 PPT

AI一键生成 PPT 操作步骤 作为一名打工人，是不是经常需要制作各种PPT来分享我的生活和想法。但是，你们知道，有时候灵感来了，时间却不够用了！😩直到我发现了Kimi AI——一个能够自动生成PPT的神奇助手！🌟 什么是Kimi？ 一款月之暗面科技有限公司开发的AI办公工具，帮助用户快速生成高质量的演示文稿。 无论你是职场人士、学生还是教师，Kimi都能够为你的办公文
阅读更多...
Andrej Karpathy最新采访：认知核心模型10亿参数就够了，AI会打破教育不公的僵局

Andrej Karpathy最新采访：认知核心模型10亿参数就够了，AI会打破教育不公的僵局

夕小瑶科技说 原创  作者 | 海野 AI圈子的红人，AI大神Andrej Karpathy，曾是OpenAI联合创始人之一，特斯拉AI总监。上一次的动态是官宣创办一家名为 Eureka Labs 的人工智能+教育公司 ，宣布将长期致力于AI原生教育。 近日，Andrej Karpathy接受了No Priors（投资博客）的采访，与硅谷知名投资人 Sara Guo 和 Elad G
阅读更多...
Retrieval-based-Voice-Conversion-WebUI模型构建指南

Retrieval-based-Voice-Conversion-WebUI模型构建指南

一、模型介绍 Retrieval-based-Voice-Conversion-WebUI（简称 RVC）模型是一个基于 VITS（Variational Inference with adversarial learning for end-to-end Text-to-Speech）的简单易用的语音转换框架。 具有以下特点 简单易用：RVC 模型通过简单易用的网页界面，使得用户无需深入了
阅读更多...
搭建Kafka+zookeeper集群调度

搭建Kafka+zookeeper集群调度

前言 硬件环境 172.18.0.5        kafkazk1        Kafka+zookeeper                Kafka Broker集群 172.18.0.6        kafkazk2        Kafka+zookeeper                Kafka Broker集群 172.18.0.7        kafkazk3
阅读更多...
透彻！驯服大型语言模型（LLMs）的五种方法，及具体方法选择思路

透彻！驯服大型语言模型（LLMs）的五种方法，及具体方法选择思路

引言 随着时间的发展，大型语言模型不再停留在演示阶段而是逐步面向生产系统的应用，随着人们期望的不断增加，目标也发生了巨大的变化。在短短的几个月的时间里，人们对大模型的认识已经从对其zero-shot能力感到惊讶，转变为考虑改进模型质量、提高模型可用性。 「大语言模型(LLMs)其实就是利用高容量的模型架构（例如Transformer）对海量的、多种多样的数据分布进行建模得到，它包含了大量的先验
阅读更多...
图神经网络模型介绍（1）

图神经网络模型介绍（1）

我们将图神经网络分为基于谱域的模型和基于空域的模型，并按照发展顺序详解每个类别中的重要模型。 1.1基于谱域的图神经网络         谱域上的图卷积在图学习迈向深度学习的发展历程中起到了关键的作用。本节主要介绍三个具有代表性的谱域图神经网络：谱图卷积网络、切比雪夫网络和图卷积网络。 （1）谱图卷积网络 卷积定理：函数卷积的傅里叶变换是函数傅里叶变换的乘积，即F{f*g}
阅读更多...

Copyright China编程 23002807@qq.com

沪ICP备2023033072号-2