基于模板设计模式对数据迁移封装

@RequiredArgsConstructor(onConstructor_={@Autowired})
@RestController
@RequestMapping("/v1/dataTransfer")
public class DataTransferController {

    private final DataTransferJobManager dataTransferJobManager;

    //用户名
    @Value("${business.data-transfer.user:aa64865487710aba549516e5c81fa2a1}")
    private String startUser;
    //用户密码
    @Value("${business.data-transfer.pwd:518be6ab336bccd6cc7e93c227fd1ac6}")
    private String startPwd;

    /**
     * 启动作业
     * @param jobName 作业名称
     * @param user 用户
     * @param pwd 密码
     * @return R
     */
    @ApiOperation(value = "启动作业", notes = "启动作业")
    @ApiImplicitParams({
            @ApiImplicitParam(name = "user",value="帐号",required = true),
            @ApiImplicitParam(name = "pwd", value = "密码", required = true),
    })
    @GetMapping("/start")
    public R<String> start(@RequestParam("jobName") String jobName,@RequestParam("user")String user,@RequestParam("pwd")String pwd){
        if(!startUser.equals(user) || !startPwd.equals(pwd)){
            throw new SystemRuntimeException("启动作业认证用户或密码不正确");
        }

        LoginAccountVO loginUser = HttpRequestHandler.getRequestUser(true);
        dataTransferJobManager.start(jobName,loginUser.getLoginName());
        return R.buildSuccess("启动作业成功");
    }

    /**
     * 重启作业
     * @param jobId 作业Id
     * @param user 用户
     * @param pwd 密码
     * @return R
     */
    @ApiOperation(value = "重启作业", notes = "重启作业")
    @ApiImplicitParams({
            @ApiImplicitParam(name = "user",value="帐号",required = true),
            @ApiImplicitParam(name = "pwd", value = "密码", required = true),
    })
    @GetMapping("/restart")
    public R<String> restart(@RequestParam("jobId") String jobId,@RequestParam("user")String user,@RequestParam("pwd")String pwd){
        if(!startUser.equals(user) || !startPwd.equals(pwd)){
            throw new SystemRuntimeException("重启作业认证用户或密码不正确");
        }

        LoginAccountVO loginUser = HttpRequestHandler.getRequestUser(true);
        dataTransferJobManager.restart(jobId,loginUser.getLoginName());
        return R.buildSuccess("重启作业成功");
    }
}

/**
 * <p>
 * 数据迁移作业 管理器接口
 * </p>
 *
 * @author liJY
 * @Date 2023-06-20
 */
public interface DataTransferJobManager {
    /**
     * 添加作业
     * @param job 作业
     */
    void add(DataTransferJob job);

    /**
     * 初始化作业
     */
    void initJobs();

    /**
     * 启动作业
     * @param jobName 作业名称
     * @param userAccount 用户帐号
     */
    void start(String jobName,String userAccount);

    /**
     * 重动作业
     * @param jobId 作业Id
     * @param userAccount 用户帐号
     */
    void restart(String jobId,String userAccount);

    /**
     * 获取执行器
     * @param jobName 作业名称
     * @param executorKey 执行器键名
     * @return 执行器
     */
    DataTransferExecutor getExecutor(String jobName,String executorKey);
}

@RequiredArgsConstructor(onConstructor_={@Autowired})
@Slf4j
@RefreshScope
public class DataTransferJobManagerImpl implements DataTransferJobManager {
    //作业映射.结构:Map<作业名称,DataTransferJob>
    private Map<String,DataTransferJob> jobMap=new HashMap<>();

    //MQ批量消息发送大小
    @Value(value="${rocketmq.batch-message-size.data-transfer:3500}")
    private int mqBatchMessageSize;
    //消息主题前缀
    @Value(value="${rocketmq.topic-prefix.data-transfer:TOPIC_DATA_TRANSFER}")
    private String mqTopicPrefix;
    //消费者组前缀
    @Value(value = "${rocketmq.consumer.group.prefix.data-transfer:CONSUMER_DATA_TRANSFER}")
    private String mqConsumerGroupPrefix;
    //NameServer
    @Value(value="${rocketmq.name-server}")
    private String mqNameServer;

    private DataTransferTaskService dataTransferTaskService;
    private IdGenerator idGenerator;
    private RocketMQTemplate rocketMQTemplate;
    private DataTransferItemService dataTransferItemService;
    private ApplicationContext applicationContext;

    @Autowired
    public void setDataTransferTaskService(DataTransferTaskService dataTransferTaskService) {
        this.dataTransferTaskService = dataTransferTaskService;
    }
    @Autowired
    public void setIdGenerator(IdGenerator idGenerator) {
        this.idGenerator = idGenerator;
    }
    @Autowired
    public void setRocketMQTemplate(RocketMQTemplate rocketMQTemplate) {
        this.rocketMQTemplate = rocketMQTemplate;
    }
    @Autowired
    public void setDataTransferItemService(DataTransferItemService dataTransferItemService) {
        this.dataTransferItemService = dataTransferItemService;
    }
    @Autowired
    public void setApplicationContext(ApplicationContext applicationContext) {
        this.applicationContext = applicationContext;
    }

    /**
     * 添加作业
     * @param job 作业
     */
    @Override
    public void add(DataTransferJob job) {
        if(jobMap.containsKey(job.getName())){
            log.warn("作业已存在,jobName:{}",job.getName());
            throw new SystemRuntimeException("当前作业已存在");
        }
        jobMap.put(job.getName(),job);
    }

    /**
     * 初始化作业
     */
    @Override
    @PostConstruct
    public void initJobs(){
        //获取所有作业
        if(jobMap.isEmpty()){
            log.warn("缺少作业信息,退出处理");
            return;
        }
        jobMap.forEach((jobName,job)->{
            try{
                log.info("开始初始化Job:{}",jobName);
                //根据作业的执行器,初始化MQ消费者.执行器结构:Map<执行器键名标识,DataTransferExecutor>
                Map<String,DataTransferExecutor> executorMap=job.getExecutorMap();
                if(executorMap.isEmpty()){
                    log.warn("作业缺少执行器信息,退出处理jobName:{}",job.getName());
                    return;
                }
                executorMap.forEach((executorKey,executor)->{
                    try{
                        log.info("开始初始化执行器:{}",executor.getKey());
                        initMqConsumer(job.getName(),executor);
                        log.info("完成初始化执行器:{}",executor.getKey());
                    }catch (Exception e){
                        log.error(MessageFormat.format("失败初始化执行器:{0}",executor.getKey()),e);
                    }
                });
                log.info("完成初始化Job:{}",jobName);
            }catch (Exception e){
                log.error(MessageFormat.format("失败初始化Job:{0}",jobName),e);
            }
        });
    }

    /**
     * 启动作业
     * @param jobName 作业名称
     * @param userAccount 用户帐号
     */
    @Override
    public void start(String jobName,String userAccount) {
        //1.获取作业任务执行器映射
        DataTransferJob job=jobMap.get(jobName);
        if(job==null){
            log.warn("缺少作业信息,jobName:{}",jobName);
            throw new SystemRuntimeException("缺少作业信息");
        }
        //结构:Map<执行器键名标识,DataTransferExecutor>
        Map<String,DataTransferExecutor> executorMap=job.getExecutorMap();

        //2.每一次运行,则属不同的作业实例,生成作业id
        String jobId=idGenerator.nextId("");

        //3.根据执行器创建任务实例
        executorMap.forEach((executorKey,executor)->{
            //通过处理器,获取数据总量
            if(executor.getHandler()==null){
                log.warn("执行器缺少处理器,executorKey:{}",executor.getKey());
                throw new SystemRuntimeException("执行器缺少处理器");
            }
            int total=executor.getHandler().readTotal();

            //创建启动模式任务实例.新启动任务时,来源任务Id为'空'
            DataTransferTaskInstance taskInstance=this.createTask(jobName,jobId,"",executor,DataTransferStartModeEnum.START,total,userAccount);
            if(taskInstance.getTotal()>0){
                //启动任务
                this.startTask(taskInstance);
            }
        });
    }
    
      /**
     * 重动作业
     * @param jobId 作业Id
     * @param userAccount 用户帐号
     */
    @Override
    public void restart(String jobId,String userAccount){
        //1.获取作业下可重启的任务
        List<DataTransferTask> tasks=this.dataTransferTaskService.getRestartTasks(jobId);
        if(CollectionUtils.isEmpty(tasks)){
            log.warn("没有需要重启的任务,jobId:{}",jobId);
            throw new SystemRuntimeException("没有需要重启的任务");
        }

        //2.获取作业任务执行器映射
        //==获取作业名称(同一作业id,作业名称一致)
        String jobName=tasks.get(0).getJobName();
        DataTransferJob job=jobMap.get(jobName);
        if(job==null){
            log.warn("缺少作业信息,jobName:{}",jobName);
            throw new SystemRuntimeException("缺少作业信息");
        }
        //结构:Map<执行器键名标识,DataTransferExecutor>
        Map<String,DataTransferExecutor> executorMap=job.getExecutorMap();

        //3.每一次运行,则属不同的作业实例,生成作业id
        String newJobId=idGenerator.nextId("");

        //4.根据重启的任务列表,创建重启任务实例
        tasks.stream().forEach(task->{
            //获取执行器
            DataTransferExecutor executor=executorMap.get(task.getExecutorKey());

            //获取处理器,并获取任务下迁移失败事项的总数据量
            if(executor.getHandler()==null){
                log.warn("执行器缺少处理器,executorKey:{}",executor.getKey());
                throw new SystemRuntimeException("执行器缺少处理器");
            }
            int total=this.dataTransferItemService.getRestartItemCount(task.getId());

            //创建重启模式的任务实例.来源任务Id为待重启的任务Id
            DataTransferTaskInstance taskInstance=this.createTask(jobName,newJobId,task.getId(),executor,DataTransferStartModeEnum.RESTART,total,userAccount);
            //启动任务
            this.startTask(taskInstance);
        });
    }

     /**
     * 获取执行器
     * @param jobName 作业名称
     * @param executorKey 执行器键名
     * @return 执行器
     */
    @Override
    public DataTransferExecutor getExecutor(String jobName,String executorKey){
        //获取作业
        DataTransferJob job=jobMap.get(jobName);
        if(job==null){
            throw new SystemRuntimeException("缺少作业信息");
        }

        //获取执行器
        DataTransferExecutor executor=job.getExecutorMap().get(executorKey);
        if(executor==null){
            throw new SystemRuntimeException("缺少执行器信息");
        }
        return executor;
    }
    
    /**
     * 初始化MQ消费者
     * @param jobName 作业名称
     * @param executor 执行器
     */
    private void initMqConsumer(String jobName, DataTransferExecutor executor){
        String consumerGroup=getConsumerGroup(mqConsumerGroupPrefix,jobName,executor.getKey());
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(consumerGroup);
        consumer.setConsumeThreadMin(executor.getThreadSize());
        consumer.setConsumeThreadMax(executor.getThreadSize());
        //每次只消费1条消息,批量消费的最大消息数量，缓存的消息数量达到参数设置的值，Push消费者SDK会将缓存的消息统一提交给消费线程，实现批量消费
        consumer.setConsumeMessageBatchMaxSize(1);
        consumer.setConsumeTimeout(executor.getTimeout());
        consumer.setNamesrvAddr(mqNameServer);
        //不作重试处理,若想重试任务,需使用任务的restart功能
        consumer.setMaxReconsumeTimes(0);
        //获取topic
        String topic=this.getMsgTopic(mqTopicPrefix,jobName);
        //获取tag
        String tag=this.getMsgTag(executor.getKey());
        try {
            /*
            说明:
            1.一个作业,使用一个topic
            2.一个执行器,使用一个consumer,但使用不同tag
            */
            consumer.subscribe(topic, tag);
            consumer.registerMessageListener(new MessageListenerConcurrently() {
                @Override
                public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
                    //每次只会消费1条消息
                    MessageExt msgExt=msgs.get(0);
                    //输出消息日志
                    Map<String,Object> msgFields=new HashMap<>();
                    msgFields.put("msgId",msgExt.getMsgId());
                    msgFields.put("reconsumeTimes",msgExt.getReconsumeTimes());
                    msgFields.put("keys",msgExt.getKeys());
                    msgFields.put("topic",msgExt.getTopic());
                    msgFields.put("tags",msgExt.getTags());
                    String body=new String(msgExt.getBody(), StandardCharsets.UTF_8);
                    msgFields.put("body",body);
                    log.info("接收到数据迁移消息,msgFields:{}", msgFields);

                    DataTransferBatch batch=null;
                    try{
                        //转换为批次信息
                        batch=JsonUtils.toObject(body,DataTransferBatch.class);
                    }catch (Exception e){
                       log.error(MessageFormat.format("数据迁移批次信息转换异常,msgFields:{}",msgFields),e);

                        //无论如何,均返回消费成功
                        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                    }

                    //执行单批次数据迁移
                    try {
                        log.info("执行单批次数据迁移开始.jobName:{},taskId:{},batchNum:{}",batch.getJobName(),batch.getTaskId(),batch.getBatchNum());
                        DataTransferBatchHandler batchHandle = (DataTransferBatchHandler) applicationContext.getBean(batch.getModel().getBeanName());
                        batchHandle.execute(batch);
                        log.info("执行单批次数据迁移完成.jobName:{},taskId:{},batchNum:{}",batch.getJobName(),batch.getTaskId(),batch.getBatchNum());
                    }catch (Exception e){
                        log.error(MessageFormat.format("执行单批次数据迁移失败.jobName:{0},taskId:{1},batchNum:{2}",batch.getJobName(),batch.getTaskId(),batch.getBatchNum()),e);
                    }

                    //无论如何,均返回消费成功
                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                }
            });
            consumer.start();
        }catch (Exception e){
            log.error(MessageFormat.format("数据迁移MQ消费者启动失败,jobName:{0},executorKey:{1}",jobName,executor.getKey()),e);
            throw new SystemRuntimeException("数据迁移MQ消费者启动失败");
        }
    }

    /**
     * 创建任务
     * @param jobName 作业名称
     * @param jobId 作业Id
     * @param sourceId 来源任务Id
     * @param executor 执行器
     * @param startMode 启动模式
     * @param total 总数据量
     * @param userAccount 用户帐号
     * @return 任务实例
     */
    private DataTransferTaskInstance createTask(String jobName, String jobId, String sourceId,DataTransferExecutor executor, DataTransferStartModeEnum startMode,int total,String userAccount){
        //获取处理的总批次量
        int batchCount=this.getBatchCount(total,executor.getBatchSize());

        //保存任务
        //==新启动任务,来源任务Id为空
        DataTransferTask task=new DataTransferTask();
        task.setJobName(jobName);
        task.setJobId(jobId);
        task.setExecutorKey(executor.getKey());
        task.setBatchSize(executor.getBatchSize());
        task.setBatchCount(batchCount);
        task.setModuleType(executor.getModuleType());
        task.setBusinessType(executor.getBusinessType());
        task.setTotal(total);
        task.setSourceId(sourceId);
        String taskId=dataTransferTaskService.add(task,userAccount);

        //创建任务实例
        DataTransferTaskInstance taskInstance=new DataTransferTaskInstance();
        taskInstance.setJobName(jobName);
        taskInstance.setJobId(jobId);
        taskInstance.setExecutorKey(executor.getKey());
        taskInstance.setTaskId(taskId);
        taskInstance.setSourceId(sourceId);
        taskInstance.setBatchSize(executor.getBatchSize());
        taskInstance.setBatchCount(batchCount);
        taskInstance.setTotal(total);
        taskInstance.setModuleType(executor.getModuleType());
        taskInstance.setBusinessType(executor.getBusinessType());
        taskInstance.setThreadSize(executor.getThreadSize());
        taskInstance.setTimeout(executor.getTimeout());
        taskInstance.setModel(startMode);
        return taskInstance;
    }


    /**
     * 获取数据批次处理总量
     * @param total 总数量
     * @param batchSize 批次大小
     * @return 批次总量
     */
    private int getBatchCount(Integer total,Integer batchSize){
        if(total==0){
            return 0;
        }
        int count=total/batchSize;
        if(total % batchSize !=0){
            count=count+1;
        }
        return count;
    }

    /**
     * 运行任务实例
     * @param taskInstance 任务实例
     */
    private void startTask(DataTransferTaskInstance taskInstance){
        //根据处理的批次总量,按批次发送MQ消息
        List<Message<DataTransferBatch>> messages = new ArrayList<>();

        for(int batchNum=1;batchNum<=taskInstance.getBatchCount();batchNum++){
            //创建批次信息
            DataTransferBatch batch=new DataTransferBatch();
            batch.setJobName(taskInstance.getJobName());
            batch.setExecutorKey(taskInstance.getExecutorKey());
            batch.setTaskId(taskInstance.getTaskId());
            batch.setSourceId(taskInstance.getSourceId());
            batch.setBatchSize(taskInstance.getBatchSize());
            batch.setBatchCount(taskInstance.getBatchCount());
            batch.setBatchNum(batchNum);
            batch.setTotal(taskInstance.getTotal());
            batch.setModel(taskInstance.getModel());
            //获取消息Key
            String msgKey=this.getMsgKey(batch.getExecutorKey(),batchNum);
            messages.add(MessageBuilder.withPayload(batch).setHeader("KEYS",msgKey).build());

            //判断是否需执行消息发送
            boolean isSend=this.isSendMessage(messages.size(),mqBatchMessageSize,batchNum,taskInstance.getBatchCount());
            if(isSend){
                //获取消息目的地
                String msgDest=MessageFormat.format("{0}:{1}",this.getMsgTopic(mqTopicPrefix,batch.getJobName()),this.getMsgTag(batch.getExecutorKey()));
                //发送消息
                SendResult sendRet=rocketMQTemplate.syncSend(msgDest,messages,10*1000);
                if(!SendStatus.SEND_OK.equals(sendRet.getSendStatus())){
                    log.warn("数据迁移消息发送失败,batch:{},sendRet:{}",batch,sendRet);
                    throw new SystemRuntimeException("消息发送失败");
                }
                //消息发送完毕,则清除
                messages.clear();
            }
        }
    }

    /**
     * 是否执行消息发送
     * @param currentMsgSize 当前消息大小
     * @param sendSize 消息发送大小
     * @param currentMsgCount 当前消息总量
     * @param msgTotal 消息总量
     * @return 是、否
     */
    private Boolean isSendMessage(Integer currentMsgSize,Integer sendSize,Integer currentMsgCount,Integer msgTotal){
        if(currentMsgSize % sendSize==0 || currentMsgCount>=msgTotal){
            //当前消息大小到达发送大小,或当前消息已是最后一条消息,则执行发送
            return true;
        }
        return false;
    }

    /**
     * 获取消息Topic
     * @param topicPrefix 消息主题前缀
     * @param jobName 作业名称
     */
    private String getMsgTopic(String topicPrefix,String jobName){
        //作业名称:作为topic
        return MessageFormat.format("{0}_{1}",topicPrefix,jobName.toUpperCase());
    }

    /**
     * 获取消息tag
     * @param executorKey 执行器键名标识
     * @return 消息tag
     */
    private String getMsgTag(String executorKey){
        //执行器键名标识:作为tag
        return MessageFormat.format("TAG_{0}",executorKey.toUpperCase());
    }

    /**
     * 获取消息Key
     * @param executorKey 执行器键名标识
     * @param batchNum 批次序号
     * @return 消息Key
     */
    private String getMsgKey(String executorKey,int batchNum){
        //Key=执行器键名标识_批次序号
        String result=MessageFormat.format("{0}_{1}",executorKey,batchNum);
        return result;
    }

    /**
     * 获取消费者组
     * @param groupPrefix 消费者组前缀
     * @param jobName 作业名称
     * @param executorKey 执行器名称
     * @return 消费者组全名称
     */
    private String getConsumerGroup(String groupPrefix,String jobName,String executorKey){
        String result=MessageFormat.format("{0}_{1}_{2}",groupPrefix,jobName.toUpperCase(),executorKey.toUpperCase());
        return result;
    }
}

/**
 * <p>
 * 数据迁移作业 对象
 * </p>
 *
 * @author liJY
 * @Date 2023-06-20
 */
@Slf4j
@Getter
@EqualsAndHashCode(callSuper = false)
@Accessors(chain = true)
@ApiModel(value="DataTransferJob对象", description="数据迁移作业 对象")
public class DataTransferJob{

    /**
     * 名称
     */
    private String name;

    /**
     * 任务执行器映射.结构:Map(执行器键名标识,DataTransferExecutor)
     */
    private Map<String,DataTransferExecutor> executorMap=new HashMap<>();

    /**
     * 构造函数
     * @param name 名称
     */
    public DataTransferJob(String name){
        this.name=name;
    }

    /**
     * 添加执行器
     * @param executor 执行器
     */
    public void add(DataTransferExecutor executor){
        if(StringUtils.isEmpty(executor.getKey())){
            log.warn("执行器缺少键名标识");
            throw new SystemRuntimeException("执行器缺少键名标识");
        }
        if(executorMap.containsKey(executor.getKey())){
            log.warn("执行器已存在,jobName:{},key:{}",this.name,executor.getKey());
            throw new SystemRuntimeException("执行器已存在");
        }
        executorMap.put(executor.getKey(),executor);
    }
}

@Data
@EqualsAndHashCode(callSuper = false)
@Accessors(chain = true)
public class DataTransferExecutor<SOURCE,TARGET>{
    /**
     * 键名标识
     */
    private String key;

    /**
     * 数据处理批次大小
     */
    private int batchSize;

    /**
     * 处理线程大小(消费者线程数)
     */
    private int threadSize;

    /**
     * 处理超时时间.单位:分钟
     */
    private int timeout=15;

    /**
     * 模块类型
     */
    private DataTransferModuleTypeEnum moduleType;

    /**
     * 业务类型
     */
    private DataTransferBusinessTypeEnum businessType;

    /**
     * 处理器
     */
    private DataTransferHandler<SOURCE,TARGET> handler;
}

/**
 * 数据迁移处理器
 * @author liJY
 * @Date 2023-06-20
 * @param <SOURCE> 数据来源对象类型
 * @param <TARGET> 数据目标对象类型
 */
public interface DataTransferHandler<SOURCE,TARGET> {
    /**
     * 读取来源处理事项总数(用于首次启动任务场景)
     * @return 总数量
     */
    int readTotal();

    /**
     * 读取来源数据(用于首次启动任务场景)
     * @param batchSize 批次大小
     * @param batchNum 批次序号
     * @return 来源数据列表.返回数量比批次数量少的,则为跳过处理,空:没有待处理数据
     */
    List<DataTransferSource<SOURCE>> read(int batchSize, int batchNum);

    /**
     * 读取来源数据(用于重启任务场景)
     * @param businessIds 业务Id列表
     * @return 来源数据列表.返回数量比入参数量少的,则为跳过处理,空:没有待处理数据
     */
    List<DataTransferSource<SOURCE>> read(List<String> businessIds);

    /**
     * 数据处理
     * @param sources 来源数据列表
     * @return 目标数据列表.返回数量比入参数量少的,则为跳过处理.空:没有待处理数据
     */
    List<DataTransferResult<TARGET>> process(List<DataTransferSource<SOURCE>> sources);

    /**
     * 写入目标数据
     * @param targets 目标数据列表.返回数量比入参数量少的,则为跳过处理.空:没有待处理数据
     */
    List<DataTransferResult> write(List<TARGET> targets);
}

@Configuration
public class DataTransferConfig {
    private final DataTransferBusinessProperties dataTransferBusinessProperties;
    private final ApplicationContext applicationContext;

    public DataTransferConfig(DataTransferBusinessProperties dataTransferBusinessProperties,ApplicationContext applicationContext){
        this.dataTransferBusinessProperties=dataTransferBusinessProperties;
        this.applicationContext=applicationContext;
    }
    
    /**
     * 实例对象注入Spring IOC容器
     * @return
     */
    @Bean
    public DataTransferJobManager jobManager(){
        DataTransferJobManager manager=new DataTransferJobManagerImpl();
        //合同交付模块Job
        manager.add(this.buildDeliverRegisterJob());
        //停复工模块Job
        manager.add(this.buildWorkResumeJob());
        //停复工模块-复工确认Job，必须跑完停复工申请的Job才能跑复工确认的Job，因为业务数据存在前后时间关联
        manager.add(this.buildResumeConfirmDetailJob());
        return manager;
    }

    /**
     * 构建合同交付数据迁移作业
     * @return 数据迁移作业
     */
    private DataTransferJob buildDeliverRegisterJob(){
        //定义合同交付申请任务执行器
        DataTransferBusinessProperties.TaskConfig applyTaskConfig=dataTransferBusinessProperties.getDeliverRegisterApply();
        DataTransferExecutor applyExecutor=new DataTransferExecutor();
        applyExecutor.setKey("apply");
        applyExecutor.setBatchSize(applyTaskConfig.getBatchSize());
        applyExecutor.setThreadSize(applyTaskConfig.getThreadSize());
        applyExecutor.setTimeout(applyTaskConfig.getTimeout());
        applyExecutor.setModuleType(DataTransferModuleTypeEnum.DELIVER_REGISTER);
        applyExecutor.setBusinessType(DataTransferBusinessTypeEnum.DELIVER_REGISTER_APPLY);
        applyExecutor.setHandler(this.applicationContext.getBean(DeliverRegisterDtApplyHandler.class));

        //定义合同交付申请明细任务执行器
        DataTransferBusinessProperties.TaskConfig detailTaskConfig=dataTransferBusinessProperties.getDeliverRegisterDetail();
        DataTransferExecutor detailExecutor=new DataTransferExecutor();
        detailExecutor.setKey("detail");
        detailExecutor.setBatchSize(detailTaskConfig.getBatchSize());
        detailExecutor.setThreadSize(detailTaskConfig.getThreadSize());
        detailExecutor.setTimeout(detailTaskConfig.getTimeout());
        detailExecutor.setModuleType(DataTransferModuleTypeEnum.DELIVER_REGISTER);
        detailExecutor.setBusinessType(DataTransferBusinessTypeEnum.DELIVER_REGISTER_APPLY_DETAIL);
        detailExecutor.setHandler(this.applicationContext.getBean(DeliverRegisterDtDetailHandler.class));

        DataTransferJob job=new DataTransferJob("deliverRegister");
        job.add(applyExecutor);
        job.add(detailExecutor);
        return job;
    }

    /**
     * 构建停复工数据迁移作业
     * @return 数据迁移作业
     */
    private DataTransferJob buildWorkResumeJob(){
        //定义停工计划申请任务执行器
        DataTransferBusinessProperties.TaskConfig stopPlanApplyTaskConfig=dataTransferBusinessProperties.getWorkResumeStopPlanApply();
        DataTransferExecutor stopPlanApplyExecutor=new DataTransferExecutor();
        stopPlanApplyExecutor.setKey("stopPlanApply");
        stopPlanApplyExecutor.setBatchSize(stopPlanApplyTaskConfig.getBatchSize());
        stopPlanApplyExecutor.setThreadSize(stopPlanApplyTaskConfig.getThreadSize());
        stopPlanApplyExecutor.setTimeout(stopPlanApplyTaskConfig.getTimeout());
        stopPlanApplyExecutor.setModuleType(DataTransferModuleTypeEnum.WORK_RESUME);
        stopPlanApplyExecutor.setBusinessType(DataTransferBusinessTypeEnum.WORK_RESUME_STOP_PLAN_APPLY);
        stopPlanApplyExecutor.setHandler(this.applicationContext.getBean(WorkResumeDtStopPlanApplyHandler.class));
        
        //定义停工计划明细任务执行器
        DataTransferBusinessProperties.TaskConfig resumeDetailTaskConfig=dataTransferBusinessProperties.getWorkResumeStopPlanDetail();
        DataTransferExecutor stopPlanDetailExecutor = new DataTransferExecutor();
        stopPlanDetailExecutor.setKey("stopPlanDetail");
        stopPlanDetailExecutor.setBatchSize(resumeDetailTaskConfig.getBatchSize());
        stopPlanDetailExecutor.setThreadSize(resumeDetailTaskConfig.getThreadSize());
        stopPlanDetailExecutor.setTimeout(resumeDetailTaskConfig.getTimeout());
        stopPlanDetailExecutor.setModuleType(DataTransferModuleTypeEnum.WORK_RESUME);
        stopPlanDetailExecutor.setBusinessType(DataTransferBusinessTypeEnum.WORK_RESUME_STOP_PLAN_DETAIL);
        stopPlanDetailExecutor.setHandler(this.applicationContext.getBean(WorkResumeDtStopPlanDetailHandler.class));
        
        //定义复工确认申请任务执行器
        DataTransferBusinessProperties.TaskConfig confirmApplyTaskConfig=dataTransferBusinessProperties.getWorkResumeConfirmApply();
        DataTransferExecutor confirmApplyExecutor=new DataTransferExecutor();
        confirmApplyExecutor.setKey("confirmApply");
        confirmApplyExecutor.setBatchSize(confirmApplyTaskConfig.getBatchSize());
        confirmApplyExecutor.setThreadSize(confirmApplyTaskConfig.getThreadSize());
        confirmApplyExecutor.setTimeout(confirmApplyTaskConfig.getTimeout());
        confirmApplyExecutor.setModuleType(DataTransferModuleTypeEnum.WORK_RESUME);
        confirmApplyExecutor.setBusinessType(DataTransferBusinessTypeEnum.WORK_RESUME_CONFIRM_APPLY);
        confirmApplyExecutor.setHandler(this.applicationContext.getBean(WorkResumeDtConfirmApplyHandler.class));
        
        DataTransferJob job=new DataTransferJob("workResume");
        job.add(stopPlanApplyExecutor);
        job.add(confirmApplyExecutor);
        job.add(stopPlanDetailExecutor);
        return job;
    }
    
    /**
     * 构建复工确认明细数据迁移作业
     */
    private DataTransferJob buildResumeConfirmDetailJob(){
        //定义停复工确认明细执行器
        DataTransferBusinessProperties.TaskConfig confirmDetailTaskConfig=dataTransferBusinessProperties.getWorkResumeConfirmDetail();
        DataTransferExecutor confirmDetailExecutor = new DataTransferExecutor();
        confirmDetailExecutor.setKey("resumeConfirmDetail");
        confirmDetailExecutor.setBatchSize(confirmDetailTaskConfig.getBatchSize());
        confirmDetailExecutor.setThreadSize(confirmDetailTaskConfig.getThreadSize());
        confirmDetailExecutor.setTimeout(confirmDetailTaskConfig.getTimeout());
        confirmDetailExecutor.setModuleType(DataTransferModuleTypeEnum.WORK_RESUME);
        confirmDetailExecutor.setBusinessType(DataTransferBusinessTypeEnum.WORK_RESUME_CONFIRM_DETAIL);
        confirmDetailExecutor.setHandler(this.applicationContext.getBean(WorkResumeDtConfirmDetailHandler.class));
        DataTransferJob job=new DataTransferJob("resumeConfirmDetail");
        job.add(confirmDetailExecutor);
        return job;
    }
}

@Component
@Data
@RefreshScope
@ConfigurationProperties(prefix = "data-transfer.business")
public class DataTransferBusinessProperties {
    /**
     * 合同交付申请迁移任务配置
     */
    TaskConfig deliverRegisterApply;

    /**
     * 合同交付申请明迁移任务配置
     */
    TaskConfig deliverRegisterDetail;
    
    /**
     * 停复工_计划停工申请任务配置
     */
    TaskConfig workResumeStopPlanApply;
    
    TaskConfig workResumeStopPlanDetail;

    /**
     * 停复工_复工确认申请任务配置
     */
    TaskConfig workResumeConfirmApply;
    
    TaskConfig workResumeConfirmDetail;

    /**
     * 任务配置
     */
    @Data
    public static class TaskConfig{
        /**
         * 数据处理批次大小
         */
        private Integer batchSize;

        /**
         * 处理线程大小
         */
        private Integer threadSize;

        /**
         * 处理超时时间.单位:分钟
         */
        private Integer timeout;

        /**
         * 来源数据迁移条件
         */
        private DataTransferSourceQuery query;
    }
}

@Data
@EqualsAndHashCode(callSuper = false)
@Accessors(chain = true)
@ApiModel(value = "数据迁移_来源数据查询条件对象", description = "数据迁移_来源数据查询条件对象")
public class DataTransferSourceQuery {
    /**
     * 开始时间
     */
    @DateTimeFormat(pattern = "yyyy-MM-dd HH:mm:ss")
    private LocalDateTime beginTime;
    /**
     * 结束时间
     */
    @DateTimeFormat(pattern = "yyyy-MM-dd HH:mm:ss")
    private LocalDateTime endTime;
    /**
     * Id列表
     */
    private List<String> ids;
}

/**
 * <p>
 * 数据迁移任务表
 * </p>
 *
 * @author liJY
 * @Date 2023-06-20
 */
@Data
@EqualsAndHashCode(callSuper = false)
@Accessors(chain = true)
@TableName("t_data_transfer_task")
public class DataTransferTask extends Model<DataTransferTask> {

    private static final long serialVersionUID = 1L;

    /**
     * 主键名称
     */
    @TableId(value = "id", type = IdType.ASSIGN_ID)
    private String id;

    /**
     * 创建人bip
     */
    @TableField("create_user")
    private String createUser;

    /**
     * 创建时间
     */
    @TableField("create_time")
    private LocalDateTime createTime;

    /**
     * 更新人bip
     */
    @TableField("update_user")
    private String updateUser;

    /**
     * 更新时间
     */
    @TableField("update_time")
    private LocalDateTime updateTime;

    /**
     * 租户编码
     */
    @TableField("tenant_id")
    private String tenantId;

    /**
     * 版本号
     */
    @TableField("version_number")
    private String versionNumber;

    /**
     * 备注
     */
    @TableField("remark")
    private String remark;

    /**
     * 作业名称
     */
    @TableField("job_name")
    private String jobName;

    /**
     * 作业Id
     */
    @TableField("job_id")
    private String jobId;

    /**
     * 执行器键名标识
     */
    @TableField("executor_key")
    private String executorKey;

    /**
     * 来源任务Id
     */
    @TableField("source_id")
    private String sourceId;

    /**
     * 批次大小
     */
    @TableField("batch_size")
    private Integer batchSize;

    /**
     * 批次总量
     */
    @TableField("batch_count")
    private Integer batchCount;

    /**
     * 开始时间
     */
    @TableField("start_time")
    private LocalDateTime startTime;

    /**
     * 结束时间
     */
    @TableField("end_time")
    private LocalDateTime endTime;

    /**
     * 模块类型
     */
    @TableField("module_type")
    private DataTransferModuleTypeEnum moduleType;

    /**
     * 业务类型
     */
    @TableField("business_type")
    private DataTransferBusinessTypeEnum businessType;

    /**
     * 状态
     */
    @TableField("status")
    private DataTransferTaskStatusEnum status;

    /**
     * 总数据量
     */
    @TableField("total")
    private Integer total;

    /**
     * 成功事项数量
     */
    @TableField("success")
    private Integer success;

    /**
     * 失败事项数量
     */
    @TableField("fail")
    private Integer fail;

    /**
     * 跳过事项数量
     */
    @TableField("skip")
    private Integer skip;

    @Override
    protected Serializable pkVal() {
        return this.id;
    }

}

@Data
@EqualsAndHashCode(callSuper = false)
@Accessors(chain = true)
@TableName("t_data_transfer_item")
public class DataTransferItem extends Model<DataTransferItem> {

    private static final long serialVersionUID = 1L;

    /**
     * 主键名称
     */
    @TableId(value = "id", type = IdType.ASSIGN_ID)
    private String id;

    /**
     * 删除标识
     */
    @TableField("delete_tag")
    private DeletedEnum deleteTag;

    /**
     * 创建人bip
     */
    @TableField("create_user")
    private String createUser;

    /**
     * 创建时间
     */
    @TableField("create_time")
    private LocalDateTime createTime;

    /**
     * 更新人bip
     */
    @TableField("update_user")
    private String updateUser;

    /**
     * 更新时间
     */
    @TableField("update_time")
    private LocalDateTime updateTime;

    /**
     * 租户编码
     */
    @TableField("tenant_id")
    private String tenantId;

    /**
     * 版本号
     */
    @TableField("version_number")
    private String versionNumber;

    /**
     * 备注
     */
    @TableField("remark")
    private String remark;

    /**
     * 任务Id
     */
    @TableField("task_id")
    private String taskId;

    /**
     * 主体Id.数据:业务Id,批次:批次序号
     */
    @TableField("subject_id")
    private String subjectId;

    /**
     * 主体类型
     */
    @TableField("subject_type")
    private DataTransferSubjectTypeEnum subjectType;

    /**
     * 状态
     */
    @TableField("status")
    private DataTransferItemStatusEnum status;

    /**
     * 数量
     */
    @TableField("number")
    private Integer number;

    /**
     * 阶段
     */
    @TableField("stage")
    private DataTransferItemStageEnum stage;

    /**
     * 追踪Id
     */
    @TableField("trace_id")
    private String traceId;

    /**
     * 错误信息
     */
    @TableField("error")
    private String error;

    @Override
    protected Serializable pkVal() {
        return this.id;
    }
}

dataTransferJobManager.start(jobName,loginUser.getLoginName());

    @Override
    public void start(String jobName,String userAccount) {
        //1.获取作业任务执行器映射
        DataTransferJob job=jobMap.get(jobName);
        if(job==null){
            log.warn("缺少作业信息,jobName:{}",jobName);
            throw new SystemRuntimeException("缺少作业信息");
        }
        //结构:Map<执行器键名标识,DataTransferExecutor>
        Map<String,DataTransferExecutor> executorMap=job.getExecutorMap();

        //2.每一次运行,则属不同的作业实例,生成作业id
        String jobId=idGenerator.nextId("");

        //3.根据执行器创建任务实例
        executorMap.forEach((executorKey,executor)->{
            //通过处理器,获取数据总量
            if(executor.getHandler()==null){
                log.warn("执行器缺少处理器,executorKey:{}",executor.getKey());
                throw new SystemRuntimeException("执行器缺少处理器");
            }
            // 查询任务迁移总数据量
            int total=executor.getHandler().readTotal();

            //创建启动模式任务实例.新启动任务时,来源任务Id为'空'
            DataTransferTaskInstance taskInstance=this.createTask(jobName,jobId,"",executor,DataTransferStartModeEnum.START,total,userAccount);
            if(taskInstance.getTotal()>0){
                //启动任务
                this.startTask(taskInstance);
            }
        });
    }

@Data
@EqualsAndHashCode(callSuper = false)
@Accessors(chain = true)
@ApiModel(value="DataTransferTaskInstance对象", description="数据迁移任务实例 对象")
public class DataTransferTaskInstance {
    /**
     * 作业名称
     */
    private String jobName;

    /**
     * 作业Id
     */
    private String jobId;

    /**
     * 执行器键名标识
     */
    private String executorKey;

    /**
     * 任务Id
     */
    private String taskId;

    /**
     * 来源任务Id
     */
    private String sourceId;

    /**
     * 数据处理批次大小
     */
    private int batchSize;

    /**
     * 数据处理批次总量
     */
    private int batchCount;

    /**
     * 总数据量
     */
    private int total;

    /**
     * 处理线程大小
     */
    private int threadSize;

    /**
     * 处理超时时间.单位:分钟
     */
    private int timeout;

    /**
     * 模块类型
     */
    private DataTransferModuleTypeEnum moduleType;

    /**
     * 业务类型
     */
    private DataTransferBusinessTypeEnum businessType;

    /**
     * 启动模式
     */
    private DataTransferStartModeEnum model;

}

if(taskInstance.getTotal()>0){
    //启动任务
    this.startTask(taskInstance);
}

/**
     * 运行任务实例
     * @param taskInstance 任务实例
     */
    private void startTask(DataTransferTaskInstance taskInstance){
        //根据处理的批次总量,按批次发送MQ消息
        List<Message<DataTransferBatch>> messages = new ArrayList<>();

        for(int batchNum=1;batchNum<=taskInstance.getBatchCount();batchNum++){
            //创建批次信息
            DataTransferBatch batch=new DataTransferBatch();
            batch.setJobName(taskInstance.getJobName());
            batch.setExecutorKey(taskInstance.getExecutorKey());
            batch.setTaskId(taskInstance.getTaskId());
            batch.setSourceId(taskInstance.getSourceId());
            batch.setBatchSize(taskInstance.getBatchSize());
            batch.setBatchCount(taskInstance.getBatchCount());
            batch.setBatchNum(batchNum);
            batch.setTotal(taskInstance.getTotal());
            batch.setModel(taskInstance.getModel());
            //获取消息Key
            String msgKey=this.getMsgKey(batch.getExecutorKey(),batchNum);
            messages.add(MessageBuilder.withPayload(batch).setHeader("KEYS",msgKey).build());

            //判断是否需执行消息发送
            boolean isSend=this.isSendMessage(messages.size(),mqBatchMessageSize,batchNum,taskInstance.getBatchCount());
            if(isSend){
                //获取消息目的地
                String msgDest=MessageFormat.format("{0}:{1}",this.getMsgTopic(mqTopicPrefix,batch.getJobName()),this.getMsgTag(batch.getExecutorKey()));
                //发送消息
                SendResult sendRet=rocketMQTemplate.syncSend(msgDest,messages,10*1000);
                if(!SendStatus.SEND_OK.equals(sendRet.getSendStatus())){
                    log.warn("数据迁移消息发送失败,batch:{},sendRet:{}",batch,sendRet);
                    throw new SystemRuntimeException("消息发送失败");
                }
                //消息发送完毕,则清除
                messages.clear();
            }
        }
    }

@Data
@EqualsAndHashCode(callSuper = false)
@Accessors(chain = true)
@ApiModel(value="DataTransferBatch对象", description="数据迁移任务批次 对象")
public class DataTransferBatch implements Serializable {

    private static final long serialVersionUID = 1L;

    /**
     * 作业名称
     */
    private String jobName;

    /**
     * 执行器键名标识
     */
    private String executorKey;

    /**
     * 任务Id
     */
    private String taskId;

    /**
     * 来源任务Id
     */
    private String sourceId;

    /**
     * 数据处理批次大小
     */
    private int batchSize;

    /**
     * 数据处理批次总量
     */
    private int batchCount;

    /**
     * 数据处理批次序号
     */
    private int batchNum;

    /**
     * 总数据量
     */
    private int total;

    /**
     * 启动模式
     */
    private DataTransferStartModeEnum model;
}

public enum DataTransferStartModeEnum implements BaseEnum<Integer> {
    /**
     * 启动
     */
    START(1,"启动","dataTransferBatchStartHandler"),
    /**
     * 重启
     */
    RESTART(2, "重启","dataTransferBatchRestartHandler");
    private final Integer value;

    private final String displayName;

    private final String beanName;

    DataTransferStartModeEnum(Integer value, String displayName,String beanName) {
        this.value = value;
        this.displayName = displayName;
        this.beanName=beanName;
    }

    /**
     * 枚举value
     *
     * @return
     */
    @Override
    public Integer getValue() {
        return value;
    }

    /**
     * 枚举的显示名字
     *
     * @return
     */
    @Override
    public String getDisplayName() {
        return displayName;
    }

    /**
     * Bean名称
     *
     * @return
     */
    public String getBeanName() {
        return beanName;
    }
}

//执行单批次数据迁移
try {
    log.info("执行单批次数据迁移开始.jobName:{},taskId:{},batchNum:{}",batch.getJobName(),batch.getTaskId(),batch.getBatchNum());
    DataTransferBatchHandler batchHandle = (DataTransferBatchHandler) applicationContext.getBean(batch.getModel().getBeanName());
    batchHandle.execute(batch);
    log.info("执行单批次数据迁移完成.jobName:{},taskId:{},batchNum:{}",batch.getJobName(),batch.getTaskId(),batch.getBatchNum());
}catch (Exception e){
    log.error(MessageFormat.format("执行单批次数据迁移失败.jobName:{0},taskId:{1},batchNum:{2}",batch.getJobName(),batch.getTaskId(),batch.getBatchNum()),e);
}

public interface DataTransferBatchHandler {
    /**
     * 执行单批次数据迁移
     * @param batch 批次信息
     */
    void execute(DataTransferBatch batch);
}

@Slf4j
public class DataTransferBatchHandlerSuper {
    protected DataTransferJobManager jobManager;
    protected DataTransferTaskService dataTransferTaskService;
    protected DataTransferItemService dataTransferItemService;
    protected IdGenerator idGenerator;
    protected ApplicationContext applicationContext;

    /**
     * 执行单批次数据处理
     * @param batch 批次信息
     * @param handler 处理器
     * @param sources 来源数据列表
     * @return 处理成功结果列表.null:没有成功处理结果或失败,不需执行后续迁移操作
     */
    public List<DataTransferResult> executeBatchProcess(DataTransferBatch batch,DataTransferHandler handler,List<DataTransferSource> sources){
        //处理成功结果
        List<DataTransferResult> processSuccess=null;
        try {
            List<DataTransferResult> processResult = handler.process(sources);
            //对处理结果生成失败事项,并返回处理成功结果.对没返回的数据,生成跳过处理事项
            //==源业务Id列表为:来源数据业务Id
            List<String> businessIds=sources.stream().map(DataTransferSource::getBusinessId).collect(Collectors.toList());
            processSuccess=this.applicationContext.getBean(this.getClass()).resultDataHandle(batch,DataTransferItemStageEnum.PROGRESS,processResult,businessIds);
            if(CollectionUtils.isEmpty(processSuccess)){
                //成功处理结果为空,则表示不需执行后续的'写入'操作
                log.info("数据处理成功的结果为空,不再执行后续阶段的迁移操作.taskId:{},batchNum:{}",batch.getTaskId(),batch.getBatchNum());
                return null;
            }
        }catch (Exception e){
            //整批次的数据处理失败,根据业务Id列表生成多个数据主体的失败事项(业务Id为:来源数据业务Id)
            List<String> businessIds=sources.stream().map(DataTransferSource::getBusinessId).collect(Collectors.toList());
            this.applicationContext.getBean(this.getClass()).batchDataHandleFail(batch,"数据处理",DataTransferItemStageEnum.PROGRESS,businessIds,e);
            return null;
        }
        return processSuccess;
    }

    /**
     * 执行单版本目标数据写入
     * @param batch 批次信息
     * @param handler 处理器
     * @param processSuccess 处理成功结果列表
     */
    public void executeBatchWrite(DataTransferBatch batch,DataTransferHandler handler,List<DataTransferResult> processSuccess){
        try {
            List<Object> targets = processSuccess.stream().map(DataTransferResult::getData).collect(Collectors.toList());
            List<DataTransferResult> writeResult = handler.write(targets);
            //对处理结果生成失败事项.对没返回的数据,生成跳过处理事项
            //==源业务Id列表为:处理成功结果业务Id
            List<String> businessIds=processSuccess.stream().map(DataTransferResult::getBusinessId).collect(Collectors.toList());
            this.applicationContext.getBean(this.getClass()).resultDataHandle(batch,DataTransferItemStageEnum.WRITE,writeResult,businessIds);
        }catch (Exception e){
            //整批次的数据处理失败,根据业务Id列表生成多个数据主体的失败事项(业务Id为:成功处理结果业务Id)
            List<String> businessIds=processSuccess.stream().map(DataTransferResult::getBusinessId).collect(Collectors.toList());
            // 获取bean的方式执行方法，避免了@Transactional等注解的失效问题
            this.applicationContext.getBean(this.getClass()).batchDataHandleFail(batch,"写入目标数据",DataTransferItemStageEnum.WRITE,businessIds,e);
        }
    }

    /**
     * 输出异常信息,并返回迁移失败状态
     * @param exception 异常信息
     * @param traceId 追踪Id
     * @param action 动作描述
     * @param jobName 作业名称
     * @param executorKey 执行器键名
     * @param batchNum 批次序号
     * @return 迁移失败状态
     */
    protected DataTransferItemStatusEnum handleException(Exception exception,String traceId,String action,String jobName,String executorKey,Integer batchNum){
        //日志提示
        String logTips=null;
        DataTransferItemStatusEnum status=null;
        if(exception instanceof SystemRuntimeException){
            //业务异常
            status=DataTransferItemStatusEnum.BUSINESS_ERROR;
            logTips= MessageFormat.format("数据迁移处理结束,{0}时出现业务异常.traceId:{1},jobName:{2},executorKey:{3},batchNum:{4}"
                    ,action,traceId,jobName,executorKey,batchNum);
        }else{
            //系统异常
            status=DataTransferItemStatusEnum.SYSTEM_ERROR;
            logTips=MessageFormat.format("数据迁移处理结束,{0}时出现系统异常.traceId:{1},jobName:{2},executorKey:{3},batchNum:{4}"
                    ,action,traceId,jobName,executorKey,batchNum);
        }
        log.error(logTips,exception);
        return status;
    }

    /**
     * 生成单个批次主体的跳过处理事项
     * @param batch 批次信息
     * @param realTotal 实际返回数量
     * @param stage 阶段
     */
    @Transactional(rollbackFor = RuntimeException.class,transactionManager = "selfTransactionManager")
    public void batchHandleSkip(DataTransferBatch batch, int realTotal,DataTransferItemStageEnum stage){
        //1.计算跳过处理数量
        //获取当前批次应处理数量
        int batchItemCount=this.getBatchItemCount(batch.getBatchSize(),batch.getBatchNum(),batch.getBatchCount(),batch.getTotal());
        //跳过数量=当前批次应处理数量-实际返回数量
        int skipCount=batchItemCount-realTotal;

        //2.生成迁移跳过处理事项
        if(skipCount<=0){
            //没有跳过处理的事项,则退出处理
            log.info("不存在跳过处理事项,退出处理.jobName:{},taskId:{},batchNum:{}",batch.getJobName(),batch.getTaskId(),batch.getBatchNum());
            return;
        }

        //3.保存跳过处理事项
        DataTransferItem skipItem=new DataTransferItem();
        //主体Id为'批次序号'
        skipItem.setSubjectId(String.valueOf(batch.getBatchNum()));
        skipItem.setStatus(DataTransferItemStatusEnum.SKIP);
        //跳过处理数量
        skipItem.setNumber(skipCount);
        skipItem.setTraceId("");
        skipItem.setError("");
        this.dataTransferItemService.batchAdd(batch.getTaskId(),DataTransferSubjectTypeEnum.BATCH,stage,Arrays.asList(skipItem),"");

        //4.递增跳过处理事项状态数量,并检测任务完成状态
        boolean isFinish=this.dataTransferTaskService.incrItemCount(batch.getTaskId(),0,0,skipCount);
        if(isFinish){
            log.info("数据迁移任务已完成,jobName:{},taskId:{},batchNum:{}",batch.getJobName(),batch.getTaskId(),batch.getBatchNum());
        }
    }

    /**
     * 获取批次实际事项数量
     * @param batchSize 批次大小
     * @param batchNum 批次序号.从1开始
     * @param batchCount 批次数量
     * @param total 总数据量
     * @return 事项数量
     */
    protected int getBatchItemCount(Integer batchSize,Integer batchNum,Integer batchCount,Integer total){
        if(batchNum<batchCount){
            //不是最后一批,事项数量为批次大小
            return batchSize;
        }
        //最后一批,事项数量=总数据量-批次大小*(批次序号-1)
        return total-batchSize*(batchNum-1);
    }

    /**
     * 整批次处理失败,生成单个批次主体的失败事项
     * @param batch 批次信息
     * @param action 动作描述
     * @param stage 阶段
     * @param exception 异常
     */
    @Transactional(rollbackFor = RuntimeException.class,transactionManager = "selfTransactionManager")
    public void batchHandleFail(DataTransferBatch batch, String action, DataTransferItemStageEnum stage, Exception exception){
        //生成追踪Id
        String traceId=idGenerator.nextId("");

        //1.处理异常信息,并返回迁移失败状态
        DataTransferItemStatusEnum status=this.handleException(exception,traceId,action,batch.getJobName(),batch.getExecutorKey(),batch.getBatchNum());

        //2.创建失败迁移事项
        //获取批次实际失败数量
        int batchFailCount=this.getBatchItemCount(batch.getBatchSize(),batch.getBatchNum(),batch.getBatchCount(),batch.getTotal());

        DataTransferItem item=new DataTransferItem();
        item.setSubjectId(String.valueOf(batch.getBatchNum()));
        item.setStatus(status);
        item.setNumber(batchFailCount);
        item.setTraceId(traceId);
        item.setError(exception.getMessage());
        this.dataTransferItemService.batchAdd(batch.getTaskId(),DataTransferSubjectTypeEnum.BATCH,stage, Arrays.asList(item),"");

        //3.递增失败数量并检测任务完成状态
        boolean isFinish=this.dataTransferTaskService.incrItemCount(batch.getTaskId(),0,batchFailCount,0);
        if(isFinish){
            log.info("数据迁移任务已完成,jobName:{},taskId:{},batchNum:{}",batch.getJobName(),batch.getTaskId(),batch.getBatchNum());
        }
    }

    /**
     * 整批次的数据处理失败,根据业务Id列表生成多个数据主体的失败事项
     * @param batch 批次信息
     * @param action 动作描述
     * @param stage 阶段
     * @param businessIds 业务Id列表
     * @param exception 异常
     */
    @Transactional(rollbackFor = RuntimeException.class,transactionManager = "selfTransactionManager")
    public void batchDataHandleFail(DataTransferBatch batch, String action, DataTransferItemStageEnum stage, List<String> businessIds, Exception exception){
        //生成追踪Id
        String traceId=idGenerator.nextId("");

        //处理异常信息,并返回迁移失败状态
        DataTransferItemStatusEnum status=this.handleException(exception,traceId,action,batch.getJobName(),batch.getExecutorKey(),batch.getBatchNum());

        //1.根据业务Id,生成整批次的迁移事项
        List<DataTransferItem> items=businessIds.stream().map(businessId->{

            DataTransferItem item=new DataTransferItem();
            //处理结果失败时,主体Id为业务Id
            item.setSubjectId(businessId);
            item.setStatus(status);
            //一个业务Id,数量值为:1
            item.setNumber(1);
            item.setTraceId(traceId);
            item.setError(exception.getMessage());
            return item;
        }).collect(Collectors.toList());
        //保存迁移事项.主体类型为'数据'
        this.dataTransferItemService.batchAdd(batch.getTaskId(),DataTransferSubjectTypeEnum.DATA,stage,items,"");

        //2.递增任务失败数量,并检测完成状态
        //检测任务完成状态
        boolean isFinish=this.dataTransferTaskService.incrItemCount(batch.getTaskId(),0,items.size(),0);
        if(isFinish){
            log.info("数据迁移任务已完成,jobName:{},taskId:{},batchNum:{}",batch.getJobName(),batch.getTaskId(),batch.getBatchNum());
        }
    }

    /**
     * 根据处理结果与源业务Id对比,生成跳过事项、失败事项,并返回成功结果列表
     * @param batch 批次信息
     * @param stage 阶段
     * @param results 处理结果列表
     * @param businessIds 源业务Id列表
     * @return 处理成功结果列表
     */
    @Transactional(rollbackFor = RuntimeException.class,transactionManager = "selfTransactionManager")
    public List<DataTransferResult> resultDataHandle(DataTransferBatch batch,DataTransferItemStageEnum stage,List<DataTransferResult> results,List<String> businessIds){
        //处埋成功结果
        List<DataTransferResult> successResults=new ArrayList<>();
        //失败事项
        List<DataTransferItem> failItems= new ArrayList<>();

        //1.获取处理成功结果、失败事项
        for(int i=0;i<results.size();i++){
            DataTransferResult result=results.get(i);
            switch (result.getStatus()){
                case SUCCESS:
                    //处理成功结果
                    successResults.add(result);
                    continue;
                case SKIP:
                    //跳过
                    throw new SystemRuntimeException(MessageFormat.format("发现无效状态处理结果.状态:{}",result.getStatus()));
            }

            //生成处理失败事项
            DataTransferItem item=new DataTransferItem();
            //==主体Id为业务Id
            item.setSubjectId(result.getBusinessId());
            item.setStatus(result.getStatus());
            item.setTraceId(result.getTraceId());
            item.setError(result.getError());
            //==一个结果对应数量为:1
            item.setNumber(1);
            failItems.add(item);
        }

        //2.生成跳过处理事项
        //获取处理结果的业务Id列表
        List<String> resultBusinessIds=results.stream().map(DataTransferResult::getBusinessId).collect(Collectors.toList());
        List<DataTransferItem> skipItems=this.getSkipItems(businessIds,resultBusinessIds);

        //3.保存迁移失败事项、跳过事项.主体类型为'数据'
        List<DataTransferItem> addItems=new ArrayList<>();
        addItems.addAll(failItems);
        addItems.addAll(skipItems);

        //5.更新迁移事项
        if(CollectionUtils.isNotEmpty(addItems)){
            this.dataTransferItemService.batchAdd(batch.getTaskId(),DataTransferSubjectTypeEnum.DATA,stage,addItems,"");
        }

        //6.更新任务失败、跳过处理数量,以及完成状态
        //获取迁移失败事项数量
        int failCount=failItems.size();
        //获取跳过处理数量
        int skipCount=skipItems.size();
        //获取迁移功事项数量
        int successCount=0;
        if(DataTransferItemStageEnum.WRITE.equals(stage)){
            //只有在写入目标数据阶段为成功,才递增成功数量
            successCount=successResults.size();
        }
        //递增事项状态数量,并检测任务完成状态
        boolean isFinish=this.dataTransferTaskService.incrItemCount(batch.getTaskId(),successCount,failCount,skipCount);
        if(isFinish){
            log.info("数据迁移任务已完成,jobName:{},taskId:{},batchNum:{}",batch.getJobName(),batch.getTaskId(),batch.getBatchNum());
        }
        return successResults;
    }

    /**
     * 根据源业务Id与返回结果业务Id对比,获得跳过的处理事项
     * @param fromBusinessIds 源业务Id
     * @param resultBusinessIds 结果业务Id
     * @return 跳过事项列表
     */
    protected List<DataTransferItem> getSkipItems(List<String> fromBusinessIds,List<String> resultBusinessIds){
        //获取跳过处理的业务Id列表
        List<String> skipBusinessIds=new ArrayList<>();
        skipBusinessIds.addAll(fromBusinessIds);
        skipBusinessIds.removeAll(resultBusinessIds);
        //生成跳过事项
        List<DataTransferItem> skipItems=skipBusinessIds.stream().map(skipBusinessId->{
            DataTransferItem item=new DataTransferItem();
            item.setSubjectId(skipBusinessId);
            item.setStatus(DataTransferItemStatusEnum.SKIP);
            item.setTraceId("");
            item.setError("");
            //一个业务Id对应数量为:1
            item.setNumber(1);
            return item;
        }).collect(Collectors.toList());
        return skipItems;
    }
}