RabbitMQ 上的去中心化编排

基于RabbitMQ的分布式流程编排：Apache Camel的替代方案

一个处理来自FTP文件和Kafka/RabbitMQ流的大数据量的系统采用了微服务架构，包含40多个服务。此前，处理链通过Apache Camel的Java DSL实现，由中心服务发起REST调用。存在的问题包括重启时状态丢失、并行流导致的扩展限制以及发布过程中的风险。

新的RabbitMQ库支持仲裁队列，实现了去中心化管理。状态保存在代理中，流程可自动恢复。初始化服务可水平扩展。

核心库实体

该库包含用于步骤、路由和执行器的类和接口：

• RouteStepInfo：标识路由中的步骤。

public class RouteStepInfo {
    private final String routeName;
    private final String stepId;

    public RouteStepInfo(String routeName, String stepId) {
        this.routeName = routeName;
        this.stepId = stepId;
    }
}

• StepExecutor：异步/同步/分离执行的接口。

public interface StepExecutor {
    String getStepName();
    void runAsync(RouteStepInfo routeStepInfo, Long objectId);
    void runSync(RouteStepInfo routeStepInfo, Long objectId);
    void runDetached(RouteStepInfo routeStepInfo, Long objectId);
    void appendListener(StepExecutionFinishedListener listener);
}

• StepExecutionFinishedListener：步骤完成时的回调接口。

public interface StepExecutionFinishedListener {
    void onStepExecutionFinished(RouteStepInfo routeStepInfo, Long objectId);
}

• RouteStep：支持ASYNC、SYNC、DETACHED模式的步骤。

public interface RouteStep {
    String getRouteName();
    String getStepId();
    String getStepName();
    ExecutionMode getExecutionMode();
    void run(Long objectId);
    void appendListener(StepExecutionFinishedListener listener);

    enum ExecutionMode {
        ASYNC,
        SYNC,
        DETACHED
    }
}

• ExecutionRoute：作为步骤列表及其索引的路由。

@Slf4j
public class ExecutionRoute {
    @Getter
    private final String routeName;
    private final List<RouteStep> routeSteps;
    private final Map<String, Integer> stepsIndex;

    public ExecutionRoute(String routeName, List<RouteStep> routeSteps) {
        this.routeName = routeName;
        this.routeSteps = routeSteps;
        this.stepsIndex = new HashMap<>();
        for (int i = 0; i < routeSteps.size(); i++) {
            stepsIndex.put(routeSteps.get(i).getStepId(), i);
        }
    }
}

此外还包括：用于配置的RoutesConfigBuilder、用于路由的RouterConfigProvider、作为监听器和执行器的ProcessRouter。

步骤执行模式

该库支持三种模式：

• ASYNC：顺序后台启动，完成后再进入下一步。
• SYNC：阻塞调用，整个链完成后返回。
• DETACHED：启动后不等待，立即前进。

组合模式支持分支和并行处理：

• 路由1：SYNC步骤1→2→3。
• 路由2：ASYNC步骤1→2→3，立即返回。
• 路由3：ASYNC与DETACHED步骤2。
• 路由5：并行子路由路由4。

实施与迁移

负载测试确认了功能。迁移分阶段进行：通过A/B切换替换Camel部分。保留现有依赖以便回滚。

关键点：

• RabbitMQ中的状态确保重启后恢复。
• 初始化器可水平扩展。
• 通过构建器实现透明配置。
• 支持分支和并行处理，无需中心服务。

此方法适用于高可靠性和可扩展性要求的系统。

— Editorial Team