如何平衡多个消息队列

Question

我有一项可能需要长时间运行（数小时）的任务。该任务由从消息队列（在我的情况下为 AWS SQS）读取的多个工作人员（在我的情况下为 AWS ECS 实例）执行。我有多个用户将消息添加到队列中。问题是，如果 Bob 向队列中添加 5000 条消息，足以让工作人员忙 3 天，然后 Alice 出现并想要处理 5 个任务，Alice 需要等待 3 天才能开始 Alice 的任何任务。

我想在 Alice 提交任务后立即以相等的速率从 Alice 和 Bob 向工作人员提供消息。

我已经在另一个上下文中解决了这个问题，方法是为每个用户（甚至用户提交的每个批次）创建多个队列（子队列），并在消费者请求下一条消息时在所有子队列之间交替。

至少在我的世界里，这似乎是一个常见问题，我想知道是否有人知道解决它的既定方法。

我没有看到 ActiveMQ 的任何解决方案。我看过 Kafka，它能够在一个主题中循环分区，这可能会奏效。现在，我正在使用 Redis 实现一些东西。

Answer 1

我会推荐 Cadence Workflow 而不是队列，因为它支持长时间运行的操作和开箱即用的状态管理。

在您的情况下，我将为每个用户创建一个工作流实例。每个新任务都将通过信号 API 发送到用户工作流。然后工作流实例会将接收到的任务排队并一一执行。

下面是实现的概要：

public interface SerializedExecutionWorkflow {

    @WorkflowMethod
    void execute();

    @SignalMethod
    void addTask(Task t);
}

public interface TaskProcessorActivity {
    @ActivityMethod
    void process(Task poll);
}

public class SerializedExecutionWorkflowImpl implements SerializedExecutionWorkflow {

    private final Queue<Task> taskQueue = new ArrayDeque<>();
    private final TaskProcesorActivity processor = Workflow.newActivityStub(TaskProcesorActivity.class);

    @Override
    public void execute() {
        while(!taskQueue.isEmpty()) {
            processor.process(taskQueue.poll());
        }
    }

    @Override
    public void addTask(Task t) {
        taskQueue.add(t);
    }
}

然后是通过信号方法将该任务排入工作流的代码：

private void addTask(WorkflowClient cadenceClient, Task task) {
    // Set workflowId to userId
    WorkflowOptions options = new WorkflowOptions.Builder().setWorkflowId(task.getUserId()).build();
    // Use workflow interface stub to start/signal workflow instance
    SerializedExecutionWorkflow workflow = cadenceClient.newWorkflowStub(SerializedExecutionWorkflow.class, options);
    BatchRequest request = cadenceClient.newSignalWithStartRequest();
    request.add(workflow::execute);
    request.add(workflow::addTask, task);
    cadenceClient.signalWithStart(request);
}

与使用队列进行任务处理相比，Cadence 提供了许多其他优势。

• 内置指数重试，无限期间隔
• 故障处理。例如，如果在配置的时间间隔内两次更新都无法成功，它允许执行通知另一个服务的任务。
• 支持长时间运行的心跳操作
• 能够实现复杂的任务依赖。例如，在发生不可恢复的故障时实现调用链或补偿逻辑 (SAGA)
• 提供对当前更新状态的完整可见性。例如，当使用队列时，您知道队列中是否有一些消息，并且您需要额外的数据库来跟踪整体进度。使用 Cadence 记录每个事件。
• 能够取消正在进行的更新。

请参阅 the presentation，了解 Cadence 编程模型。