Zdecentralizowana orkiestracja na RabbitMQ

Zdecentralizowana orkiestracja procesów na RabbitMQ: alternatywa dla Apache Camel

System przetwarzający duże ilości danych z plików FTP i strumieni Kafka/RabbitMQ wykorzystuje architekturę mikrousług z ponad 40 serwisami. Wcześniej łańcuchy przetwarzania były realizowane w Apache Camel za pomocą Java DSL z wywołaniami REST z centralnego serwisu. Problemy: utrata stanu przy restartach, ograniczenia skalowania z powodu równoległych strumieni i ryzyka przy wydaniach.

Nowa biblioteka na RabbitMQ z kolejkami kworum zapewnia zdecentralizowane zarządzanie. Stan jest zachowywany w brokerze, proces wznawia się automatycznie. Serwis inicjalizujący skaluje się horyzontalnie.

Główne encje biblioteki

Biblioteka zawiera klasy i interfejsy dla kroków, tras i wykonawców:

• RouteStepInfo: identyfikuje krok w trasie.

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

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

• StepExecutor: interfejs dla wykonania async/sync/detached.

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: callback na zakończenie kroku.

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

• RouteStep: krok z trybami 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: trasa jako lista kroków z indeksami.

@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);
        }
    }
}

Dodatkowo: RoutesConfigBuilder do konfiguracji, RouterConfigProvider do tras, ProcessRouter jako słuchacz i wykonawca.

Tryby wykonania kroków

Biblioteka obsługuje trzy tryby:

• ASYNC: sekwencyjne uruchomienie w tle, przejście do następnego po zakończeniu.
• SYNC: blokujące wywołanie, powrót po całym łańcuchu.
• DETACHED: uruchomienie bez oczekiwania, natychmiastowe przejście dalej.

Kombinacje umożliwiają rozgałęzienia i równoległość:

• Trasa 1: kroki SYNC 1→2→3.
• Trasa 2: kroki ASYNC 1→2→3, powrót natychmiast.
• Trasa 3: ASYNC z krokiem DETACHED 2.
• Trasa 5: równoległa podtrasa Trasa 4.

Wdrożenie i migracja

Testy obciążeniowe potwierdziły funkcjonalność. Migracja etapowa: zastępowanie części Camel z przełączaniem A/B. Istniejące zależności są zachowane do wycofania.

Co jest ważne:

• Stan w RabbitMQ zapewnia wznowienie po restartach.
• Horyzontalne skalowanie inicjalizatora.
• Przejrzysta konfiguracja przez builder.
• Obsługa rozgałęzień i równoległości bez centralnego serwisu.

Podejście nadaje się dla systemów z wysokimi wymaganiami niezawodności i skali.

— Editorial Team