跳转到正文

重复执行

以 Markdown 格式查看 在 GitHub 上编辑 文档位置 2cc7b2b

Queuety 工作流由按顺序排列的步骤链组成。使用重复控制步骤可返回之前命名的步骤，而不必手动编写 _next_step 逻辑。

选择符合工作流需求的形式：

  • 条件简单时使用公开的 state_key
  • 重复逻辑较复杂时使用 condition_class
  • 需要为重复操作设置硬性停止条件时使用 max_iterations

两种方法都会指向之前命名的步骤，因此重复执行仍可序列化、重放，并受与其他步骤相同的工作流运行时保护。

前置步骤完成后，系统会根据已持久化的公开工作流状态评估重复条件，因此每次迭代在工作进程重启后仍能持久保留。

使用状态键的 repeat_until()

PHP
Queuety::workflow( 'approval_revision_loop' )
    ->max_transitions( 12 )
    ->then( DraftBriefStep::class, 'draft' )
    ->wait_for_decision( result_key: 'review' )
    ->then( NormalizeReviewOutcomeStep::class, 'review_outcome' )
    ->repeat_until( 'draft', 'review_approved', true, 'repeat_draft_until_approved' )
    ->then( PublishBriefStep::class )
    ->dispatch( [ 'brief_id' => 42 ] );

NormalizeReviewOutcomeStep 可以将决策负载展平为一个简单的公开键：

PHP
class NormalizeReviewOutcomeStep implements \Queuety\Step {
    public function handle( array $state ): array {
        return [
            'review_approved' => 'approved' === ( $state['review']['outcome'] ?? null ),
        ];
    }

    public function config(): array {
        return [];
    }
}

如果 review_approved 仍不是 true，重复步骤就会跳回 draft。一旦变为 true，工作流就会继续执行 PublishBriefStep。

信号会被存储，而不会被消费。第二次经过 wait_for_decision() 时，之前的 rejected 信号仍然匹配，等待会立即结束，因此循环会持续运行，直到 max_transitions() 将其停止。为每轮设置独立的键：让 DraftBriefStep 返回递增的 review_round，将 correlation_key: 'review_round' 传给 wait_for_decision()，并让审核者在信号负载中发送相同的 review_round。只有当信号的 review_round 与当前状态值相等时，等待才会结束。

使用条件类的 repeat_until()

PHP
use App\Workflows\ReviewApprovedCondition;

Queuety::workflow( 'approval_revision_loop' )
    ->then( DraftBriefStep::class, 'draft' )
    ->wait_for_decision( result_key: 'review' )
    ->repeat_until(
        target_step: 'draft',
        condition_class: ReviewApprovedCondition::class,
        max_iterations: 5,
        name: 'repeat_draft_until_approved',
    )
    ->then( PublishBriefStep::class )
    ->dispatch( [ 'brief_id' => 42 ] );
PHP
namespace App\Workflows;

use Queuety\Contracts\RepeatCondition;

final class ReviewApprovedCondition implements RepeatCondition {
    public function matches( array $state ): bool {
        return 'approved' === ( $state['review']['outcome'] ?? null );
    }
}

条件类接收公开的工作流状态，因此谓词可序列化且可检查，无需依赖闭包。

repeat_while()

PHP
Queuety::workflow( 'poll_remote_status' )
    ->max_transitions( 20 )
    ->then( PollRemoteStatusStep::class, 'poll' )
    ->delay( seconds: 30 )
    ->repeat_while( 'poll', 'should_poll_again', true, 'repeat_poll' )
    ->then( FinalizeImportStep::class )
    ->dispatch( [ 'import_id' => 99 ] );

当某个步骤写入简单布尔值（例如 should_poll_again、has_more_pages 或 needs_revision）时，这种模式很有用。

为什么重复执行与 _next_step 分开

你仍可以通过在普通步骤中返回 _next_step 来手动分支。重复辅助方法适用于以下情况：

  • 目标是之前的步骤
  • 条件来自公开的工作流状态
  • 你希望重复逻辑直接显示在工作流定义中

这样一来，与将回边隐藏在任意步骤代码中相比，重复操作更容易在导出内容、事件日志和文档中检查。

为重复操作设置保护

当某次重复操作需要本地硬性停止条件时，使用 max_iterations：

PHP
Queuety::workflow( 'poll_remote_status' )
    ->then( PollRemoteStatusStep::class, 'poll' )
    ->repeat_while( 'poll', 'should_poll_again', true, max_iterations: 8 )
    ->dispatch();

当整个工作流需要更广泛的自动停止开关时，使用 max_transitions()：

PHP
Queuety::workflow( 'safe_revision_loop' )
    ->max_transitions( 10 )
    ->then( DraftBriefStep::class, 'draft' )
    ->wait_for_decision( result_key: 'review' )
    ->then( NormalizeReviewOutcomeStep::class )
    ->repeat_until( 'draft', 'review_approved', true )
    ->dispatch();

如果重复次数超过 max_iterations，或工作流超过 max_transitions()，Queuety 会使此次运行失败，避免其无限循环。