★ FEATURED ARTICLE 简介智能算法中台管理系统是一套面向企业智能化转型的Java开发平台完整覆盖样本中心、算法中心与模型中心三大核心模块。样本中心负责数据清洗、预处理与标注算法中心集成支持向量机、决策树、随机森林以及CNN、RNN等机器学习与深度学习算法支持测试、比较与调优模型中心则管理模型创建、训练、验证、部署与监控并提供版本控制和准确率、召回率、F1分数等评估指标适合需要整合算法资源、规范模型全生命周期管理的团队。压缩包共2个文件以SQL数据库脚本和RAR源码包为主SQL文件提供系统初始库表结构和示例数据RAR包内含Java后端及前端代码整体大小38.8MB部署轻量。目前已有559人学习浏览。借助三中心架构读者既能快速理解算法中台落地路径也可直接搭建可运行环境用于毕业设计、课程项目或企业内部系统扩展为算法调度、数据预处理和模型监控等二次开发提供可靠基础。1. 智能算法中台管理系统把散落的算法模型收编成统一服务做Java后端的人大概率都有过这种经历算法工程师训练好的模型躺在服务器目录里没有版本号、没有调用文档上线要靠人肉拷贝监控要靠群里喊一嗓子。这个智能算法中台管理系统干的事就是把算法模型从“黑匣子”变成“标准化服务”——算法包上传、版本管理、在线调试、任务调度、算力监控走一条完整链路开发可以自助申请调用运维能看到每个模型的资源消耗和运行状态。系统面向两类人一类是被算法工程化折腾过的Java开发拿来当天就能把公司里零散的算法脚本管理起来另一类是写算法的同学不用再追着后端要接口自己在控制台上传模型、跑个测试用例就能看到效果。整个项目基于Spring Boot生态搭建数据库用MySQL存元数据、Redis扛热数据调度层用消息队列解耦算法执行走独立的工作节点。下面按照从拿到代码到跑通、再到动手改的顺序把关键模块逐个拆开。2. 技术选型与整体架构Java服务端如何编排算法生命周期2.1 技术栈全景为什么是Spring Boot 消息队列 独立执行节点这个项目的技术栈选型很有代表性。基础框架用Spring Boot 2.x这是当前Java中后台系统最稳妥的选择ORM用MyBatis-Plus分页和条件构造器写起来比原生MyBatis快得多缓存用Redis存算法目录的热点数据和任务状态数据库按“一主一从”的常见部署方式设计DDL脚本里已经带好了主从同步的配置注释。比较关键的选型在任务调度这一层。算法任务往往不是同步返回的——训练一个模型可能要几分钟跑一个批量预测可能更久。项目用了RabbitMQ做异步任务队列任务提交后立刻返回任务ID执行节点从队列里拉消息去跑。这种设计的好处是提交端和执行端完全解耦算法脚本卡死不会拖垮Web服务。执行节点单独拆成一个Spring Boot模块部署时和Web服务分开跑资源隔离做得干净。# application.yml 核心片段 spring: datasource: url: jdbc:mysql://127.0.0.1:3306/algo_platform?useUnicodetruecharacterEncodingutf8 username: root password: your_password redis: host: 127.0.0.1 port: 6379 rabbitmq: host: 127.0.0.1 port: 5672 username: guest password: guest listener: simple: default-requeue-rejected: false algo: worker: script-root: /data/algo-worker/scripts result-root: /data/algo-worker/results参数说明default-requeue-rejected设为false很重要算法脚本抛异常的消息直接进入死信队列不会无限重试把执行节点打爆。script-root是算法包解压后的存放目录result-root是任务结果输出的根目录。这两个目录建议挂到单独的磁盘分区上后面讲到避坑时会具体说为什么。2.2 模块边界控制台、执行节点、存储层各管一摊项目拆成了三个可独立部署的模块。algo-console是管理控制台负责算法目录维护、模型版本管理、任务提交、权限控制是用户直接面对的部分algo-worker是执行节点监听MQ队列拉取算法包、跑脚本、回传结果algo-common放公共类库包括统一返回结构、异常处理、算法包解析工具。存储层分三块MySQL存算法包元数据、模型版本记录、任务历史Redis存算法目录的树形结构缓存、正在运行的任务状态、接口调用频率控制文件系统存算法包本体、执行日志、结果文件。这种“元数据进库、实体文件落盘”的方式比把算法包二进制直接塞进数据库靠谱得多备份和迁移都简单。algo-platform/ ├── algo-console/ │ ├── controller/ # 算法目录、版本、任务提交接口 │ ├── service/ # 业务逻辑层 │ ├── mapper/ # MyBatis-Plus数据访问 │ ├── entity/ # 数据库实体 │ ├── mq/ # 任务消息发送器 │ └── config/ # 安全、Redis、Swagger配置 ├── algo-worker/ │ ├── listener/ # MQ消息监听 │ ├── executor/ # 算法脚本执行器 │ ├── model/ # 任务消息模型 │ └── util/ # 脚本进程工具 ├── algo-common/ │ ├── result/ # 统一返回包装 │ ├── exception/ # 全局异常定义 │ └── utils/ # 算法包解压、MD5校验 └── sql/ ├── schema.sql # 建表语句 └── data.sql # 初始化数据2.3 数据库设计算法包、模型版本、任务三张主表的关系数据库是这套系统的地基三张核心表撑起了所有业务。algo_info表存算法的基本信息算法名称、算法标识、所属分类、描述、当前线上版本algo_version表存每个算法包的发布记录版本号、文件路径、MD5值、上传人、发布时间、状态algo_task表存任务记录任务ID、算法标识、输入参数、状态PENDING/RUNNING/SUCCESS/FAILED、开始时间、结束时间、日志路径。CREATE TABLE algo_info ( id bigint(20) NOT NULL AUTO_INCREMENT, algo_code varchar(64) NOT NULL COMMENT 算法唯一标识, algo_name varchar(128) NOT NULL COMMENT 算法名称, category varchar(64) DEFAULT NULL COMMENT 算法分类, online_version varchar(32) DEFAULT NULL COMMENT 当前线上版本, description varchar(512) DEFAULT NULL, created_by varchar(64) DEFAULT NULL, create_time datetime DEFAULT NULL, PRIMARY KEY (id), UNIQUE KEY uk_algo_code (algo_code) ) ENGINEInnoDB DEFAULT CHARSETutf8mb4 COMMENT算法基本信息表;建表时注意两点algo_code必须唯一它是算法对外暴露的服务标识后面所有的调用都靠它路由版本号不要用int自增用“v1.0.0”这种语义化版本号方便算法同学理解。algo_version表里存MD5值是为了在上传算法包时做一致性校验防止文件在传输过程中损坏。任务表里的algo_code和version联合起来能查到某次任务运行的是哪个版本的算法——排障时这是最有价值的线索。3. 核心模块拆解从算法包上传到任务结果返回的完整链路3.1 算法包上传与校验格式约定、MD5校验、版本自动生成算法包上传是整条链路的起点。项目约定上传的算法包必须是ZIP格式里面至少包含一个main.py作为算法入口和一个requirements.txt记录依赖。控制台接收上传后先算MD5和已有版本做对比同一个文件不允许重复上传然后解析ZIP包内的文件结构校验入口文件是否存在最后自动生成递增的版本号写入algo_version表。public AlgoVersion uploadAlgoPackage(MultipartFile file, Long algoId) { // 校验文件格式 String originalFilename file.getOriginalFilename(); if (!originalFilename.endsWith(.zip)) { throw new BizException(仅支持ZIP格式的算法包); } // 计算MD5 String md5 DigestUtils.md5DigestAsHex(file.getBytes()); // 查询是否已存在相同MD5的版本 LambdaQueryWrapperAlgoVersion wrapper new LambdaQueryWrapper(); wrapper.eq(AlgoVersion::getMd5, md5); if (this.getOne(wrapper) ! null) { throw new BizException(该算法包已存在请勿重复上传); } // 存储文件到磁盘 String storedPath fileStorageService.store(file); // 生成新版本号 String nextVersion versionService.generateNextVersion(algoId); // 解析ZIP包结构 ZipStructure zipInfo ZipParser.parse(storedPath); if (!zipInfo.containsEntry(main.py)) { FileUtils.deleteQuietly(new File(storedPath)); throw new BizException(算法包内缺少main.py入口文件); } // 写入数据库 AlgoVersion entity new AlgoVersion(); entity.setAlgoId(algoId); entity.setVersion(nextVersion); entity.setFilePath(storedPath); entity.setMd5(md5); entity.setStatus(0); // 0-待发布 1-已发布 2-下架 this.save(entity); return entity; }逻辑说明这个方法分六步走——校验扩展名、计算MD5、查重、落盘、解析ZIP、写库。其中versionService.generateNextVersion的实现是查该算法已有的最大版本号然后主版本号加1。这里有个细节处理上传失败时手动删除已落盘的文件避免产生孤儿文件占用磁盘。ZIP解析用的是Java自带的ZipInputStream遍历entry时只校验文件是否存在不递归解压安全性考虑到位。参数说明MultipartFile是Spring MVC接收文件的标准类型前端用表单格式提交字段名和Controller里RequestParam(file)对应即可。fileStorageService.store内部做了目录按月分片的处理/data/algo-packages/2024/06/这样的结构避免单个目录下文件过多影响IO性能。3.2 任务提交与调度MQ消息结构、队列路由、幂等处理任务提交是Web侧最核心的入口。用户调用POST /api/task/submit传算法标识和参数字典控制台先做参数合法性校验然后组装一个AlgoTaskMessage发到RabbitMQ的algo.task.queue队列。消息结构里包含任务ID、算法包路径、参数JSON、回调地址——执行节点拿到后可以直接干活。public String submitTask(Long algoId, MapString, Object params) { AlgoInfo algoInfo algoInfoService.getById(algoId); if (algoInfo null) { throw new BizException(算法不存在); } // 找到当前线上版本 AlgoVersion onlineVersion algoVersionService.getOnlineVersion(algoId); if (onlineVersion null) { throw new BizException(该算法未发布任何版本); } // 生成任务ID String taskId UUID.randomUUID().toString().replace(-, ); AlgoTaskMessage message new AlgoTaskMessage(); message.setTaskId(taskId); message.setAlgoCode(algoInfo.getAlgoCode()); message.setVersion(onlineVersion.getVersion()); message.setScriptPath(onlineVersion.getFilePath()); message.setParams(params); // 发送到MQ rabbitTemplate.convertAndSend(algo.task.exchange, algo.task.routing, message); // 记录任务状态 algoTaskService.saveTask(taskId, algoId, onlineVersion.getVersion(), JSON.toJSONString(params), PENDING); return taskId; }逻辑说明提交接口的返回值是taskId不是算法结果本身。这个设计是异步任务系统和同步接口最大的区别——调用方拿到任务ID后可以轮询GET /api/task/result/{taskId}查询结果也可以等执行节点回调。convertAndSend第三个参数传的是消息体RabbitTemplate会自动序列化为JSON存入队列。如果发送失败比如MQ不可用代码会抛出异常事务回滚保证任务状态不会停留在PENDING。参数说明params是MapString, Object类型对应算法脚本里的输入参数执行节点运行时会把这个Map序列化成JSON文件传给算法入口。需要注意的是参数的值只能是基本类型或JSON可序列化的对象如果传了复杂的Java对象消息体序列化时会直接报错。3.3 执行节点监听队列、拉起Python进程、日志回传执行节点是真正跑算法的地方。它监听algo.task.queue拿到消息后先更新任务状态为RUNNING然后在独立的工作目录下解压算法包用ProcessBuilder拉起Python子进程执行main.py把参数JSON和结果JSON的路径通过命令行参数传进去。RabbitListener(queues algo.task.queue) public void handleTask(AlgoTaskMessage message) { AlgoTaskLog taskLog new AlgoTaskLog(); taskLog.setTaskId(message.getTaskId()); taskLog.setStatus(RUNNING); // 创建工作目录/data/algo-worker/workspace/{taskId} Path workDir Paths.get(workRoot, message.getTaskId()); FileUtils.forceMkdir(workDir.toFile()); // 解压算法包 ZipUtils.unzip(message.getScriptPath(), workDir.toString()); // 构造Python命令 ListString command new ArrayList(); command.add(python3); command.add(main.py); command.add(--params workDir.resolve(params.json)); command.add(--output workDir.resolve(result.json)); ProcessBuilder builder new ProcessBuilder(command); builder.directory(workDir.toFile()); builder.redirectErrorStream(true); long startTime System.currentTimeMillis(); try { Process process builder.start(); // 捕获输出流写日志 try (BufferedReader reader new BufferedReader( new InputStreamReader(process.getInputStream()))) { String line; while ((line reader.readLine()) ! null) { taskLog.appendLog(line); } } int exitCode process.waitFor(); if (exitCode 0) { taskLog.setStatus(SUCCESS); } else { taskLog.setStatus(FAILED); taskLog.setError(Python进程退出码: exitCode); } } catch (Exception e) { taskLog.setStatus(FAILED); taskLog.setError(ExceptionUtils.getStackTrace(e)); } // 持久化日志和状态 taskLogService.save(taskLog); FileUtils.deleteQuietly(workDir.toFile()); }逻辑说明这段是执行节点的核心逻辑。RabbitListener是Spring AMQP提供的监听注解方法收到消息后自动执行。redirectErrorStream(true)把Python的stderr合并到stdout这样算法脚本的print和traceback都能被统一读到日志文件里。执行完成后无论成功失败都会清理工作目录这是一个容易被忽略的运维细节——如果每次跑完不清理磁盘早晚被临时文件和中间产物塞满。参数说明--params和--output是传给Python脚本的自定义参数算法同学在使用规范里会被要求解析这两个参数。waitFor()会阻塞当前线程直到子进程结束这里存在一个隐患——如果算法脚本死循环或者等待输入进程会一直挂着。改进方案是把waitFor改成带超时时间的方法比如process.waitFor(30, TimeUnit.MINUTES)超时直接destroyForcibly()这个在避坑章节会展开讲。3.4 模型版本发布与灰度不删历史数据、不打断线上服务版本发布是算法中台最有价值的功能之一。算法同学把新模型打包上传后版本状态是“待发布”此时只有开发环境能调用管理台点“发布”状态变更为“上线”下一次任务提交才会路由到新版本。这里用了一个简单的灰度逻辑发布接口里可以传一个weight参数从0到100决定新版本承接流量的百分比。Transactional(rollbackFor Exception.class) public void publishVersion(Long versionId, Integer weight) { AlgoVersion newVersion algoVersionService.getById(versionId); if (newVersion null || newVersion.getStatus() ! 0) { throw new BizException(版本状态异常无法发布); } AlgoInfo algoInfo algoInfoService.getById(newVersion.getAlgoId()); Integer oldWeight algoInfo.getOnlineWeight(); // 新版本权重为0时只对测试请求生效 if (weight null || weight 0 || weight 100) { throw new BizException(灰度权重必须在0-100之间); } // 更新算法信息表的线上版本 algoInfoService.updateOnlineVersion(newVersion.getAlgoId(), newVersion.getVersion()); // 更新灰度权重 algoInfoService.updateWeight(newVersion.getAlgoId(), weight); // 旧版本状态改为已下线不删除保留历史 algoVersionService.updateStatus(newVersion.getAlgoId(), newVersion.getVersion(), 2); newVersion.setStatus(1); algoVersionService.updateById(newVersion); }逻辑说明这个方法的精髓在于“不删历史数据”。版本下线只是改了状态标记算法包文件仍然保留在磁盘上。好处是第一线上出了问题随时可以回滚到上一个版本第二历史任务排查时需要知道当时跑的是哪个版本记录还完整存在。灰度权重存在算法信息表里任务提交时根据权重做随机路由——这个随机逻辑放在submitTask方法里根据ThreadLocalRandom.current().nextInt(100)和权重的比较来选择版本。这里要注意事务边界加Transactional是因为要把版本状态更新和算法信息更新绑在一起要么都成功要么都失败。但MQ消息发送不在事务内所以会有一种极端情况——事务提交成功后MQ发送失败导致数据库显示已发布但任务跑不起来。常见做法是把发送MQ的代码放到事务提交后的监听器里或者用本地消息表记录待发送状态定时任务扫表补发。这个项目用了前一种方案。4. 环境搭建与部署从零开始跑起整套系统的完整步骤4.1 本地开发环境初始化JDK、MySQL、Redis、RabbitMQ跑这套系统之前先把环境准备好。JDK要求1.8以上推荐用1.8版本项目里没有用到高版本特性MySQL要求5.7以上建库时注意字符集要指定utf8mb4不然算法名称里存中文会乱码Redis不需要特殊配置默认端口即可RabbitMQ建议安装3.8以上版本安装完要确认guest账号能本地登录。# 以CentOS 7为例安装基础依赖 yum install -y java-1.8.0-openjdk-devel mysql-server redis rabbitmq-server # 启动MySQL并初始化 systemctl start mysqld mysql -uroot -p sql/schema.sql # 启动Redis和RabbitMQ systemctl start redis systemctl start rabbitmq-server rabbitmq-plugins enable rabbitmq_management命令说明schema.sql文件放在项目根目录的 sql 文件夹下一次性执行会创建全部数据表。执行完成后可以用show tables;验证一下应该能看到algo_info、algo_version、algo_task等表。如果MySQL是8.0版本要注意用户认证方式默认是caching_sha2_passwordSpring Boot连接会报错需要在建用户时指定mysql_native_password。4.2 启动Worker和Console端口规划、配置文件修改、验证接口环境就绪后先启动执行节点再启动控制台。执行节点默认端口是8081控制台是8080。修改application.yml里的数据库密码、MQ连接信息、文件存储路径然后分别用Maven打包启动。# 打包在项目根目录执行 mvn clean package -DskipTests # 启动worker节点 java -jar algo-worker/target/algo-worker-1.0.0.jar \ --server.port8081 \ --algo.worker.script-root/data/algo-worker/scripts \ --algo.worker.result-root/data/algo-worker/results # 启动console控制台 java -jar algo-console/target/algo-console-1.0.0.jar \ --server.port8080 \ --spring.datasource.passwordyour_password参数说明命令里用--前缀覆盖了配置文件里的默认值这样不用改文件就能跑。script-root是执行节点解压算法包的工作根目录result-root是结果输出目录启动前确保目录已创建并给了写权限。控制台启动后访问http://localhost:8080/swagger-ui/index.html能看到接口文档在Swagger页面上可以直接测试“上传算法包”和“提交任务”两个核心接口。验证流程是这样的先在Swagger里调用“创建算法信息”接口录入一个算法标识demo_forecast然后上传一个最简单的ZIP算法包里面main.py只打印挣输入参数再调用任务提交接口传入{days: 7}这样的预测天数参数最后轮询任务结果接口能看到返回的JSON文件内容。整个过程跑通说明Web、MQ、Worker、文件存储这四层已经串联正常。4.3 Docker Compose一键编排本地体验与生产部署的差距项目里带了一个docker-compose.yml把MySQL、Redis、RabbitMQ、Console、Worker五个服务编排在一起。用Docker跑的好处是不用在自己机器上装一堆依赖拿来即用坏处是生产环境和本地容器镜像版本有差异踩坑概率反而更高。如果只是想快速看效果建议用Docker Compose如果要正式部署还是推荐拆开手动管理。version: 3.8 services: mysql: image: mysql:5.7 environment: MYSQL_ROOT_PASSWORD: root123 MYSQL_DATABASE: algo_platform ports: - 3306:3306 volumes: - ./sql/schema.sql:/docker-entrypoint-initdb.d/schema.sql redis: image: redis:6-alpine ports: - 6379:6379 rabbitmq: image: rabbitmq:3.8-management ports: - 5672:5672 - 15672:15672 console: build: ./algo-console depends_on: - mysql - redis - rabbitmq ports: - 8080:8080 worker: build: ./algo-worker depends_on: - queue - mysql ports: - 8081:8081注意schema.sql挂载到MySQL容器的初始化目录这是MySQL官方镜像支持的启动时自动执行脚本机制。Console和Worker的Dockerfile里做了分层构建先Maven打包再打镜像这里有个坑Maven构建阶段会拉取依赖如果网络不好构建很慢建议在宿主机先mvn install把依赖打到本地仓库再用-o离线模式构建。5. 踩坑记录从运行到上线最容易翻车的几个问题5.1 MySQL字符集导致算法名称乱码现象算法名称里带有中文描述保存到数据库后变成问号或乱码。排查原因建库时没有指定utf8mb4字符集MySQL 5.7默认的latin1不支持中文。解决重建数据库并指定字符集CREATE DATABASE algo_platform DEFAULT CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci;同时检查JDBC连接串是否带了characterEncodingutf8。这个坑几乎每个连MySQL的应用都会踩到我一般会在建库脚本里直接写死字符集不给默认值留余地。5.2 RabbitMQ消费失败导致消息无限重试现象算法脚本因为缺依赖抛了异常任务一直不结束日志里反复出现同样的报错MQ的管理台显示queue堆积。查明原因Spring AMQP默认的defaultRequeueRejected为true消息消费失败会被重新放回队列一直循环。这样的后果是故障消息把队列堵死后续正常任务全部阻塞。解决在配置文件里把default-requeue-rejected设为false让失败消息进死信队列同时监听死信队列做人工介入处理。从那以后我每次搭MQ消费者第一件事就是设置这个参数。5.3 Python子进程超时无自动中断现象算法脚本卡在某个循环里一直不返回任务状态永远是RUNNING磁盘空间被逐步写满。原因Process.waitFor()不带超时参数会无限期等待子进程。解决给waitFor传超时时间和时间单位比如process.waitFor(30, TimeUnit.MINUTES)超时后执行process.destroyForcibly()杀掉进程同时把任务标记为FAILED并写入超时错误信息。另外可以再加一层保障执行节点启动时监听一个守护线程周期性检查所有运行中任务的执行时长超过阈值直接清理。5.4 上传大文件时Nginx请求超时现象算法包超过200MB上传时浏览器等待很久后报504 Gateway Timeout。原因Nginx的client_max_body_size默认只有1m且代理转发到后端有默认超时时间。解决在Nginx配置里设置client_max_body_size 1024m;和proxy_read_timeout 600s;同时Spring Boot的spring.servlet.multipart.max-file-size也要同步调大。注意这两个地方必须保持一致前端如果走网关网关的超时时间也要改不然总有某一个环节提前掐断连接。5.5 日志文件与结果文件路径混淆现象排障时看任务日志发现里面混着其他任务的内容。原因执行节点把日志写到共享目录/tmp/algo-logs多个任务并发时日志文件互相追加。解决任务日志按任务ID隔离目录结构改成/tmp/algo-logs/{taskId}/run.log清理逻辑用任务ID匹配而不是时间扫全目录。这个问题的教训是所有临时文件必须纳入任务ID维度的隔离不能依赖文件名唯一性。6. 二次开发把中台改造成贴合自己业务的三个方向6.1 增加算法超时自动熔断与告警通知到手跑通之后第一个值得改的地方是任务执行的鲁棒性。前面提到waitFor超时的问题我在实际改造时加了一个TaskTimeoutGuard的定时任务组件每30秒扫描一次正在运行的任务如果超过配置的超时阈值就调用进程强制销毁同时往钉钉群发一个告警——这样算法卡死不用等人发现系统自己会报告。Component public class TaskTimeoutGuard { Autowired private TaskRuntimeRegistry taskRegistry; Autowired private AlarmNotifier alarmNotifier; Scheduled(fixedDelay 30000) public void scan() { MapString, Process runningTasks taskRegistry.getRunningTasks(); long now System.currentTimeMillis(); for (Map.EntryString, Process entry : runningTasks.entrySet()) { long startTime taskRegistry.getStartTime(entry.getKey()); long timeoutMs taskRegistry.getTimeoutMs(entry.getKey()); if (now - startTime timeoutMs) { entry.getValue().destroyForcibly(); taskRegistry.unregister(entry.getKey()); alarmNotifier.send(任务 entry.getKey() 执行超时已强制终止); } } } }逻辑说明TaskTimeoutGuard的核心是用Scheduled注解实现固定频率扫描配合一个内存注册表TaskRuntimeRegistry记录当前由本节点执行的任务、启动时间、超时阈值。destroyForcibly()会强杀子进程unregister清理注册表。这两个方法的实现可以在algo-worker模块里直接扩展不影响Web侧代码。改造时要注意一个边界单节点杀进程没问题但如果有多台worker节点内存注册表只在进程内部可见扫描只能覆盖本节点任务。有条件的话可以改成把任务注册信息写到Redis用ZSet按启动时间排序所有worker节点共享一套超时扫描逻辑。6.2 支持HTTP在线调用从消息队列模式扩展到同步API原项目默认走MQ异步模式但很多内部对接方更希望有同步接口——即拿到任务提交后阻塞等待结果类似RPC调用。可以在algo-console里增加一个“在线预测”接口先复用submitTask把任务发到MQ然后用Redis的发布订阅或者轮询等待结果加上一个超时时间上限。PostMapping(/api/predict/sync) public AlgoTaskResult syncPredict(RequestBody SyncPredictRequest request) { String taskId taskService.submitTask(request.getAlgoId(), request.getParams()); long deadline System.currentTimeMillis() 20000; // 20秒超时 while (System.currentTimeMillis() deadline) { AlgoTaskResult result taskService.getResult(taskId); if (result ! null) { return result; } try { Thread.sleep(300); } catch (InterruptedException e) { Thread.currentThread().interrupt(); throw new BizException(等待结果被中断); } } throw new BizException(在线预测超时请使用异步模式获取结果); }逻辑说明这种方式本质上是把异步任务转成同步等待适用场景是算法本身执行很快比如一两秒内出结果的内部调用。轮询间隔设置为300毫秒避免高频打Redis。如果任务跑到20秒还没结果直接抛异常让调用方走异步接口——这个超时设计很重要不然同步接口挂起太久会占满服务器的线程池。从架构角度看不太优雅但作为快速接入的过渡方案很实用。6.3 扩展算法类型从纯Python脚本到支持镜像执行项目的算法执行目前依赖宿主机Python环境如果算法需要特定版本的TensorFlow或自定义运行环境排列组合会变得不可维护。这块我做过的方案是在algo_version表增加一个runtime_type字段取值python或docker执行节点判断runtime_type如果为docker就不直接启Python进程而是docker run指定的镜像把算法包挂载进容器参数和结果通过挂载的JSON文件传递。if (docker.equals(message.getRuntimeType())) { ListString dockerCmd Arrays.asList( docker, run, --rm, -v, workDir :/workspace, -w, /workspace, imageName, python3, main.py, --params/workspace/params.json, --output/workspace/result.json ); builder new ProcessBuilder(dockerCmd); }逻辑说明这里用到docker run --rm参数容器执行完自动删除不残留容器实例。挂载卷-v把宿主机的任务工作目录映射到容器内的/workspace算法包解压、参数、结果都在这个目录里容器内外对文件的操作天然隔离。改造时要注意镜像拉取策略——如果每次执行都重新拉取镜像会非常慢可以在添加算法信息时预先把镜像docker pull到worker节点运行时直接docker run。6.4 数据权限细化目录级别的算法可见性控制原项目的权限控制是简单的用户-角色-算法三层模型每个用户可以分配可访问的算法标识列表。在二次开发时可以参照algo_info表里的category字段把权限粒度细化到目录级别——用户被分配到某个算法分类分类下所有算法自动可见新上传到该分类的算法也不用重复授权。CREATE TABLE user_category_permission ( id bigint(20) NOT NULL AUTO_INCREMENT, user_id bigint(20) NOT NULL, category varchar(64) NOT NULL, permission_level tinyint(4) DEFAULT 1 COMMENT 1-只读 2-调用 3-管理, PRIMARY KEY (id), UNIQUE KEY uk_user_category (user_id, category) ) ENGINEInnoDB DEFAULT CHARSETutf8mb4 COMMENT用户算法目录权限表;权限判断的逻辑放在submitTask接口的入口处从登录用户的上下文里拿到用户ID查user_category_permission表再和算法信息的category做匹配没有权限直接抛403。注意这里要加缓存不然每个任务提交都查一次数据库在线接口的响应时间会明显上升。我改造完这套权限模型后有个习惯每次新增算法分类第一时间检查现有用户是否有对应的目录权限而不是依赖默认的全量放行——权限这种事宁可保守一点出问题再宽松比出事故再收紧好处理得多。希望这些思路能帮你把这套智能算法中台管理系统更快落到自己的业务里。本文还有配套的精品资源点击获取
阅读完成 · 觉得有帮助？
