Shoulder-Batch 批量处理

@Component
public class TestDataImportValidator implements BatchTaskSliceHandler {

    @Override
    public boolean support(String dataType, String operationType) {
        // 如: 该处理器适用于 用户数据批量导入并校验 这个操作
        return "testBatchDataType".equals(dataType) && "IMPORT_VALIDATE".equals(operationType);
    }

    @SneakyThrows
    @Override
    public List<BatchRecordDetail> handle(BatchDataSlice task) {

        ... 校验逻辑

        // 处理校验逻辑，如名字、性别是否格式合法、查数据库是否重复、调接口确认身份证是否真实有效等...
        return task.getBatchList().stream()
                .peek(data -> System.out.println("模拟校验: " + data))
                // 转为校验处理结果
                .map(data -> new BatchRecordDetail(data.getIndex(), data.serialize(), BatchDetailResultStatusEnum.SUCCESS.getCode()))
                .collect(Collectors.toList());

    }

}

@Component
public class TestDataImportHandler extends BaseImportHandler {

    public TestDataImportHandler(BatchRecordDetailPersistentService batchRecordDetailPersistentService) {
        // 只需要根据模型id修改 testBatchDataType 即可
        super("testBatchDataType", Operations.IMPORT, batchRecordDetailPersistentService);
    }

    @Override
    protected List<BatchRecordDetail> updateData(List<BatchRecordDetail> toUpdateList) {

        ... 更新数据

        return toUpdateList.stream()
                // 1.select for update，2.check data version，3.update or reject as fail
                .peek(data -> System.out.println("模拟更新数据，记得并发秘诀：先锁后判再更新。 " + data))
                // 设置这条数据的处理结果状态为成功 or 失败
                .map(data -> BatchRecordDetail.builder()
                        .index(data.getIndex())
                        .status(BatchDetailResultStatusEnum.UPDATE_REPEAT.getCode())
                        .source(data.getSource())
                        .build())
                .collect(Collectors.toList());

    }

    @Override
    protected List<BatchRecordDetail> saveData(List<BatchRecordDetail> toImportList) {

        ... 保存数据

        return toImportList.stream()
                // insert To DB
                .peek(data -> System.out.println("模拟保存数据 " + data.getSource()))
                // 根据保存是否成功，设置这条数据的处理结果状态为成功 or 失败
                .map(data -> BatchRecordDetail.builder()
                        .index(data.getIndex())
                        .status(BatchDetailResultStatusEnum.SUCCESS.getCode())
                        .source(data.getSource())
                        .build())
                .collect(Collectors.toList());

    }
}

POST http://localhost:8080/api/v1/batch/testBatchDataType/advance
Content-Type: application/json

{
  "batchId": ... 批处理任务id
  "dataType": ... 数据模型id
  "currentOperation": ... 该批处理id对应任务当前的操作
  "nextOperation": ... 下一步要干什么
  "updateRepeat": ... 是否更新重复数据
}

import org.shoulder.batch.progress.BatchActivityEnum;

/**
 * <a href="http://localhost:8080/ui/activities/page.html?progressId=_shoulderMockAndTest&activityId=MyAiFlow"></a>
 */
public enum MyAiFlow implements BatchActivityEnum<MyAiFlow> {
    TASK_1("👂", "识别用户意图", 0, 0),
    TASK_2("🧠", "改写查询、扩写查询", 0, 0),
    TASK_3("🔍", "查找相关数据", 0, 0),
    TASK_4("🧠", "总结信息", 0, 0),
    ;

    ... 实现接口方法

    private final String taskKey;
    private final String description;
    private final int displayBlockNum;
    private final int getDisplayColumnNum;
    private final String displayEmoji;

    MyAiFlow(String displayEmoji, String description, int displayBlockNum, int getDisplayColumnNum) {
        this.taskKey = name();
        this.displayEmoji = displayEmoji;
        this.description = description;
        this.displayBlockNum = displayBlockNum;
        this.getDisplayColumnNum = getDisplayColumnNum;
    }

    @Override
    public String getKey() {
        return taskKey;
    }

    @Override
    public boolean hasSubTask() {
        return false;
    }

    @Override
    public String getDisplayName() {
        return description;
    }

    @Override
    public String getDisplayEmoji() {
        return displayEmoji;
    }

    @Override
    public int displayBlockNum() {
        return displayBlockNum;
    }

    @Override
    public int getDisplayColumnNum() {
        return getDisplayColumnNum;
    }

}

@Configuration
public class BatchConfiguration {

    @Bean
    public BatchActivityEnumRepositoryCustomizer batchActivityEnumRepositoryCustomizer() {
        return repository -> {
            repository.register(MyAiFlow.class, "MyAiFlow");
        };
    }
}

public class DemoController {

    @GetMapping("chat")
    public String chat(String userInput) {
        String progressId = UUID.randomUUID().toString();
        // 开启异步线程处理，并直接返回处理进度页面
        Threads.execute("enhancedAiProcess", () -> enhancedAiProcess(userInput, progressId));
        return "redirect:/ui/activities/page.html?progressId=" + progressId + "&activityId=MyAiFlow";
    }

    public String enhancedAiProcess(String userInput, String progressId) {
        // 收到用户输入，解析用户意图
        MySimpleTaskEnum.TASK1.start(progressId);
        String userIntent = AiClient.analyzeUserIntent(userInput);
        MySimpleTaskEnum.TASK1.finish(progressId);

        // 改写查询、扩写查询
        MySimpleTaskEnum.TASK2.start(progressId);
        String rewriteQuery = AiClient.rewriteUserInput(userInput, userIntent);
        MySimpleTaskEnum.TASK2.finish(progressId);

        // 让 AI 去网上搜索
        MySimpleTaskEnum.TASK3.start(progressId);
        String aiWithSearchedResult = AiClient.chatWithTools(rewriteQuery, List.of("WebSearchTool"));
        MySimpleTaskEnum.TASK3.finish(progressId);

        // 根据用户输入、意图、查找结果 总结信息返回用户
        MySimpleTaskEnum.TASK3.start(progressId);
        String finalOutput = AiClient.summary(aiWithSearchedResult, userIntent, userInput);
        MySimpleTaskEnum.TASK3.finish(progressId);

        return finalOutput;
    }
}

public class Demo {

    public void process() {

        MySimpleTaskEnum.TASK1.startOneStageTask(progressId);
        Thread.sleep(1000);
        MySimpleTaskEnum.TASK1.finish(progressId);

        MySimpleTaskEnum.TASK2.setTotalAndStart(progressId, 3);
        Thread.sleep(1000);
        MySimpleTaskEnum.TASK2.addSuccess(progressId);
        Thread.sleep(1000);
        MySimpleTaskEnum.TASK2.addSuccess(progressId);
        Thread.sleep(1000);
        MySimpleTaskEnum.TASK2.addSuccess(progressId);

        // ...
    }
}