# BI APP
**Repository Path**: whiteinjava/bi-app
## Basic Information
- **Project Name**: BI APP
- **Description**: 智能BI平台
- **Primary Language**: Java
- **License**: Not specified
- **Default Branch**: master
- **Homepage**: None
- **GVP Project**: No
## Statistics
- **Stars**: 3
- **Forks**: 1
- **Created**: 2023-10-26
- **Last Updated**: 2025-01-09
## Categories & Tags
**Categories**: Uncategorized
**Tags**: None
## README
# BI智能平台说明文档
# 背景诉求
BI即所谓商业智能，是一个数据可视化的数据分析系统。市场上已有的主流BI平台，像帆软BI以及微软的Power BI。这个项目与上述主流的BI平台类似，不同于传统BI平台给使用人员带来的专业上的局限性；使用该BI平台能够通过输入诉求以及上传原始数据即可实现便捷的数据分析以及可视化。
## 1.整体需求分析
- 用户输入分析需求，上传原始数据，能够得到想要的诉求结果
- 用户、图表管理
- 由于API接口调用可能存在并发延迟，因此要求图表的生成能够实现异步化
- 后端服务对接AI
## 2.系统架构
### 2.1 基础架构
### 2.2 进阶架构
## 3.技术栈选型
### 3.1 前端
- React
- Umi + Ant Design Pro
- 可视化开发库 Echarts
- Umi openAPI后端代码生成
### 3.2 后端
- SpringBoot、SpringMVC
- MyBatis-Plus
- MySQL
- RabbitMQ
- Easy Excel 图表数据的上传解析
- Swagger + Knife4j 文档
- HuTool、Lombok
- openAPI接口开发
## 第一期 - 项目初始化
### 1. 前端Ant Design Pro初始化
### 2. 后端SpringBoot初始化
#### 2.1 Maven依赖项
```xml
4.0.0
org.springframework.boot
spring-boot-starter-parent
2.7.2
com.shuai
BaiBI
0.0.1-SNAPSHOT
BaiBI
1.8
org.springframework.boot
spring-boot-starter-freemarker
org.springframework.boot
spring-boot-starter-web
org.springframework.boot
spring-boot-starter-aop
org.mybatis.spring.boot
mybatis-spring-boot-starter
2.2.2
com.baomidou
mybatis-plus-boot-starter
3.5.2
org.springframework.boot
spring-boot-starter-data-redis
org.springframework.session
spring-session-data-redis
org.springframework.boot
spring-boot-starter-data-elasticsearch
com.github.xiaoymin
knife4j-spring-boot-starter
3.0.3
com.qcloud
cos_api
5.6.89
org.apache.commons
commons-lang3
com.google.code.gson
gson
2.9.1
com.alibaba
easyexcel
3.1.1
cn.hutool
hutool-all
5.8.8
org.springframework.boot
spring-boot-devtools
runtime
true
mysql
mysql-connector-java
runtime
org.springframework.boot
spring-boot-configuration-processor
true
org.projectlombok
lombok
true
org.springframework.boot
spring-boot-starter-test
test
org.springframework.boot
spring-boot-maven-plugin
org.projectlombok
lombok
```
#### 2.2 SpringBoot配置文件
```yaml
spring:
application:
name: BaiBI # 项目名称
profiles:
active: dev # 默认开发环境
# 支持swagger3
mvc:
pathmatch:
matching-strategy: ant_path_matcher
session:
timeout: 2592000 # 配置session 30天过期
# 配置数据库链接
datasource:
driver-class-name: com.mysql.cj.jdbc.Driver
url: jdbc:mysql://localhost:3306/baixs_bi_db
username: root
password: 123456
servlet:
multipart:
max-file-size: 10MB # 配置上传文件的大小限制
server:
port: 8081
servlet:
context-path: /api
session:
cookie:
max-age: 2592000 # 设置cookie 30天过期
# MyBatis-Plus配置
mybatis-plus:
configuration:
map-underscore-to-camel-case: false
log-impl: org.apache.ibatis.logging.stdout.StdOutImpl
global-config:
db-config:
logic-delete-field: isDelete # 全局逻辑删除的实体字段名
logic-delete-value: 1 # 逻辑已删除值（默认为 1）
logic-not-delete-value: 0 # 逻辑未删除值（默认为 0）
```
#### 2.3 统一返回数据格式
```java
/**
* 通用返回类
*
* @param
* @author 程序员鱼皮
* @from 编程导航知识星球
*/
@Data
public class BaseResponse implements Serializable {
private int code;
private T data;
private String message;
public BaseResponse(int code, T data, String message) {
this.code = code;
this.data = data;
this.message = message;
}
public BaseResponse(int code, T data) {
this(code, data, "");
}
public BaseResponse(ErrorCode errorCode) {
this(errorCode.getCode(), null, errorCode.getMessage());
}
}
```
```java
/**
* 返回工具类
*
* @author 程序员鱼皮
* @from 编程导航知识星球
*/
public class ResultUtils {
/**
* 成功
*
* @param data
* @param
* @return
*/
public static BaseResponse success(T data) {
return new BaseResponse<>(0, data, "ok");
}
/**
* 失败
*
* @param errorCode
* @return
*/
public static BaseResponse error(ErrorCode errorCode) {
return new BaseResponse<>(errorCode);
}
/**
* 失败
*
* @param code
* @param message
* @return
*/
public static BaseResponse error(int code, String message) {
return new BaseResponse(code, null, message);
}
/**
* 失败
*
* @param errorCode
* @return
*/
public static BaseResponse error(ErrorCode errorCode, String message) {
return new BaseResponse(errorCode.getCode(), null, message);
}
}
```
```java
/**
* 自定义错误码
*/
public enum ErrorCode {
SUCCESS(0, "ok"),
PARAMS_ERROR(40000, "请求参数错误"),
NOT_LOGIN_ERROR(40100, "未登录"),
NO_AUTH_ERROR(40101, "无权限"),
NOT_FOUND_ERROR(40400, "请求数据不存在"),
FORBIDDEN_ERROR(40300, "禁止访问"),
SYSTEM_ERROR(50000, "系统内部异常"),
OPERATION_ERROR(50001, "操作失败");
/**
* 状态码
*/
private final int code;
/**
* 信息
*/
private final String message;
ErrorCode(int code, String message) {
this.code = code;
this.message = message;
}
public int getCode() {
return code;
}
public String getMessage() {
return message;
}
}
```
#### 2.4 统一应用异常处理
```java
/**
* 自定义业务异常类
*/
public class BusinessException extends RuntimeException {
/**
* 错误码
*/
private final int code;
public BusinessException(int code, String message) {
super(message);
this.code = code;
}
public BusinessException(ErrorCode errorCode) {
super(errorCode.getMessage());
this.code = errorCode.getCode();
}
public BusinessException(ErrorCode errorCode, String message) {
super(message);
this.code = errorCode.getCode();
}
public int getCode() {
return code;
}
}
```
```java
/**
* 全局异常处理器
*/
@RestControllerAdvice
@Slf4j
public class GlobalExceptionHandler {
@ExceptionHandler(BusinessException.class)
public BaseResponse businessExceptionHandler(BusinessException e) {
log.error("BusinessException", e);
return ResultUtils.error(e.getCode(), e.getMessage());
}
@ExceptionHandler(RuntimeException.class)
public BaseResponse runtimeExceptionHandler(RuntimeException e) {
log.error("RuntimeException", e);
return ResultUtils.error(ErrorCode.SYSTEM_ERROR, "系统错误");
}
}
```
#### 2.5 Mybatis-Plus配置
##### 2.5.1 分页插件配置
```java
@Configuration
@MapperScan("com.shuai.springbootinit.mapper")
public class MyBatisPlusConfig {
/**
* 拦截器配置
*/
@Bean
public MybatisPlusInterceptor mybatisPlusInterceptor() {
MybatisPlusInterceptor interceptor = new MybatisPlusInterceptor();
// 分页插件
interceptor.addInnerInterceptor(new PaginationInnerInterceptor(DbType.MYSQL));
return interceptor;
}
}
```
##### 2.5.2 日志输出与逻辑删除字段配置
```yaml
# MyBatis-Plus配置
mybatis-plus:
configuration:
map-underscore-to-camel-case: false
log-impl: org.apache.ibatis.logging.stdout.StdOutImpl
global-config:
db-config:
logic-delete-field: isDelete # 全局逻辑删除的实体字段名
logic-delete-value: 1 # 逻辑已删除值（默认为 1）
logic-not-delete-value: 0 # 逻辑未删除值（默认为 0）
```
#### 2.6 用户密码处理工具类
```java
/**
* 加密思路：
* 用户密码加盐的基本思路为：
* 随机生成一个去除分隔符的32位UUID；
* 将生成的32位UUID与用户明文密码拼接进行MD5加密；
* 将32位UUID与MD5加密后的（UUID + 用户密码）进行拼再添加一个分隔符，方便后续校验密码，将其存储到数据库中
* 最终的密文密码格式为：生成的32位UUID + # + MD5(生成的32位UUID + 用户明文密码)
*/
/**
* 解密思路：
* 先从数据库中根据用户名获得用户信息
* 将用户加密后的密码的前32位UUID取出
* 将取出的32位UUID与用户输入的密码拼接并进行MD5加密
* 将32位UUID与MD5加密后的32位UUID和用户明文密码进行拼接
* 比较按照用户明文密码加密后的密文和数据库中存储的密文是否相同
*/
public abstract class PasswordUtil {
private static final String SEPARATOR = "#";
//加密
public static String encryptPwd(String password) {
String frontMatter = UUID.randomUUID().toString().replace("-", ""); //生成32位的一个前置校验
String endMatter = DigestUtils.md5DigestAsHex((frontMatter + password).getBytes()); // 将明文密码和前置校验拼接进行md5加密
return frontMatter + SEPARATOR + endMatter; // 将前置校验加上一个$加上前置校验和明文密码加在一起加密后的后置校验共65位
}
private static String getExpectPwd(String frontMatter, String originalPwd) {
return frontMatter + SEPARATOR + DigestUtils.md5DigestAsHex((frontMatter + originalPwd).getBytes());
}
//校验
public static boolean check(String encryptedPwd, String originalPwd) {
//取得加密后密码的前32位
String frontMatter = encryptedPwd.split(SEPARATOR)[0];
String expectPwd = getExpectPwd(frontMatter, originalPwd);
return encryptedPwd.equals(expectPwd);
}
}
```
### 3.库表设计
```sql
drop database if exists baixs_bi_db;
-- 创建库
create database if not exists baixs_bi_db;
-- 切换库
use baixs_bi_db;
-- 用户表
drop table if exists user;
create table if not exists user
(
id bigint auto_increment comment 'id' primary key,
userAccount varchar(256) not null comment '账号',
userPassword varchar(512) not null comment '密码',
userName varchar(256) null comment '用户昵称',
userAvatar varchar(1024) null comment '用户头像',
userRole varchar(256) default 'user' not null comment '用户角色：user/admin',
createTime datetime default CURRENT_TIMESTAMP not null comment '创建时间',
updateTime datetime default CURRENT_TIMESTAMP not null on update CURRENT_TIMESTAMP comment '更新时间',
isDelete tinyint default 0 not null comment '是否删除',
index idx_unionId (userAccount)
) comment '用户' collate = utf8mb4_unicode_ci;
-- 图表信息表
drop table if exists chart;
create table if not exists chart
(
id bigint auto_increment comment 'id' primary key,
`name` varchar(128) null comment '图标名称';
goal text null comment '分析目标',
chartType varchar(128) null comment '图表类型',
chartData text null comment '图表数据',
genChartData text null comment '生成的图表数据',
genChartResult text null comment '生成的分析结论',
userId bigint not null comment '创建用户 id',
createTime datetime default CURRENT_TIMESTAMP not null comment '创建时间',
updateTime datetime default CURRENT_TIMESTAMP not null on update CURRENT_TIMESTAMP comment '更新时间',
isDelete tinyint default 0 not null comment '是否删除'
) comment '帖子' collate = utf8mb4_unicode_ci;
```
---
## 第二期 - 用户管理和前后端联调测试
### 接口处理逻辑
### 1. 用户注册
```java
@PostMapping("/register")
public BaseResponse register(@RequestBody UserRegisterRequest userRegisterRequest) {
if (userRegisterRequest == null) {
throw new BusinessException(ErrorCode.PARAMS_ERROR);
}
String userAccount = userRegisterRequest.getUserAccount();
String checkPassword = userRegisterRequest.getCheckPassword();
String userPassword = userRegisterRequest.getUserPassword();
if (StringUtils.isAnyBlank(userAccount, userPassword, checkPassword)) {
throw new BusinessException(ErrorCode.PARAMS_ERROR);
}
// 检验输入密码和校验密码是否一致
if (!StringUtils.equals(userPassword, checkPassword)) {
throw new BusinessException(ErrorCode.PARAMS_ERROR, "密码和校验密码不一致！");
}
// 对用户输入的密码进行加密
User user = new User();
String encryptedPwd = PasswordUtil.encryptPwd(userPassword);
BeanUtils.copyProperties(userRegisterRequest, user);
user.setUserPassword(encryptedPwd);
// 保存注册用户
synchronized (lock) {
// 校验数据库中是否已经有该用户名
LambdaQueryWrapper userQueryWrapper = new LambdaQueryWrapper<>();
userQueryWrapper.eq(User::getUserAccount, userAccount);
if (userService.count(userQueryWrapper) > 0) {
throw new BusinessException(ErrorCode.PARAMS_ERROR, "用户名已存在");
}
// 保存用户
userService.save(user);
}
UserVO userVO = new UserVO();
BeanUtils.copyProperties(user, userVO);
return ResultUtils.success(userVO);
}
```
```json
# 请求参数
{
"checkPassword": "12345678",
"userAccount": "root",
"userPassword": "12345678"
}
# 响应参数
{
"code": 0,
"data": {
"id": "1717523771382636546",
"userAccount": "root",
"userName": null,
"userAvatar": null,
"userRole": null,
"createTime": null,
"updateTime": null,
"isDelete": null
},
"message": "ok"
}
```
### 2. 用户登录
```java
@PostMapping("/login")
public BaseResponse login(@RequestBody UserLoginRequest userLoginRequest, HttpServletRequest request) {
if (userLoginRequest == null
|| StringUtils.isAllBlank(userLoginRequest.getUserAccount(), userLoginRequest.getUserPassword())) {
throw new BusinessException(ErrorCode.PARAMS_ERROR);
}
// 根据用户名从数据库中查询用户信息
LambdaQueryWrapper userQueryWrapper = new LambdaQueryWrapper<>();
String userAccount = userLoginRequest.getUserAccount();
String userPassword = userLoginRequest.getUserPassword();
userQueryWrapper.eq(User::getUserAccount, userAccount);
User user = userService.getOne(userQueryWrapper);
if (user == null) {
throw new BusinessException(ErrorCode.PARAMS_ERROR, "用户不存在！");
}
if (!PasswordUtil.check(user.getUserPassword(), userPassword)) {
throw new BusinessException(ErrorCode.PARAMS_ERROR, "用户名或密码错误！");
}
// 存储用户会话对象
request.getSession().setAttribute(UserConstant.LOGIN_STATE_KEY, user.getId());
// 返回登录成功信息
UserVO userVO = new UserVO();
BeanUtils.copyProperties(user, userVO);
return ResultUtils.success(userVO);
}
```
```json
# 请求参数
{
"userAccount": "root",
"userPassword": "12345678"
}
# 响应参数
{
"code": 0,
"data": {
"id": "1717523771382636546",
"userAccount": "root",
"userName": null,
"userAvatar": null,
"userRole": "user",
"createTime": "2023-10-26T12:49:11.000+00:00",
"updateTime": "2023-10-26T12:49:11.000+00:00",
"isDelete": 0
},
"message": "ok"
}
```
### 3.联调用户登录前后端
```jsx
openAPI: [
{
requestLibPath: "import { request } from '@umijs/max'",
// 或者使用在线的版本
// schemaPath: "https://gw.alipayobjects.com/os/antfincdn/M%24jrzTTYJN/oneapi.json",
schemaPath: "http://localhost:8081/api/v2/api-docs",
// schemaPath: join(__dirname, 'oneapi.json'),
projectName: 'baiBI_front',
mock: false,
},
],
```
```jsx
/**
* 在用户登录页面发送登录请求，测试前后端联调效果
*/
useEffect(() => {
loginUsingPOST({}).then(res => {
console.log(res)
})
})
```
## 第三期 - 智能AI分析
### 需求分析
1. 后端校验上传参数正确性，并进行AI接口调用的成本控制（次数统计限制、用户鉴权）
2. 由于openAPI接口调用字数的限制以及准确性问题，手动优化用户分析需求
3. 将用户上传的excel格式的文件转换为csv格式
4. 根据用户提问，创建用户提词
### excel转csv
```java
/**
* 将用户上传的excel文件转换成字符串
*/
@Slf4j
public class ExcelHandleUtils {
public static String handleExcelToStr(MultipartFile file) {
List> dataList = null ;
try {
dataList = EasyExcel.read(file.getInputStream())
.excelType(ExcelTypeEnum.XLSX)
.sheet()
.headRowNumber(0)
.doReadSync();
} catch (IOException e) {
log.error("读取excel文件出错");
throw new BusinessException(ErrorCode.SYSTEM_ERROR, "读取excel文件出错");
}
if (dataList == null || dataList.size() == 0) {
return null;
}
StringBuilder csvStr = new StringBuilder();
// 获取表头数据
List headerData = dataList.get(0).values().stream().filter(StringUtils::isNotEmpty).collect(Collectors.toList());
csvStr.append(StringUtils.join(headerData, ","));
csvStr.append("\n");
// 读取表体数据
for (int i = 1; i < dataList.size(); i++) {
Map dataItem = dataList.get(i);
List dataItemHandled = dataItem.values().stream().filter(StringUtils::isNotEmpty).collect(Collectors.toList());
csvStr.append(StringUtils.join(dataItemHandled, ","));
csvStr.append("\n");
}
return csvStr.toString();
}
}
```
### 转化用户请求内容
```java
// 将用户请求内容转换为对AI请求的统一请求格式
private String buildAskStr(String goal, String chartType, String chartData) {
StringBuilder askStr = new StringBuilder();
askStr.append("分析需求：");
askStr.append("\n");
askStr.append(goal);
if (StringUtils.isNotBlank(chartType)) {
askStr.append(", 请使用");
askStr.append(chartType);
}
askStr.append("\n");
askStr.append("原始数据：");
askStr.append("\n");
askStr.append(chartData);
return askStr.toString();
}
```
### 同步请求AI接口
```java
/**
* 调用鱼聪明AI接口并取得返回结果
*/
@Service
public class YuAIManager {
@Autowired
private YuCongMingClient yuCongMingClient;
public String requestAIWithParams(Long id, String askStr) {
DevChatRequest devChatRequest = new DevChatRequest();
devChatRequest.setMessage(askStr);
devChatRequest.setModelId(id);
BaseResponse responseMessage = yuCongMingClient.doChat(devChatRequest);
if (responseMessage == null || responseMessage.getData() == null) {
throw new BusinessException(ErrorCode.SYSTEM_ERROR, "请求AI接口失败");
}
return responseMessage.getData().getContent();
}
}
```
## 第四期 - 智能分析优化（线程池）
### 解决问题
1. 程序调用AI接口获得响应速度较慢，同步的请求方式会让用户发送请求后一直阻塞在当前页面，无法进行其他操作而只能等待接口返回响应结果
2. 提高系统任务量的吞吐量，提高系统处理任务的效率
### 校验用户上传文件
对于用户来说，网络资源是相对便宜的。而对于我们的服务器来说是十分昂贵的。因此我们有必要限制用户上传文件的大小以及校验文件的格式，防止用户恶意攻击服务器。
```java
/**
* 校验用户上传的文件格式是否正确
*/
private boolean checkFileStyle(MultipartFile multipartFile) {
String originalFilename = multipartFile.getOriginalFilename();
// 文件名校验
String fileSuffix = FileNameUtil.getSuffix(originalFilename);
if (!FileConstant.ANALYSIS_FILE_SUFFIXES.contains(fileSuffix)) {
return false;
}
return multipartFile.getSize() <= FileConstant.ANALYSIS_FILE_SIZE;
}
```
### 限流用户请求
用户对于图表API接口的调用对我们服务后台来说是有一定成本的，因此有必要限制用户的请求频率，防止用户恶刷造成自己破产😔。
```java
@Configuration
@ConfigurationProperties(prefix = "spring.redis") // 添加这个住节后，程序会将配置项中的值加载到当前类的属性中
@Data
public class RedissonConfig {
private String host; // redis服务器的主机地址
private Integer database; // 使用Redis中的那个数据库
private String port; // redis所在主机上服务的端口号
@Bean
public RedissonClient redissonClient() {
Config config = new Config();
config.useSingleServer()
.setDatabase(database)
.setAddress("redis://" + host + ":" + port);
return Redisson.create(config);
}
}
```
```java
// 图标接口完整代码
@RequestMapping(value = "/file", method = RequestMethod.POST)
public BaseResponse aiAnalysisToGenChartData(@RequestPart("file") MultipartFile multipartFile,
ChartRequest chartRequest,
HttpServletRequest request) {
String goal = chartRequest.getGoal();
String name = chartRequest.getName();
if (name == null) {
name = "<未设置图表名称>";
}
String chartType = chartRequest.getChartType();
if (chartType == null) {
chartType = "<未设置图表类型>";
}
if (multipartFile == null || multipartFile.getSize() <= 0 || StringUtils.isBlank(goal)) {
throw new BusinessException(ErrorCode.PARAMS_ERROR);
}
if (!this.checkFileStyle(multipartFile)) {
throw new BusinessException(ErrorCode.PARAMS_ERROR, "文件格式错误或大小超限(10MB)");
}
if (name.length() > 100) {
throw new BusinessException(ErrorCode.PARAMS_ERROR, "指定图表名称过长");
}
UserVO user = (UserVO) request.getSession(false).getAttribute(UserConstant.LOGIN_STATE_KEY);
// 对用户的请求进行限流
redissonManager.doRate(String.valueOf(user.getId())); // 针对用户id分发限流器
// 将excel文件转换为字符串
String csvStr = ExcelHandleUtils.handleExcelToStr(multipartFile);
String askStr = this.buildAskStr(goal, chartType, csvStr);
// 调用AI接口并取得返回结果
String responseMessage = yuAIManager.requestAIWithParams(AI_ID, askStr);
// 对相应数据进行处理
Map responseMap = this.processResponseData(responseMessage);
// 保存图表数据
Chart chart = new Chart();
chart.setName(name);
chart.setGoal(goal);
chart.setChartType(chartType);
chart.setChartData(csvStr);
chart.setGenChartData(responseMap.get("genChartData"));
chart.setGenChartResult(responseMap.get("genChartResult"));
chart.setUserId(user.getId());
if (!chartService.save(chart)) {
throw new BusinessException(ErrorCode.SYSTEM_ERROR, "图表数据保存失败");
}
// 返回生成的图表数据分析结果
ChartVO chartVO = new ChartVO();
BeanUtils.copyProperties(chart, chartVO);
return ResultUtils.success(chartVO);
}
```
### 异步请求AI接口（计划接入讯飞星火）
#### 异步优化
##### 异步优化解决问题
- 解决后台调用AI接口的响应时间较长，前台用户提交任务后阻塞等待的业务场景，优化用户体验
##### 异步优化的业务流程
1. 创建线程池实例用来异步处理用户图表分析请求
2. 接收到图表分析请求后，现将用户上传的图表信息保存到数据库中，将状态设置为waiting扥等待处理状态
3. 当线程池中的线程开始处理用户的图表请求时，将图表信息表中的该条图表状态修改为running处理中
4. 当线程池中的线程即将此次图表分析请求处理结束后，更新图表信息表中的状态为succeed完成状态
5. 当处理此次用户图表分析请求过程中出现错误时，更新图表状态为failed失败状态，并追加失败原因到该条数据记录的字段中。
#### 线程池异步
```java
@Configuration
public class ThreadPoolExecutorConfig {
private static final int CORE_POOL_SIZE = 2; // 核心线程数
private static final int MAXIMUM_POOL_SIZE = 4; // 最大线程数
private static final long KEEP_ALIVE_TIME = 10; // 非核心线程数存活时间
private static final TimeUnit TIMEUNIT = TimeUnit.SECONDS; // 非核心线程数存活时间单位
private static final ArrayBlockingQueue ARRAY_BLOCKING_QUEUE = new ArrayBlockingQueue<>(20); // 阻塞队列
private static final RejectedExecutionHandler REJECTED_EXECUTION_HANDLER = new EsAbortPolicy(); // 阻塞队列满后的拒绝策略
// 线程创建是经过的线程池工厂
private static final ThreadFactory THREAD_FACTORY = new ThreadFactory() {
private int threadNum = 1;
@Override
public Thread newThread(@NotNull Runnable r) {
Thread thread = new Thread(r);
thread.setName("线程-" + threadNum);
threadNum++;
return thread;
}
};
@Bean
public ThreadPoolExecutor threadPoolExecutor() {
return new ThreadPoolExecutor(CORE_POOL_SIZE,
MAXIMUM_POOL_SIZE,
KEEP_ALIVE_TIME,
TIMEUNIT,
ARRAY_BLOCKING_QUEUE,
THREAD_FACTORY,
REJECTED_EXECUTION_HANDLER);
}
}
```
#### 图表信息表字段添加
```sql
`status` int(11) NOT NULL DEFAULT '0' COMMENT '图表处理状态',
`execMessage` varchar(1024) COLLATE utf8mb4_unicode_ci DEFAULT NULL COMMENT '图表状态描述'
```
#### 异步处理图表分析请求
```java
CompletableFuture.runAsync(new Runnable() {
@Override
public void run() {
// 先更新图表分析表中的数据为running-1处理中状态
chart.setStatus(ChartStatusEnum.RUNNING.getStatus());
chartService.updateById(chart);
String responseMessage = null;
Map responseMap = null;
// try {
// 调用AI接口并取得返回结果
try {
responseMessage = yuAIManager.requestAIWithParams(AI_ID, askStr);
} catch (Exception e) {
log.error("[" + LocalDateTime.now() + "]调用AI请求分析图表出错,图表数据：" + chart);
// 更新图表信息表中的图表状态为2-failed
chart.setStatus(ChartStatusEnum.FAILED.getStatus());
setExecMessage(chart, "分析图表数据失败，请尝试重新提交（-1）");
return;
}
// 对相应数据进行处理
try {
responseMap = processResponseData(responseMessage);
} catch (Exception e) {
log.error("[" + LocalDateTime.now() + "]AI响应数据格式：" + chart);
// 更新图表信息表中的图表状态为2-failed
chart.setStatus(ChartStatusEnum.FAILED.getStatus());
setExecMessage(chart, "分析图表数据失败，请尝试重新提交（-2）");
return;
}
// 保存图表数据
chart.setStatus(ChartStatusEnum.SUCCEED.getStatus());
chart.setGenChartData(responseMap.get("genChartData"));
chart.setGenChartResult(responseMap.get("genChartResult"));
chart.setStatus(ChartStatusEnum.SUCCEED.getStatus());
chartService.updateById(chart);
}
}, threadPoolExecutor);
```
## 第四期 - 智能分析优化（Ratelimiter分布式限流）
### 解决问题
## 第五期 - 智能分析优化（RabbitMQ）
### 解决问题
1. 服务器宕机，导致线程池正在执行的任务和阻塞队列中的任务丢失
2. 基于线程池实现的异步只对单机架构项目生效，如果后期项目分布式部署，每个分布式项目都维护一个线程池，则会导致异步处理任务量的上限增加，无法精准把控AI服务的服务量上限
### Spring Boot - AMQP实现
```xml
org.springframework.boot
spring-boot-starter-amqp
2.7.2
```
#### Producer
```java
@Slf4j
@Component
public class BIChartMessageMQProducer {
@Autowired
private RabbitTemplate rabbitTemplate; // SpringBoot RabbitMQ Template
public void sendMessage(String message) {
rabbitTemplate.convertAndSend(BIMQConstant.BI_MQ_EXCHANGE_NAME, BIMQConstant.BI_ROUTING_KEY, message);
}
}
```
#### Consumer
```java
@Component
@Slf4j
public class BIChartMessageMQConsumer {
@Autowired
private ChartService chartService;
@Autowired
private XunFeiAPIManager xunFeiAPIManager;
@RabbitListener(queues = {BIMQConstant.BI_QUEUE_NAME}, ackMode = "MANUAL")
public void consumeMessage(String message, Channel channel, @Header(AmqpHeaders.DELIVERY_TAG) long deliverTag) throws IOException {
if (StringUtils.isBlank(message)) {
channel.basicAck(deliverTag, Boolean.FALSE);
log.error("MQ-Consumer-从消息队列中获取到的消息为blank，MQMessage:{}", deliverTag);
throw new BusinessException(ErrorCode.SYSTEM_ERROR, "从消息队列中获取到的消息为blank");
}
long chartId = Long.parseLong(message);
// 根据消息队列拿到的消息从数据库中查询该条消息的信息
Chart chartOriginal = chartService.getById(chartId); // 用户上传的图表数据
if (chartOriginal == null) {
log.error("MQ-Consumer-根据消息tag：" + deliverTag + " 无法查询到对应的图表信息");
channel.basicAck(deliverTag, Boolean.FALSE);
throw new BusinessException(ErrorCode.SYSTEM_ERROR, "MQ-Consumer-根据消息tag：" + deliverTag + " 无法查询到对应的图表信息");
}
String goal = chartOriginal.getGoal(); // 用户上传的分析目标
String chartType = chartOriginal.getChartType(); // 用户上传的图表类型
String chartData = chartOriginal.getChartData(); // 用户上传的处理后的图表数据（csv）
String askStr = chartService.buildXFAskStr(goal, chartType, chartData); // 构建AI请求语句
// 先更新图表分析表中的数据为running-1处理中状态
chartOriginal.setStatus(ChartStatusEnum.RUNNING.getStatus());
if (!chartService.updateById(chartOriginal)) {
log.error("【" + LocalDateTime.now() + "】error-数据库服务异常-保存chart" + BIChartMessageMQConsumer.class);
throw new BusinessException(ErrorCode.SYSTEM_ERROR, "服务异常，正在加急修复中...");
}
String responseMessage = null;
Map responseMap = null;
// 调用AI接口并取得返回结果
try {
responseMessage = xunFeiAPIManager.ask(askStr);
} catch (Exception e) {
log.error("[" + LocalDateTime.now() + "]调用AI请求分析图表出错,图表数据：" + chartOriginal);
// 更新图表信息表中的图表状态为2-failed
chartOriginal.setStatus(ChartStatusEnum.FAILED.getStatus());
chartService.setExecMessage(chartOriginal, "分析图表数据失败，请尝试重新提交（-1）");
return;
}
// 对相应数据进行处理
try {
responseMap = chartService.processXFResponseData(responseMessage);
} catch (Exception e) {
log.error("[" + LocalDateTime.now() + "]AI响应数据格式异常：" + chartOriginal);
// 更新图表信息表中的图表状态为2-failed
chartOriginal.setStatus(ChartStatusEnum.FAILED.getStatus());
chartService.setExecMessage(chartOriginal, "分析图表数据失败，请尝试重新提交（-2）");
return;
}
if (responseMap == null) {
log.error("【" + LocalDateTime.now() + "】error-AI接口响应数据格式错误" + BIChartMessageMQConsumer.class);
chartOriginal.setStatus(ChartStatusEnum.FAILED.getStatus());
chartService.setExecMessage(chartOriginal, "分析图标数据失败，请尝试重新提交（-3）");
return;
}
// 保存图表数据
chartOriginal.setStatus(ChartStatusEnum.SUCCEED.getStatus());
chartOriginal.setGenChartData(responseMap.get("genChartData"));
chartOriginal.setGenChartResult(responseMap.get("genChartResult"));
if (!chartService.updateById(chartOriginal)) {
log.error("AI生成的图表数据保存失败！数据库服务异常。{}", chartOriginal);
throw new BusinessException(ErrorCode.SYSTEM_ERROR, "服务异常，正在快马加鞭修复中...");
}
}
}
```
#### 任务推送消息队列
```java
/**
* 异步处理用户请求接口（消息队列 + 调用讯飞大模型接口）
*/
@RequestMapping(value = "/file/async/mq", method = RequestMethod.POST)
public BaseResponse aiAnalysisToGenChartDataAsyncMQ(@RequestPart("file") MultipartFile multipartFile,
ChartRequest chartRequest,
HttpServletRequest request) {
String goal = chartRequest.getGoal();
String name = chartRequest.getName();
if (name == null) {
name = "<未设置图表名称>";
}
String chartType = chartRequest.getChartType();
if (chartType == null) {
chartType = "<未设置图表类型>";
}
if (multipartFile == null || multipartFile.getSize() <= 0 || StringUtils.isBlank(goal)) {
throw new BusinessException(ErrorCode.PARAMS_ERROR);
}
if (!this.checkFileStyle(multipartFile)) {
throw new BusinessException(ErrorCode.PARAMS_ERROR, "文件格式错误或大小超限(" + FileConstant.ANALYSIS_FILE_SIZE + "MB)");
}
if (name.length() > 100) {
throw new BusinessException(ErrorCode.PARAMS_ERROR, "指定图表名称过长");
}
UserVO user = (UserVO) request.getSession(false).getAttribute(UserConstant.LOGIN_STATE_KEY);
// 对用户的请求进行限流
redissonManager.doRate(String.valueOf(user.getId())); // 针对用户id分发限流器
Chart chart = new Chart();
chart.setUserId(user.getId());
chart.setName(name);
chart.setGoal(goal);
chart.setChartType(chartType);
// 将excel文件转换为csv字符串
String csvStr = ExcelHandleUtils.handleExcelToStr(multipartFile);
chart.setChartData(csvStr);
chart.setStatus(ChartStatusEnum.WAITING.getStatus());
// 将用户的分析图表请求保存到数据库中，并设置状态为等待中-waiting-0
if (!chartService.save(chart)) {
log.error("【" + LocalDateTime.now() + "】error-数据库服务异常-保存chart" + ChartController.class);
throw new BusinessException(ErrorCode.SYSTEM_ERROR, "服务异常，正在加急修复中...");
}
// 将消息编码传送进消息队列
Long chartId = chart.getId();
messageMQProducer.sendMessage(String.valueOf(chartId));
// 返回用户上柴男图表的id登原始信息
ChartVO chartVO = new ChartVO();
BeanUtils.copyProperties(chart, chartVO);
return ResultUtils.success(chartVO);
}
```