datai/datai-scenes/datai-scene-salesforce/datai-salesforce-integration/systemPatterns.md

Salesforce数据集成模块 - 系统架构说明

系统架构概述

本系统采用经典的分层架构设计，结合工厂模式、代理模式等设计模式，实现了高内聚、低耦合的Salesforce数据集成解决方案。系统基于Spring Boot框架，使用MyBatis作为持久层框架，支持多种Salesforce API协议。

1. 分层架构

1.1 表现层（Controller Layer）

职责：接收HTTP请求，参数校验，调用业务层，返回响应

核心组件：

• DataiIntegrationObjectController：对象同步控制
• DataiIntegrationFieldController：字段管理
• DataiIntegrationBatchController：批次管理
• DataiIntegrationSyncLogController：同步日志管理

技术特性：

• 基于Spring MVC的RESTful API设计
• 使用Swagger/OpenAPI 3进行接口文档
• 基于Spring Security的权限控制
• 统一的异常处理
• 操作日志记录（@Log注解）

设计模式：

• 控制器模式：统一请求处理流程
• DTO模式：数据传输对象，隔离外部接口和内部模型

1.2 业务层（Service Layer）

职责：实现业务逻辑，事务管理，调用数据访问层

核心组件：

• ISalesforceDataPullService：Salesforce数据拉取服务
• IDataiIntegrationObjectService：对象管理服务
• IDataiIntegrationFieldService：字段管理服务
• IDataiIntegrationBatchService：批次管理服务
• IDataiIntegrationSyncLogService：同步日志服务
• IDataiIntegrationApiCallLogService：API调用日志服务
• IDataiIntegrationRateLimitService：限流管理服务
• IDataiIntegrationMetadataChangeService：元数据变更服务

技术特性：

• 基于Spring的依赖注入
• 使用@Transactional进行事务管理
• 使用@Async进行异步处理
• 使用@Cacheable进行缓存管理
• 使用@Scheduled进行定时任务

设计模式：

• 服务层模式：封装业务逻辑
• 策略模式：不同的同步策略（全量/增量）
• 模板方法模式：定义同步流程骨架

1.3 数据访问层（Mapper Layer）

职责：数据库操作，SQL映射

核心组件：

• DataiIntegrationObjectMapper：对象数据访问
• DataiIntegrationFieldMapper：字段数据访问
• DataiIntegrationBatchMapper：批次数据访问
• DataiIntegrationBatchHistoryMapper：批次历史数据访问
• DataiIntegrationSyncLogMapper：同步日志数据访问
• DataiIntegrationApiCallLogMapper：API调用日志数据访问
• DataiIntegrationRateLimitMapper：限流数据访问
• DataiIntegrationMetadataChangeMapper：元数据变更数据访问
• CustomMapper：自定义SQL
• SalesforceMapper：Salesforce相关SQL

技术特性：

• 基于MyBatis的ORM映射
• 使用XML配置SQL语句
• 支持动态SQL
• 使用ResultMap进行结果映射

设计模式：

• DAO模式：数据访问对象
• Mapper模式：SQL映射

1.4 模型层（Model Layer）

职责：数据模型定义，数据传输

核心组件：

领域模型（Domain）：

• DataiIntegrationObject：对象实体
• DataiIntegrationField：字段实体
• DataiIntegrationBatch：批次实体
• DataiIntegrationBatchHistory：批次历史实体
• DataiIntegrationSyncLog：同步日志实体
• DataiIntegrationApiCallLog：API调用日志实体
• DataiIntegrationRateLimit：限流实体
• DataiIntegrationMetadataChange：元数据变更实体
• DataiIntegrationPicklist：选择列表实体
• DataiIntegrationFilterLookup：过滤查找实体

数据传输对象（DTO）：

• DataiIntegrationObjectDto：对象DTO
• DataiIntegrationFieldDto：字段DTO
• DataiIntegrationBatchDto：批次DTO
• DataiIntegrationSyncLogDto：同步日志DTO
• DataiIntegrationApiCallLogDto：API调用日志DTO
• DataiIntegrationRateLimitDto：限流DTO
• DataiIntegrationMetadataChangeDto：元数据变更DTO
• DataiConfigAuditLogDto：配置审计日志DTO
• DataiConfigEnvironmentDto：配置环境DTO
• DataiConfigSnapshotDto：配置快照DTO

视图对象（VO）：

• DataiIntegrationObjectVo：对象VO
• DataiIntegrationFieldVo：字段VO
• DataiIntegrationBatchVo：批次VO
• DataiIntegrationSyncLogVo：同步日志VO
• DataiIntegrationApiCallLogVo：API调用日志VO
• DataiIntegrationRateLimitVo：限流VO
• DataiIntegrationMetadataChangeVo：元数据变更VO

参数对象（Param）：

• DataiSyncParam：同步参数

技术特性：

• 使用Lombok简化代码
• 使用Swagger注解生成API文档
• 继承BaseEntity基类
• 使用Jackson进行JSON序列化

设计模式：

• DTO模式：数据传输对象
• VO模式：视图对象
• Builder模式：对象构建

2. 核心子系统架构

2.1 连接管理子系统

职责：管理Salesforce连接，支持多种API协议

核心组件：

• SessionManager：会话管理器
• AbstractConnectionFactory：连接工厂抽象基类
• RESTConnectionFactory：REST连接工厂
• SOAPConnectionFactory：SOAP连接工厂
• BulkV1ConnectionFactory：Bulk V1连接工厂
• BulkV2ConnectionFactory：Bulk V2连接工厂
• ConnectionProxy：连接代理

接口定义：

• ISalesforceConnectionFactory：连接工厂接口
• IRESTConnection：REST连接接口
• IPartnerV1Connection：SOAP连接接口
• IBulkV1Connection：Bulk V1连接接口
• IBulkV2Connection：Bulk V2连接接口

实现类：

• RESTConnection：REST连接实现
• PartnerV1Connection：SOAP连接实现
• BulkV1Connection：Bulk V1连接实现
• BulkV2Connection：Bulk V2连接实现

设计模式：

• 工厂模式：创建不同类型的连接
• 抽象工厂模式：定义连接工厂的抽象
• 模板方法模式：定义连接创建流程
• 单例模式：连接实例缓存
• 代理模式：连接代理

技术特性：

• 连接池管理
• 会话缓存
• 自动重连
• 线程安全

2.2 数据同步子系统

职责：实现Salesforce数据同步功能

核心组件：

• SalesforceDataPullServiceImpl：数据拉取服务实现

同步流程：

1. 全量同步流程：

   • 查询所有全量批次
   • 使用线程池并行拉取批次数据
   • 统计同步结果
   • 更新对象状态

2. 增量同步流程：

   • 检查增量更新开关
   • 创建增量批次
   • 拉取增量数据
   • 更新对象状态

设计模式：

• 策略模式：全量同步和增量同步策略
• 模板方法模式：同步流程模板
• 观察者模式：同步状态监听

技术特性：

• 多线程并行处理
• 批次级别的错误隔离
• 自动重试机制
• 进度跟踪

2.3 元数据管理子系统

职责：管理Salesforce对象和字段的元数据

核心组件：

• 对象元数据同步
• 字段元数据同步
• 元数据变更监控

元数据同步流程：

1. 从Salesforce获取对象元数据
2. 保存到本地数据库
3. 获取对象字段元数据
4. 保存字段信息
5. 获取选择列表值
6. 保存选择列表信息
7. 获取过滤查找信息
8. 保存过滤查找信息
9. 记录元数据变更

设计模式：

• 建造者模式：元数据对象构建
• 观察者模式：元数据变更监听

2.4 监控与日志子系统

职责：监控系统运行状态，记录操作日志

核心组件：

• API调用日志
• 同步日志
• 限流监控
• 统计分析

日志记录流程：

1. 拦截器捕获操作
2. 记录操作信息
3. 记录操作结果
4. 记录执行时间
5. 保存到数据库

设计模式：

• 观察者模式：日志监听
• 责任链模式：日志处理链

2.5 定时任务子系统

职责：执行定时任务

核心组件：

• SalesforceSyncTask：同步任务
• RateLimitResetTask：限流重置任务

技术特性：

• 基于Spring @Scheduled的定时任务
• 支持Cron表达式
• 任务执行状态监控

3. 数据库设计

3.1 核心表结构

对象表（datai_integration_object）：

• 存储Salesforce对象的元数据和配置信息
• 支持对象级别的同步配置
• 记录对象同步状态

字段表（datai_integration_field）：

• 存储Salesforce对象的字段信息
• 记录字段属性和权限
• 支持字段级别的配置

批次表（datai_integration_batch）：

• 存储数据同步的批次配置
• 支持全量批次和增量批次
• 记录批次执行状态

批次历史表（datai_integration_batch_history）：

• 存储批次执行历史
• 记录批次执行结果
• 支持批次执行分析

同步日志表（datai_integration_sync_log）：

• 存储数据同步操作日志
• 记录同步操作类型和状态
• 支持同步操作审计

API调用日志表（datai_integration_api_call_log）：

• 存储Salesforce API调用日志
• 记录API调用时间和耗时
• 支持API使用分析

限流表（datai_integration_rate_limit）：

• 存储API限流信息
• 记录API使用量
• 支持限流监控

元数据变更表（datai_integration_metadata_change）：

• 存储元数据变更记录
• 记录字段新增、修改、删除
• 支持元数据变更审计

选择列表表（datai_integration_picklist）：

• 存储选择列表值
• 记录选择列表配置
• 支持选择列表管理

过滤查找表（datai_integration_filter_lookup）：

• 存储过滤查找信息
• 记录字段依赖关系
• 支持字段依赖分析

3.2 分库分表方案

当前方案：

• 单库单表
• 基于对象API的表隔离
• 每个对象对应一张表

分区方案：

• 对于大数据量对象（> 500万条），使用分区表
• 按日期字段进行范围分区
• 支持按月、按季度、按年分区

分区表示例：

CREATE TABLE sf_account (
    id VARCHAR(18) PRIMARY KEY,
    name VARCHAR(255),
    created_date DATETIME,
    last_modified_date DATETIME
) PARTITION BY RANGE (YEAR(created_date)) (
    PARTITION p2023 VALUES LESS THAN (2024),
    PARTITION p2024 VALUES LESS THAN (2025),
    PARTITION p2025 VALUES LESS THAN (2026),
    PARTITION pmax VALUES LESS THAN MAXVALUE
);

分表方案：

• 对于超大数据量对象（> 1亿条），考虑分表
• 按对象ID哈希分表
• 支持动态扩容

分表示例：

-- 按ID哈希分表
CREATE TABLE sf_account_0 (
    id VARCHAR(18) PRIMARY KEY,
    name VARCHAR(255),
    ...
);
CREATE TABLE sf_account_1 (
    id VARCHAR(18) PRIMARY KEY,
    name VARCHAR(255),
    ...
);

3.3 索引设计

主键索引：

• 所有表都使用自增ID作为主键
• Salesforce对象表使用Salesforce ID作为主键

唯一索引：

• 对象API唯一索引
• 字段API+字段名称唯一索引

普通索引：

• 同步状态索引
• 同步时间索引
• 批次类型索引
• 创建时间索引

复合索引：

• 对象API+同步状态复合索引
• 批次API+批次类型复合索引

4. 并发控制

4.1 线程池配置

SalesforceExecutor：

• 核心线程数：CPU核心数
• 最大线程数：CPU核心数 * 2
• 队列容量：100
• 线程存活时间：60秒

4.2 并发控制策略

连接并发控制：

• 使用连接池管理连接
• 使用ReentrantLock保证线程安全
• 使用ConcurrentHashMap缓存连接

数据同步并发控制：

• 使用线程池执行同步任务
• 使用CopyOnWriteArrayList存储结果
• 使用Future管理异步任务

数据库并发控制：

• 使用事务保证数据一致性
• 使用乐观锁防止并发冲突
• 使用悲观锁保证关键操作

4.3 分布式锁

当前方案：

• 单机部署，使用本地锁

未来方案：

• 分布式部署，使用Redis分布式锁
• 支持集群部署

5. 缓存策略

5.1 会话缓存

缓存内容：

• Salesforce会话信息
• 连接实例

缓存策略：

• 使用Spring Cache
• 基于内存的缓存
• 会话过期自动清理

5.2 元数据缓存

缓存内容：

• 对象元数据
• 字段元数据

缓存策略：

• 使用Spring Cache
• 基于内存的缓存
• 定时刷新

5.3 配置缓存

缓存内容：

• 对象配置
• 批次配置

缓存策略：

• 使用Spring Cache
• 基于内存的缓存
• 配置变更时刷新

6. 事务管理

6.1 事务策略

本地事务：

• 使用Spring @Transactional
• 基于数据库事务
• 支持事务传播

6.2 事务边界

同步事务：

• 批次级别的事务
• 单个批次失败不影响其他批次

元数据事务：

• 对象级别的事务
• 对象同步失败不影响其他对象

6.3 事务隔离级别

默认隔离级别：

• READ_COMMITTED

特殊场景：

• 关键操作使用SERIALIZABLE

7. 异常处理

7.1 异常分类

业务异常：

• 对象不存在
• 批次不存在
• 配置错误

系统异常：

• 连接失败
• 网络超时
• 数据库错误

第三方异常：

• Salesforce API错误
• 限流触发
• 会话过期

7.2 异常处理策略

自动重试：

• 网络异常自动重试
• 限流触发自动重试
• 会话过期自动重连

错误记录：

• 记录错误日志
• 记录错误信息
• 记录错误堆栈

错误恢复：

• 批次级别的错误恢复
• 手动重试机制
• 断点续传

8. 安全设计

8.1 认证授权

认证：

• 基于Spring Security
• 支持多种认证方式

授权：

• 基于角色的访问控制（RBAC）
• 细粒度的权限控制
• 接口级别的权限控制

8.2 数据安全

数据加密：

• 敏感数据加密存储
• 传输数据加密

数据隔离：

• 基于部门的数据隔离
• 基于用户的数据隔离

8.3 审计日志

操作审计：

• 记录所有操作
• 记录操作人
• 记录操作时间

数据审计：

• 记录数据变更
• 记录变更前后值
• 记录变更原因

9. 性能优化

9.1 查询优化

索引优化：

• 合理创建索引
• 避免全表扫描
• 使用覆盖索引

SQL优化：

• 避免SELECT *
• 使用批量操作
• 使用分页查询

9.2 缓存优化

多级缓存：

• 一级缓存：MyBatis缓存
• 二级缓存：Redis缓存
• 本地缓存：Caffeine缓存

缓存策略：

• 热点数据缓存
• 定时刷新
• 缓存预热

9.3 并发优化

异步处理：

• 使用异步任务
• 使用消息队列
• 使用事件驱动

批量处理：

• 批量插入
• 批量更新
• 批量删除

10. 可扩展性设计

10.1 插件化设计

连接插件：

• 支持自定义连接工厂
• 支持自定义连接实现

同步插件：

• 支持自定义同步策略
• 支持自定义数据转换

10.2 配置化设计

动态配置：

• 支持运行时配置变更
• 支持配置热加载

策略配置：

• 支持同步策略配置
• 支持批次策略配置

10.3 模块化设计

模块独立：

• 模块间低耦合
• 模块可独立部署
• 模块可独立扩展

接口标准：

• 定义标准接口
• 支持多种实现
• 支持接口替换