功能迭代时间线

第一版：基础同步功能（直接保存，问题多多）
第二版：增加去重逻辑（解决重复数据问题）
第三版：增加平台推送功能（但是同步推送，很慢）
第四版：引入异步处理（性能大幅提升）
第五版：优化推送去重逻辑（避免重复推送）
第六版：增加缓存优化（减少90%重复调用）
第七版：增加定时任务支持（实现自动化）

业务流程图

数据流程时序图

代码详解

@Slf4j
@Service
public class SyncPolicyServiceBO implements Job, SyncPolicyService {

    @Autowired
    private GeneralDAO generalDAO;

    @Autowired
    private CruxUrlConfig urlConfig;

    @Autowired
    private CruxUtil cruxUtil;

    private static Cache<String, Triple<String, String, String>> mofCache = CacheBuilder.newBuilder().expireAfterWrite(3, TimeUnit.HOURS).build();

    private static ThreadPoolExecutor poolExecutor = new ThreadPoolExecutor(5, 10, 10,TimeUnit.SECONDS, new LinkedBlockingQueue<>(100), new ThreadFactoryBuilder().setNameFormat("syncPolicy-pool-%d").build());

类继承了quartz的定时任务接口，然后注入了三个类，三个都是内部封装的类，作用分别是，操控数据库，封装url配置和http请求相关类。
接下来声明了一个Cache类型，key是string类型，value是一个三元组Triple包含三个 String 值：第一个 String: mofDivId - 财政区划的唯一标识ID，第二个 String: typeId - 要素类型的唯一标识ID，第三个 String: lastModifiedVersion - 最后修改版本号，CacheBuilder.newBuilder()，创建一个新的缓存构建器，使用建造者模式来配置缓存的各种参数，然后expireAfterWrite(3, TimeUnit.HOURS)是缓存过期策略配置，expireAfterWrite: 基于写入时间的过期策略，表示缓存项在写入后经过指定时间就会过期，3: 过期时间的数值，TimeUnit.HOURS: 时间单位，build()根据前面配置的参数构建并返回 Cache 实例，

然后采用的是Executor框架创建ThreadPoolExecutor 构造函数的corePoolSize = 5，线程池中始终保持活跃的线程数量，即使这些线程处于空闲状态，也不会被回收，当有任务提交时，如果当前线程数少于5个，会创建新线程来处理任务，实际意义: 这个政策同步服务保证至少有5个线程随时待命，maximumPoolSize = 10，线程池中允许存在的最大线程数量，当任务很多时，线程池最多可以扩展到10个线程，超过这个数量的任务会被放入队列等待，实际意义: 政策推送高峰期最多同时运行10个推送任务，keepAliveTime = 10，当线程数超过核心线程数时，多余的线程在空闲状态下的最大存活时间，new LinkedBlockingQueue<>(100)，LinkedBlockingQueue: 基于链表的阻塞队列，容量100: 队列最多可以存放100个待处理的任务，阻塞特性: 当队列满时，新提交的任务会被阻塞，new ThreadFactoryBuilder().setNameFormat("syncPolicy-pool-%d").build()；setNameFormat("syncPolicy-pool-%d"): 设置线程命名格式；%d: 会被替换为线程编号

execute方法

@Override
    @SneakyThrows
    public void execute(JobExecutionContext ctx) throws JobExecutionException {
        JobDataMap taskParams = ctx.getJobDetail().getJobDataMap();
        String reqUrl = taskParams.getString("reqUrl");
        Map<String, Object> params = new HashMap<>();
        params.put("reqUrl", reqUrl);
        syncPolicy(params);
    }

这是一个定时触发同步政策的方法，参数: JobExecutionContext ctx - Quartz 提供的执行上下文
ctx: JobExecutionContext，包含了任务执行的所有上下文信息

getJobDetail(): 获取 JobDetail 对象，包含任务的详细配置；getJobDataMap(): 获取任务参数映射表；JobDataMap: 类似于 Map，存储任务配置的键值对
然后从参数中提取url，然后构建参数对象params，把requrl put进去，最后再调用核心业务方法，syncPolicy
PS：如果定时任务时效的话，可以配置Quartz的重试策略

SyncPolicy方法（）

@Override
    public void syncPolicy(a<String, Object> params) {
        List<PolicyResDTO> findData = pullData(params);
        List<PolicyResDTO> noSave = filterNoSave(findData);
        if(CollectionUtils.isNotEmpty(noSave)){
            //保存之前先过滤 避免同一个事物能查到当前保存的数据
            List<PolicyResDTO> pushPolicy = filterPushData(noSave);
            save(noSave);
            if(CollectionUtils.isNotEmpty(pushPolicy)){
                for(PolicyResDTO dto : pushPolicy){
                    Triple<String, String, String> mofAndEleInfo = getMofAndEleInfo(dto.getMof_div_code(), PayConstant.HOLD15_ELE);
                    CompletableFuture.runAsync(()->pushData(dto, mofAndEleInfo), poolExecutor);
                }
            }else{
                log.info("【{}区划推送平台数据】此次获取的该区划中的政策在平台都已存在", noSave.get(0).getMof_div_code());
            }
        }

    }

这是政策同步的核心流程，分为4个主要阶段：
// 1. 数据拉取 → 2. 数据过滤 → 3. 数据保存 → 4. 平台推送
方法接收hashmap类型的参数，然后进行数据拉取，目的: 从博思拉取政策数据；参数: params 包含请求URL和查询条件；返回: PolicyResDTO 列表，包含政策的完整信息，然后对拉取获得的数据进行第一次过滤，目的: 过滤掉数据库中已经存在的政策，避免重复保存；过滤逻辑: 根据 agency_code + hold15_code 的组合判断是否已存在。数据库查询: 查询 PAY_NBYB_AGENCYPOLICYRELATION 表，返回结果: 只包含数据库中不存在的新政策，然后进入判断逻辑，如果noSave列表是空的，直接结束，如果不为空，则进入下一段业务逻辑，系统拉取最新的政策数据，将本地没有的政策数据保存到本地数据库，再将这些数据推送到其他平台，这里有一个场景问题，

 1. 事务A正在保存数据到数据库
// 2. 同时事务B在查询数据库，准备推送到平台
// 3. 如果B能看到A还未提交的数据，可能导致：
//    - 推送了不完整的数据
//    - 数据状态不一致

// 解决方案：
// 先确定哪些需要推送，再保存数据
// 这样推送的都是基于"保存前"状态的数据

对需要保存的数据进行二次过滤，确定哪些需要推送到平台，去重逻辑: 同一个区划的同一个政策只推送一条，避免重复推送: 检查数据库中是否已有该政策的推送记录，然后过滤完了，我们再保存，然后再次进入判断逻辑，如果有数据需要推送 → 执行推送逻辑，如果没有数据需要推送 → 输出日志说明，然后进入逻辑开始遍历逐个处理每个需要推送的政策，每个政策都会创建一个独立的异步任务。

pullData

这个 pullData 方法接受一个 Map 类型的参数，首先通过 debug 级别记录传入的参数信息用于调试。接着从参数 Map 中提取两个关键配置：reqUrl（博思系统的API接口地址）和 updateDay（时间偏移天数）。方法会先校验 updateDay 参数是否为空，如果为空则抛出运行时异常提示配置缺失。然后将 updateDay 字符串转换为整数，并调用 getDateTimeAfterDays 方法计算出查询的起始时间点，这个时间点通常是当前时间减去指定天数，用于实现增量数据同步。紧接着创建一个 PolicyReqDTO 请求对象，从用户会话中获取财政区划编码，并对特定的区划编码"330299000"做特殊处理，将其转换为"330200000"以适配博思系统的编码规范。随后将区划编码和计算出的时间设置到请求对象中，使用 FastJSON 将请求对象序列化为 JSON 字符串，其中空值会被转换为空字符串而非 null。然后通过 BSRequestUtil.post 方法向博思系统发送 HTTP POST 请求，并将响应自动反序列化为 PolicyResDTO 对象列表。如果响应结果为空，则记录错误日志并抛出运行时异常。最后对返回的数据进行反向转换处理，将博思系统返回的区划编码"330200000"转换回内部系统的"330299000"，确保数据在内部系统中使用正确的编码格式，最终返回处理后的政策数据列表。

getDateTimeAfterDays

这个 getDateTimeAfterDays 静态工具方法接受一个整数类型的天数参数，用于计算相对于当前时间的日期时间字符串。方法首先通过 LocalDateTime.now() 获取当前系统的本地日期时间，然后使用 plusDays() 方法在当前时间基础上增加指定的天数（如果传入负数则实际是减去相应天数，这样可以计算过去的时间点）。接着创建一个日期时间格式化器，使用 "yyyy-MM-dd HH:mm:ss" 的标准格式模式，这种格式便于数据库查询和API调用。最后通过 format() 方法将计算得到的 LocalDateTime 对象按照指定格式转换为字符串并返回。整个方法设计简洁且线程安全，能够自动处理月末、年末、闰年等各种边界情况，为政策数据的增量同步提供精确的时间基准点。

filternosave

这是整个业务的多次过滤的第一次过滤主要作用是过滤掉数据库中已经存在的政策数据，只返回新增保存的数据，避免重复插入。然后进入到方法内，接受一个list对象，然后对数据用stream进行一个处理，只要agencycode和hold15-code，再得到一个列表，然后用stringbuilder构建mysql的查询语句，然后进入过滤逻辑，