使用fp-ts时如何用TypeScript序列化任务执行？

我有一段将数组中的值转换为异步调用的代码：

const createDiscordChannels = flow(
    A.map((channel: CreateChannelDto) => ({
        ...channel,
        provider,
        externalServerId: providerId,
    })),
    //* 📘 We need to check all the channels before we do anything
    //* because otherwise we can end up in an inconsistent state
    A.map(agentDispatcher.isChannelValid), // 👈 returns TaskEither
    A.map(TE.flatMap(agentDispatcher.ensureChannelExists)), // 👈 creates race condition
    A.map(RTE.fromTaskEither),
    RTE.sequenceArray
);

问题在于A.map(TE.flatMap(agentDispatcher.ensureChannelExists))会创建多个并行执行的TaskEither，从而产生竞态条件。有没有办法改为顺序执行而非并行执行？

要实现顺序执行，核心是避免先用A.map把所有异步任务都初始化出来，而是改用逐个遍历并链式调用的方式，利用ReaderTaskEither的链式执行特性来保证顺序。

基础改造方案 #

直接通过A.reduce遍历数组，把每个通道的校验、创建逻辑串联起来，确保前一个任务完成后再执行下一个：

const createDiscordChannels = flow(
    // 先处理通道的基础数据转换
    (channels: CreateChannelDto[]) => channels.map(channel => ({
        ...channel,
        provider,
        externalServerId: providerId,
    })),
    // 用reduce顺序执行每个通道的异步逻辑
    A.reduce(
        // 初始值：一个包含空数组的成功RTE
        RTE.right<Error, YourResultType[]>([]),
        (accRTE, channel) => 
            accRTE.chain(accResults => 
                // 先校验通道有效性
                RTE.fromTaskEither(agentDispatcher.isChannelValid(channel))
                // 校验通过后执行创建操作
                .flatMap(() => RTE.fromTaskEither(agentDispatcher.ensureChannelExists(channel)))
                // 将当前结果追加到累计数组中
                .map(result => [...accResults, result])
            )
    )
);

兼顾预校验的方案 #

如果要保留原注释中“先检查所有通道再执行操作”的需求，可以先批量并行校验所有通道，确认全部有效后再顺序执行创建逻辑：

const createDiscordChannels = flow(
    (channels: CreateChannelDto[]) => channels.map(channel => ({
        ...channel,
        provider,
        externalServerId: providerId,
    })),
    // 第一步：批量并行校验所有通道，确保都符合要求
    (processedChannels) => 
        RTE.sequenceArray(processedChannels.map(channel => 
            RTE.fromTaskEither(agentDispatcher.isChannelValid(channel))
        )).map(() => processedChannels),
    // 第二步：校验通过后，顺序执行每个通道的创建操作
    RTE.chain(processedChannels => 
        A.reduce(
            RTE.right<Error, YourResultType[]>([]),
            (accRTE, channel) => 
                accRTE.flatMap(accResults => 
                    RTE.fromTaskEither(agentDispatcher.ensureChannelExists(channel))
                    .map(result => [...accResults, result])
                )
        )(processedChannels)
    )
);

关键说明 #

• A.reduce是实现顺序执行的核心：它会把前一个任务的结果作为下一个任务的输入，强制任务按数组顺序逐个执行。
• 原代码中A.map+RTE.sequenceArray的模式会一次性创建所有异步任务并并行执行，这是竞态条件的根源，改造后彻底避免了这种情况。
• 兼顾预校验的方案中，并行校验不会产生竞态（只是读操作），而后续的创建操作是顺序执行，既保证了效率又避免了不一致状态。

内容的提问来源于stack exchange，提问作者Adam A