智能BI总结

可以学习的点

1. 异步化

   由于调用三方API使用AI，需要等待一定的时间，为了优化用户体验，采用异步：

   1. 线程池：（单机推荐）
   2. 消息队列：（集群推荐）

2. 线程池：

   使用CompletableFuture和ThreadPoolExecutor自定义线程池来做

3. 消息队列:

   分布式的环境下（推荐），解耦代码，削峰（预防流量激增），异步化 考虑到第三方API会有限流的性质，不用想像线程池那样麻烦，自定义决绝策略和调用API失败的处理 在消息队列中可以统一为消费失败，消费失败的消息，可以再次消费

4. AI使用以及AIGC Prompt 优化

线程池的常见问题

• 线程池大小设置不合理：
  • 线程数过少会导致任务积压，线程数过多会导致资源浪费。
• 任务队列选择不当：
  • 使用无界队列可能导致内存溢出，使用有界队列可能导致任务被拒绝。
    • 无界队列是指任务队列的容量没有上限，可以无限添加任务。如果任务提交速度远大于任务处理速度，可能导致内存溢出（OOM），所以最好设置队列长度
      • LinkedBlockingQueue：默认无界，适合任务量不确定的场景。
      • PriorityBlockingQueue：基于优先级的无界队列。
      • SynchronousQueue：容量为 0 的特殊无界队列。
      • DelayQueue：支持延迟执行的无界队列。
      • LinkedTransferQueue：支持高效传递任务的无界队列。
      • ConcurrentLinkedQueue：无界非阻塞队列。
• 线程泄漏：
  • 线程池中的线程未正确释放，导致线程数不断增加。
• 死锁：
  • 任务之间相互等待，导致线程池无法继续执行任务。

线程池的最佳实践

• 合理设置线程池大小：
  • 对于 CPU 密集型任务，线程数设置为 CPU 核心数 + 1。
  • 对于 I/O 密集型任务，线程数设置为 CPU 核心数 * (1 + 平均等待时间 / 平均计算时间)。
• 使用合适的任务队列：
  • 对于短任务，可以使用 SynchronousQueue。
  • 对于长任务，可以使用 LinkedBlockingQueue 或 ArrayBlockingQueue。
• 设置合理的拒绝策略：
  • 根据业务需求选择合适的拒绝策略，避免任务丢失或系统崩溃。
• 监控和调优线程池：
  • 定期监控线程池状态，动态调整参数。

线程池的拒绝策略

当任务队列已满且线程数达到最大线程数时，线程池会触发拒绝策略。常见的拒绝策略包括：

• AbortPolicy：直接抛出 RejectedExecutionException。
• CallerRunsPolicy：由提交任务的线程执行该任务。
• DiscardPolicy：直接丢弃任务，不抛出异常。
• DiscardOldestPolicy：丢弃队列中最旧的任务，然后重新提交新任务。
• 还可以自定义拒绝策略:例如重要的数据存入数据库或redis等，后去在取出来

线程池的监控与调优

• 监控线程池状态：
  • 使用 ThreadPoolExecutor 提供的方法监控线程池状态，如：
    • getPoolSize()：当前线程数。
    • getActiveCount()：活跃线程数。
    • getQueue().size()：任务队列大小。
    • getCompletedTaskCount()：已完成任务数。
• 动态调整线程池参数：
  • 根据系统负载动态调整核心线程数、最大线程数等参数。

消息队列的基本概念

消息队列是一种应用程序之间的通信方式，消息生产者将消息发送到队列中，消息消费者从队列中获取消息并处理。

消息队列的作用：

• 解耦：生产者和消费者不需要直接通信，通过消息队列解耦。
• 异步：生产者发送消息后无需等待消费者处理，提高系统响应速度。
• 削峰填谷：通过队列缓冲消息，避免系统过载。
• 可靠性：消息队列通常支持消息持久化，确保消息不丢失。

消息队列的工作模式

• 点对点（Point-to-Point）
  • 生产者将消息发送到队列，消费者从队列中获取消息。
  • 每条消息只能被一个消费者处理。
• 发布/订阅（Publish/Subscribe）：
  • 生产者将消息发送到交换机，交换机将消息广播到所有绑定的队列。
  • 每条消息可以被多个消费者处理。

消息队列的常见协议

• AMQP（Advanced Message Queuing Protocol）
  • 一种通用的消息队列协议，支持多种消息队列（如 RabbitMQ）。
• MQTT（Message Queuing Telemetry Transport）：
  • 轻量级的消息协议，适合物联网场景。
• Kafka Protocol：
  • Apache Kafka 自定义的协议，支持高吞吐量的消息处理。
• STOMP（Simple Text Oriented Messaging Protocol）：
  • 基于文本的协议，适合简单的消息队列场景。

消息队列的常见实现

• RabbitMQ：
  • 基于 AMQP 协议，支持多种消息模式，适合中小规模系统。
• Kafka：
  • 高吞吐量的分布式消息系统，适合大数据场景。
• RocketMQ：
  • 阿里巴巴开源的分布式消息队列，适合高并发场景。
• ActiveMQ：
  • 基于 JMS（Java Message Service）的消息队列，适合 Java 生态系统。
• Redis Streams：
  • Redis 提供的轻量级消息队列功能，适合简单场景。

消息队列的核心特性

• 消息持久化：
  • 将消息存储到磁盘，确保消息不丢失。
• 消息确认机制（ACK）：
  • 消费者处理完消息后，向消息队列发送确认，确保消息被正确处理。
• 消息重试机制：
  • 当消息处理失败时，消息队列可以重新投递消息。
• 消息顺序性：
  • 确保消息按照发送顺序被处理。
• 消息过滤：
  • 根据消息的属性（如标签、优先级）过滤消息。

消息队列的使用场景

• 异步处理：
  • 将耗时操作（如发送邮件、生成报表）异步化，提高系统响应速度。
• 应用解耦：
  • 通过消息队列解耦系统模块，降低系统复杂性。
• 流量削峰：
  • 通过队列缓冲消息，避免系统过载。
• 日志收集：
  • 将系统日志发送到消息队列，供后续处理和分析。
• 事件驱动架构：
  • 通过消息队列实现事件驱动的系统架构。

消息队列的常见问题

• 消息丢失：
  • 由于网络故障或系统崩溃，消息可能丢失。
  • 解决方案：启用消息持久化和确认机制。
• 消息重复：
  • 由于网络延迟或消费者故障，消息可能被重复处理。
  • 解决方案：实现幂等性（Idempotence）。
• 消息积压：
  • 消费者处理速度跟不上生产者发送速度，导致消息积压。
  • 解决方案：增加消费者数量或优化消费者逻辑。
• 消息顺序性：
  • 在分布式环境中，消息可能不按顺序到达。
  • 解决方案：使用单分区队列或顺序消息机制。

消息队列的最佳实践

• 合理设计消息格式：
  • 消息体应简洁明了，避免包含过多冗余信息。
• 启用消息持久化：
  • 确保消息在系统崩溃时不丢失。
• 实现幂等性：
  • 消费者处理消息时应支持重复处理。
• 监控消息队列状态：
  • 监控队列长度、消费者延迟等指标，及时发现和解决问题。
• 优化消费者性能：
  • 使用批量处理、异步处理等技术提高消费者性能。