分布式微服务系统架构第125集：AI大模型

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：分页查询怎么用？

你写完这个配置后，可以这样分页查询：

Page<User> page = new Page<>(1, 5); // 第1页，每页5条
Page<User> resultPage = userMapper.selectPage(page, null);
List<User> 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.java`

public interface UserMapper extends BaseMapper<User> {
    // 无需添加任何代码，MyBatis-Plus 会自动实现分页功能
}

4️⃣ Service 示例 `UserService.java`

public interface UserService extends IService<User> {
    Page<User> searchUsers(String username, Integer minAge, int current, int size);
}

实现类 `UserServiceImpl.java`

@Service
public class UserServiceImpl extends ServiceImpl<UserMapper, User> implements UserService {

    @Override
    public Page<User> searchUsers(String username, Integer minAge, int current, int size) {
        QueryWrapper<User> queryWrapper = new QueryWrapper<>();

        // 如果用户名不为空，模糊查询
        if (StringUtils.isNotBlank(username)) {
            queryWrapper.like("username", username);
        }

        // 如果设置了最小年龄，做大于等于筛选
        if (minAge != null) {
            queryWrapper.ge("age", minAge);
        }

        Page<User> 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 Page<User> searchUsers(
            @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：客户端对象，用来发起请求给大模型。

构建对话内容

List<SparkMessage> messages = 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 → 发起内容审核 → 等待审核结果 → 如果不合格就抛异常，阻止后续逻辑执行。

秒客网

分布式微服务系统架构第125集：AI大模型

1. MongoDB

2. Cassandra

3. Elasticsearch

4. Redis

5. ClickHouse

6. MySQL

通用最佳实践

1️⃣ `@Target(ElementType.METHOD)`

2️⃣ `@Retention(RetentionPolicy.RUNTIME)`

3️⃣ `public @interface AuthCheck { ... }`

4️⃣ `String mustRole() default ""`

???? 使用场景示例（配合 AOP 做权限控制）

???? `@Aspect` + `@Component`

???? 拦截逻辑的主函数

???? 获取注解中的角色要求

???? 获取当前请求与用户

???? 方法参数说明（很关键）

???? 拼接 AI

???? 调用 AI，生成结果

✅ 确认消息处理成功

✅ 什么是 CORS？

???? Bonus：分页查询怎么用？

1️⃣ 数据库建表语句（MySQL）

2️⃣ 实体类 `User.java`

3️⃣ Mapper 接口 `UserMapper.java`

4️⃣ Service 示例 `UserService.java`

实现类 `UserServiceImpl.java`

5️⃣ Controller 示例 `UserController.java`

???? 创建线程池本体

使用线程池的方法（举个例子）

???? 注入 Redisson 客户端

???? 创建限流器

???? 最终实现效果：

构建对话内容

???? 获取文件流并上传到 COS

相关文章

分布式微服务系统架构第125集：AI大模型

1. MongoDB

2. Cassandra

3. Elasticsearch

4. Redis

5. ClickHouse

6. MySQL

通用最佳实践

1️⃣ @Target(ElementType.METHOD)

2️⃣ @Retention(RetentionPolicy.RUNTIME)

3️⃣ public @interface AuthCheck { ... }

4️⃣ String mustRole() default ""

???? 使用场景示例（配合 AOP 做权限控制）

???? @Aspect + @Component

???? 拦截逻辑的主函数

???? 获取注解中的角色要求

???? 获取当前请求与用户

???? 方法参数说明（很关键）

???? 拼接 AI

???? 调用 AI，生成结果

✅ 确认消息处理成功

✅ 什么是 CORS？

???? Bonus：分页查询怎么用？

1️⃣ 数据库建表语句（MySQL）

2️⃣ 实体类 User.java

3️⃣ Mapper 接口 UserMapper.java

4️⃣ Service 示例 UserService.java

实现类 UserServiceImpl.java

5️⃣ Controller 示例 UserController.java

???? 创建线程池本体

使用线程池的方法（举个例子）

???? 注入 Redisson 客户端

???? 创建限流器

???? 最终实现效果：

构建对话内容

???? 获取文件流并上传到 COS

相关文章

1️⃣ `@Target(ElementType.METHOD)`

2️⃣ `@Retention(RetentionPolicy.RUNTIME)`

3️⃣ `public @interface AuthCheck { ... }`

4️⃣ `String mustRole() default ""`

???? `@Aspect` + `@Component`

2️⃣ 实体类 `User.java`

3️⃣ Mapper 接口 `UserMapper.java`

4️⃣ Service 示例 `UserService.java`

实现类 `UserServiceImpl.java`

5️⃣ Controller 示例 `UserController.java`