分布式微服务系统架构第125集：AI大模型
加群联系作者vx：xiaoda0423
仓库地址：https://webvueblog.github.io/JavaPlusDoc/
https://1024bat.cn/
一、user 表（用户表）
sql
复制编辑
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 '用户头像',
userProfile varchar(512) null comment '用户简介',
userRole varchar(256) default 'user' not null comment '用户角色：user/admin/ban',
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_userAccount (userAccount)
) comment '用户' collate = utf8mb4_unicode_ci;列名 | 类型 | 约束 / 默认值 | 说明 |
|---|---|---|---|
id | bigint | auto_increment，主键 | 自增唯一 ID |
userAccount | varchar(256) | NOT NULL | 登录账号（如手机号、邮箱或自定义账号） |
userPassword | varchar(512) | NOT NULL | 存储密码（建议储存 bcrypt 等加密后的结果） |
userName | varchar(256) | NULL | 用户昵称 |
userAvatar | varchar(1024) | NULL | 用户头像 URL |
userProfile | varchar(512) | NULL | 个人简介 |
userRole | varchar(256) | NOT NULL DEFAULT 'user' | 角色： |
createTime | datetime | NOT NULL DEFAULT CURRENT_TIMESTAMP | 记录创建时间 |
updateTime | datetime | NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP | 最后更新时间 |
isDelete | tinyint | NOT NULL DEFAULT 0 | 软删除标志： |
| 索引 | idx_userAccount(userAccount) | 登录查询或唯一校验可加唯一索引 | |
| 表注释 | collate = utf8mb4_unicode_ci | 支持 4 字节 UTF-8（表情等） |
二、chat 表（图表/会话信息表）
sql
复制编辑
create table if not exists chat
(
id bigint auto_increment comment 'id' primary key,
goal text null comment '分析目标',
`name` varchar(128) null comment '图表名称',
chartData text null comment '图表数据',
chartType varchar(128) null comment '图表类型',
userId bigint null comment '用户id',
genChart text null comment '生成的图表',
genResult text null comment '生成的数据',
status varchar(128) not null default 'wait' comment 'wait,running,succeed,failed',
execMessage text null comment '执行信息',
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;列名 | 类型 | 约束 / 默认值 | 说明 |
|---|---|---|---|
id | bigint | auto_increment，主键 | 自增唯一 ID |
goal | text | NULL | 分析目标，通常是用户输入的需求文本 |
name | varchar(128) | NULL | 图表/会话名称 |
chartData | text | NULL | 输入的原始图表数据（JSON 等） |
chartType | varchar(128) | NULL | 图表类型，如 |
userId | bigint | NULL | 关联 |
genChart | text | NULL | AI 或后端生成的图表（如图表配置 JSON） |
genResult | text | NULL | AI 或后端生成的结果数据 |
status | varchar(128) | NOT NULL DEFAULT 'wait' | 任务状态： |
execMessage | text | NULL | 执行日志或错误信息 |
createTime | datetime | NOT NULL DEFAULT CURRENT_TIMESTAMP | 记录创建时间 |
updateTime | datetime | NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP | 最后更新时间 |
isDelete | tinyint | NOT NULL DEFAULT 0 | 软删除标志： |
| 表注释 | collate = utf8mb4_unicode_ci | 支持 4 字节 UTF-8 |
三、建议 & 说明
外键约束
建议在
chat.userId上增加外键：alter table chat add constraint fk_chat_user foreign key(userId) references user(id);
唯一索引
如果
userAccount需要唯一，改成unique key。
状态字段
status建议改为enum('wait','running','succeed','failed')，以节省存储并保证值域。
分表 / 分库
若
chat表数据量巨大，可考虑按时间或用户分表。
软删除
isDelete字段已支持软删除，查询时请加where isDelete=0。
字符集
utf8mb4_unicode_ci可存储 Emoji 等 4 字节字符，适合昵称、头像 URL 等。
mongo，cassandra，elasticsearch，redis，clickhouse，mysql，postgresql集群，扩展，多节点分布
1. MongoDB
复制集（Replica Set）
一主多从（Primary／Secondary），自动故障切换。
从节点可做读分担读流量。
分片集群（Sharded Cluster）
数据依据 shard key 切分到多个分片（Shard）；每个分片内部为复制集。
Config Servers 存储元数据；Mongos 负责路由查询。
扩展
增加分片实现水平扩展；增添 Secondary 提升读可用。
Online Resharding（4.4+）支持在线调整分片键。
部署建议：至少 3 个 Config Server，偶数分片时用仲裁节点避免 split-brain。
2. Cassandra
对等节点（Peer-to-Peer）
每个节点都相同，无主从。
基于 Gossip 协议 交换拓扑，一致性哈希环 分配数据。
复制因子（RF） & 一致性级别
RF 控制每条数据复制到多少节点；读写可配置 ONE/QUORUM/ALL 等。
扩展
横向自动扩容：加入新节点，数据自动重分布（Rebalancing）。
部署建议：跨机房多机架部署，保证 RF 的多可用区冗余。
3. Elasticsearch
节点类型：Master、Data、Ingest、Coordinating（Client）
索引与分片
每个索引可设置 N 个 Primary Shard + M 个 Replica Shard。
Primary/Replica 分布在不同 Data 节点。
集群协调
Master 负责元数据管理；Coordinating 节点负责聚合请求。
扩展
增加 Data 节点后，可通过
cluster.routing.allocation平衡分片。增加 Master 候选节点增强稳定性（奇数个，>=3）。
部署建议：Master 节点隔离，Client 节点对外负载均衡。
4. Redis
主从复制 + 哨兵（Sentinel）
一主多从，Sentinel 监控故障并自动 failover。
集群模式（Cluster）
数据分为 16384 个槽（hash slots），分布到不同主节点；每个主节点可有从节点。
客户端自动感知重定向（MOVED、ASK）。
扩展
增加主节点提升写吞吐；增加从节点提升读吞吐。
部署建议：至少 3 个 Sentinel，保证 Quorum；Cluster 模式至少 3 主 3 从。
5. ClickHouse
Shard + Replica
每个表在多个分片（Shard）上水平分区；每个分片可有多个副本（Replica）。
Table Engines
ReplicatedMergeTree系列支持复制和副本间自动同步。
扩展
增加分片：编辑集群配置文件，重新部署新分片节点并重平衡。
部署建议：使用 ZooKeeper 管理分布式元数据、跨机房部署副本。
6. MySQL
主从复制（Replication）
一主多从，基于 binlog。读写分离提升读性能。
组复制 / InnoDB Cluster
Group Replication 实现多主（Multi-Primary）或单主模式，自动仲裁。
分库分表
Sharding Proxy（如 ShardingSphere）、Vitess、ProxySQL 等中间件支持水平拆分。
扩展
增加只读从实例；使用 Proxy 做负载均衡；使用分片中间件做写扩展。
部署建议：异地多机房部署 GTID 复制，结合 ProxySQL 做读写分离。
主从复制（Streaming Replication）
物理或逻辑复制；热备从库用于故障切换或读扩展。
Patroni / Pgpool-II / PgBouncer
Patroni + Etcd/ZooKeeper 管理高可用；Pgpool-II 做负载均衡和故障切换；PgBouncer 做连接池。
分布式扩展
Citus 扩展：将表分布到 worker 节点，实现分片和并行查询。
扩展
增加从节点做读扩展；使用 Citus 或 Citus Cloud 做写扩展。
部署建议：至少 3 个 Etcd/Consul 实例+3 个 Patroni 主体节点，保证高可用。
通用最佳实践
奇数节点：无论是仲裁节点还是选主节点，保持奇数个以达成多数派。
跨机架/机房部署：避免单点故障，保证副本分布在不同物理位置。
监控与告警：部署 Prometheus + Grafana 或 ELK、Graylog 等监控各节点健康与性能。
备份与恢复：定期全量备份（快照）+增量备份，确保可快速恢复。
自动化运维：使用 Ansible、Terraform、Kubernetes Operators（如 MongoDB Operator、Etcd Operator）等工具管理集群生命周期。
http://localhost:8108/api/doc.html#/home
1️⃣
@Target(ElementType.METHOD)意思：这个注解只能加在“方法”上。
举例：
@AuthCheck(mustRole = "admin") public void deleteUser() { ... }✅ 可以用在方法上
❌ 不可以用在类、字段、构造器等位置
2️⃣
@Retention(RetentionPolicy.RUNTIME)意思：这个注解在程序运行的时候依然“存在”，并可以通过反射读取它的内容。
说明：你可以在运行时动态读取
mustRole的值，来做权限校验。
3️⃣
public @interface AuthCheck { ... }定义一个叫
AuthCheck的注解。你就可以像使用@Override那样使用它。
4️⃣
String mustRole() default ""这是一个 注解的参数，叫
mustRole，类型是String，表示“必须的角色”。default ""表示默认值是空字符串。举例：
@AuthCheck(mustRole = "admin") // 必须是 admin 角色 @AuthCheck() // 不传参数，相当于默认空字符串
📌 使用场景示例（配合 AOP 做权限控制）
比如你定义了以下方法：
@AuthCheck(mustRole = "admin") public void deleteUserAccount() { // 删除账号逻辑 }你可以在后台通过 AOP 切面读取这个注解的值，判断当前登录用户是否是
admin，否则拒绝执行这个方法。🔹
@Aspect+@Component@Aspect @Component@Aspect：标识这是一个 切面类，Spring AOP 会识别它。@Component：让这个类变成 Spring 管理的 Bean，才能自动注入和工作。
🔹 拦截逻辑的主函数
@Around("@annotation(authCheck)")@Around：表示“环绕通知”，就是在目标方法执行 之前和之后都能插手干预。@annotation(authCheck)：只拦截打了@AuthCheck注解的方法。authCheck就是注解的实例。
🔹 获取注解中的角色要求
String mustRole = authCheck.mustRole();如果你写的是：
@AuthCheck(mustRole = "admin")那
mustRole就是"admin"。🔹 获取当前请求与用户
HttpServletRequest request = ((ServletRequestAttributes) requestAttributes).getRequest(); User loginUser = userService.getLoginUser(request);拿到当前请求对象，然后调用
userService拿出登录用户。@Component：把这个类交给 Spring 管理，后续可以自动注入使用。BiMessageProducer：就是“发送消息的人”。
假设你在一个点餐系统中：
用户点单后，你需要把这个订单数据发到“厨房系统”；
那么你可以：
biMessageProducer.sendMessage("订单：奶茶1杯");然后 RabbitMQ 就会“帮你传话”，把这条消息送给对应的消费者。
注入两个服务类：
AiManager：调用 AI 分析图表。ChartService：查和更新图表信息。
@RabbitListener( bindings = @QueueBinding( value = @Queue(name = BiMqConstant.BI_QUEUE_NAME), exchange = @Exchange(name = BiMqConstant.BI_EXCHANGE_NAME, type = ExchangeTypes.DIRECT), key = BiMqConstant.BI_ROUTING_KEY ), ackMode = "MANUAL" )监听器的配置，告诉 RabbitMQ：
哪个队列（
BI_QUEUE_NAME）；哪个交换机（
BI_EXCHANGE_NAME）；用什么路由键（
BI_ROUTING_KEY）；
ackMode = "MANUAL"表示：消息手动确认，不确认就不会从队列移除。
📨 方法参数说明（很关键）
private void receiveMessage(String message, Channel channel, @Header(AmqpHeaders.DELIVERY_TAG) long deliveryTag)message：收到的消息内容（这里是图表 ID 的字符串）。channel：RabbitMQ 通信通道，用于确认/拒绝消息。deliveryTag：消息唯一编号，确认时要用。
/** * 分析目标 */ private String goal;/** * 图表类型 */ private String chartType;/** * 图表数据 */ private String chartData;🧠 拼接 AI
StringBuffer userInput = new StringBuffer(); userInput.append("你是一个数据分析师...").append("分析目标:").append(chart.getGoal())...把 chart 里的信息拼成 AI 要求的输入格式：
分析目标
图表类型（如柱状图）
原始数据（CSV 格式）
🧠 调用 AI，生成结果
String result = aiManager.doChat(userInput.toString()); String[] splits = result.split("【【【【【");✅ 确认消息处理成功
channel.basicAck(deliveryTag, false);RabbitMQ 要求你手动告诉它“这条消息处理完了”，否则它会反复发。
一个流水线机器人：
等待 RabbitMQ 叫它干活；
拿到图表 ID；
去数据库拿详细信息；
拼成 AI 能看懂的输入；
让 AI 输出图表代码和结论；
写回数据库；
告诉 RabbitMQ：我干完了！
public enum ErrorCode { // ========== 成功 ========== SUCCESS(0, "成功", "操作成功"), // ========== 参数错误 ========== PARAMS_ERROR(40000, "请求参数错误", "请求参数不符合预期"), VALIDATION_ERROR(40001, "参数校验失败", "字段格式或约束错误"), // ========== 认证 & 授权 ========== NOT_LOGIN_ERROR(40100, "未登录", "用户未登录，请先登录"), NO_AUTH_ERROR(40101, "无权限", "当前用户无访问权限"), // ========== 资源访问错误 ========== FORBIDDEN_ERROR(40300, "禁止访问", "服务器理解请求但拒绝执行"), NOT_FOUND_ERROR(40400, "资源不存在", "请求的数据或页面不存在"), // ========== 系统错误 ========== SYSTEM_ERROR(50000, "系统错误", "服务器内部发生错误，请联系管理员"), OPERATION_ERROR(50001, "操作失败", "请求的操作未成功执行"), SERVICE_UNAVAILABLE(50300, "服务不可用", "系统维护中，请稍后再试"), // ========== 频率限制 ========== TOO_MANY_REQUEST(42900, "请求频繁", "请求过于频繁，请稍后重试"), // ========== 业务逻辑错误 ========== DUPLICATE_OPERATION(60001, "重复操作", "请勿重复提交"), DATA_CONFLICT(60002, "数据冲突", "存在重复或冲突的数据"), DATA_NOT_READY(60003, "数据未就绪", "请稍后重试，数据正在处理中"); private final int code; private final String message; private final String description; ErrorCode(int code, String message, String description) { this.code = code; this.message = message; this.description = description; } public int getCode() { return code; } public String getMessage() { return message; } public String getDescription() { return description; } }CORS 跨域配置类，用于让前端页面可以访问后端接口，即使它们的域名或端口不同。
✅ 什么是 CORS？
CORS 全称是 Cross-Origin Resource Sharing（跨域资源共享） 。
默认情况下，浏览器不允许前端去请求一个不同域名/端口/协议的接口，需要配置后端允许“跨域访问”。比如：
前端在
http://localhost:3000后端在
http://localhost:8080
👉 这就叫“跨域”@Configuration：告诉 Spring 这是一个配置类（类似全局设置）。WebMvcConfigurer：这是 Spring Web 的一个接口，你可以用它来自定义 Web 行为（比如 CORS、静态资源、拦截器等）。
.allowCredentials(true) 允许前端携带 Cookie、token 等凭证。
⚠️ 注意：如果允许携带 Cookie，就不能用 allowedOrigins("")，要用 allowedOriginPatterns("")。
@Configuration：这个类是一个 配置类，会被 Spring 自动加载。@MapperScan("com.sss.bibackend.mapper")：告诉 MyBatis 去扫描你指定的包下的
Mapper接口（也叫 DAO 层）。例如你有一个接口
UserMapper放在com.sss.bibackend.mapper里，这里就会自动注册它。
📌 Bonus：分页查询怎么用？
你写完这个配置后，可以这样分页查询：
Pagepage = new Page<>(1, 5); // 第1页，每页5条 Page resultPage = userMapper.selectPage(page, null); List records = resultPage.getRecords(); // 查到的数据 long total = resultPage.getTotal(); // 总条数 我们有一个
User表，字段如下：id：主键username：用户名email：邮箱age：年龄
1️⃣ 数据库建表语句（MySQL）
CREATE TABLE user ( id BIGINT PRIMARY KEY AUTO_INCREMENT, username VARCHAR(50), email VARCHAR(100), age INT );2️⃣ 实体类
User.java@Data @TableName("user") public class User { private Long id; private String username; private String email; private Integer age; }3️⃣ Mapper 接口
UserMapper.javapublic interface UserMapper extends BaseMapper{ // 无需添加任何代码，MyBatis-Plus 会自动实现分页功能 } 4️⃣ Service 示例
UserService.javapublic interface UserService extends IService{ Page searchUsers(String username, Integer minAge, int current, int size); } 实现类
UserServiceImpl.java@Service public class UserServiceImpl extends ServiceImplsearchUsers(String username, Integer minAge, int current, int size) { QueryWrapper queryWrapper = new QueryWrapper<>(); // 如果用户名不为空，模糊查询 if (StringUtils.isNotBlank(username)) { queryWrapper.like("username", username); } // 如果设置了最小年龄，做大于等于筛选 if (minAge != null) { queryWrapper.ge("age", minAge); } Page page = new Page<>(current, size); return this.page(page, queryWrapper); // 调用 MyBatis-Plus 分页查询 } } 5️⃣ Controller 示例
UserController.java@RestController @RequestMapping("/user") public class UserController { @Resource private UserService userService; @GetMapping("/search") public PagesearchUsers( @RequestParam(required = false) String username, @RequestParam(required = false) Integer minAge, @RequestParam(defaultValue = "1") int page, @RequestParam(defaultValue = "10") int size) { return userService.searchUsers(username, minAge, page, size); } } @Configuration：告诉 Spring Boot 这是一个配置类，要自动加载。@ConfigurationProperties(prefix = "threadpool")：表示你可以在配置文件（如
application.yml）里写threadpool.xxx的配置，自动绑定到这个类的字段上。
@Data：Lombok 提供的注解，自动生成 getter/setter、toString 等。@Bean：把这个线程池交给 Spring 管理，以后可以通过注入@Autowired来使用它。Runnable r：每个要执行的任务（也可以看成一个函数）。
newThread(r) ：线程池内部需要新线程时就会调用这个方法。
设置线程名字为：
线程1、线程2……，方便日志调试。
🚀 创建线程池本体
ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor( corePoolSize, // 核心线程数 maximumPoolSize, // 最大线程数 keepAliveTime, // 非核心线程的存活时间 TimeUnit.MILLISECONDS, // 时间单位 new ArrayBlockingQueue<>(4), // 队列容量 4 的阻塞队列 threadFactory // 上面自定义的线程工厂 );线程池怎么工作的？
核心线程数（比如 2）：一直保留，即使空闲；
最大线程数（比如 4）：超过核心线程后再用，任务多了就扩容；
队列（大小为 4）：任务先进入队列排队；
自定义线程工厂：创建线程时用你定义的名字；
超出容量（线程 + 队列都满）时会报错（默认策略是
AbortPolicy）。
使用线程池的方法（举个例子）
@Autowired private ThreadPoolExecutor threadPoolExecutor; public void doSomethingAsync() { threadPoolExecutor.execute(() -> { // 异步任务代码 System.out.println("当前线程：" + Thread.currentThread().getName()); }); }public class BusinessException extends RuntimeException
这个类继承了 RuntimeException，说明它是一个运行时异常。
使用起来很方便，可以不写 try...catch 也能抛出。
同一个用户每秒最多请求 2 次
🧩 注入 Redisson 客户端
@Resource private RedissonClient redissonClient;RedissonClient是操作 Redis 的 Java 客户端，功能强大，支持分布式锁、限流器等；这里我们用的是它的限流器（RateLimiter）功能。
🧱 创建限流器
RRateLimiter rateLimiter = redissonClient.getRateLimiter(key); rateLimiter.trySetRate(RateType.OVERALL, 2, 1, RateIntervalUnit.SECONDS);getRateLimiter(key)：从 Redis 中拿一个名为key的限流器，如果不存在就创建；trySetRate(...)：RateType.OVERALL：总的限流（所有客户端共享这个限制）；2：允许每个周期（1 秒）最多执行 2 次；RateIntervalUnit.SECONDS：单位是秒。
✔️ 翻译成人话就是：
“我允许这个 key（比如 user:123）每秒最多只能干 2 次，再多就不行。”
你在 Controller 或 Service 中这样使用：
redisLimiterManager.doRateLimit("user:" + userId); // 如果没抛异常，就可以执行后续逻辑Redisson 限流 + 接口注解限流
🧱 最终实现效果：
@RateLimit(key = "user:limit", permitsPerSecond = 2, time = 1, unit = TimeUnit.SECONDS) @GetMapping("/test") public String test() { return "请求成功！"; }请求超过每秒 2 次 → 自动抛出限流异常！
星火大模型（Spark）发送用户消息
sparkRequest：后面会构建这个对象，封装用户要发给 AI 的内容；sparkClient：客户端对象，用来发起请求给大模型。
构建对话内容
Listmessages = new ArrayList<>(); messages.add(SparkMessage.systemContent("你是一个数据分析师...")); messages.add(SparkMessage.userContent(message)); 创建对话消息列表：
第一条是系统设定（告诉 AI 它扮演谁）；
第二条是用户真正输入的分析目标+数据。
数据万象文本审核调用腾讯云 COS（对象存储）进行文档内容合规检测
把用户上传的文件上传到腾讯云 COS，然后调用内容审核接口判断该文件是否包含违规内容（如暴力、涉黄等），如果违规就抛出异常。
📁 获取文件流并上传到 COS
inputStream = multipartFile.getInputStream(); ObjectMetadata objectMetadata = new ObjectMetadata(); String bucketName = "bi-1324629091"; String key = "folder/" + multipartFile.getOriginalFilename(); PutObjectRequest putObjectRequest = new PutObjectRequest(bucketName, key, inputStream, objectMetadata); PutObjectResult putObjectResult = cosClient.putObject(putObjectRequest);将
MultipartFile转成InputStream；构造上传请求
PutObjectRequest；key是文件路径 + 名称；上传文件到 COS（腾讯云对象存储）。
用户上传文件 → 上传到腾讯 COS → 发起内容审核 → 等待审核结果 → 如果不合格就抛异常，阻止后续逻辑执行。
