Opnieuw instelbare uitvoerders

Overzicht

Uitvoerders in werkstromen zijn vaak stateful, ze kunnen bijvoorbeeld berichten verzamelen, aantal beurten bijhouden of tussenliggende resultaten in de cache opslaan. Wanneer een werkstroom opnieuw wordt gebruikt voor meerdere uitvoeringen met gedeelde uitvoerders, kan achtergebleven toestand van een vorige uitvoering doorsijpelen naar volgende uitvoeringen, wat onverwacht gedrag of gegevensbeschadiging veroorzaakt.

De IResettableExecutor interface lost dit probleem op door een contract aan te bieden waarmee executors hun interne toestand tussen sessies kunnen wissen. De runtime van de werkstroom roept ResetAsync() automatisch aan op gedeelde uitvoerexemplaren wanneer een uitvoering is voltooid, zodat er een schone start voor de volgende uitvoering ontstaat.

Het probleem

Overweeg een uitvoerder die berichten verzamelt tijdens een werkstroomuitvoering:

internal sealed partial class AggregationExecutor() : Executor("AggregationExecutor")
{
    private readonly List<string> _messages = [];

    [MessageHandler]
    private async ValueTask HandleAsync(string message, IWorkflowContext context)
    {
        this._messages.Add(message);
        // Process aggregated messages...
    }
}

Als deze uitvoerder wordt gedeeld tussen werkstroomuitvoeringen, _messages behoudt gegevens uit de vorige uitvoering. Tijdens de tweede uitvoering worden verouderde berichten weergegeven die er niet bij horen.

De interface IResettableExecutor

IResettableExecutor definieert één methode die door de werkstroomruntime tussen de uitvoeringen wordt aangeroepen:

public interface IResettableExecutor
{
    ValueTask ResetAsync();
}

Wanneer een uitvoerder deze interface implementeert, kan de runtime deze na elke uitvoering veilig opnieuw instellen, zodat de werkstroom opnieuw kan worden gebruikt zonder verlopen status.

IResettableExecutor implementeren

Om een toestandsvolle uitvoerder opnieuw instelbaar te maken, implementeert u de interface en wist u alle veranderlijke statussen in ResetAsync().

internal sealed partial class AggregationExecutor()
    : Executor("AggregationExecutor"), IResettableExecutor
{
    private readonly List<string> _messages = [];

    [MessageHandler]
    private async ValueTask HandleAsync(string message, IWorkflowContext context)
    {
        this._messages.Add(message);
        // Process aggregated messages...
    }

    public ValueTask ResetAsync()
    {
        this._messages.Clear();
        return default;
    }
}

Zie het voorbeeld workflowAsAnAgent voor een volledig werkvoorbeeld van een werkstroom die gebruikmaakt van herstelbare uitvoerders.

Wanneer moet ik implementeren?

Niet alle uitvoerders moeten implementeren IResettableExecutor. Gebruik deze handleiding voor beslissingen:

Scenario	Implementeren?	Reden
De uitvoerder heeft een onveranderbare status (lijsten, tellers, caches) en wordt gedeeld tussen uitvoeringen	Ja	De toestand van de ene uitvoering kan overslaan naar de volgende
Executor is staatloos	No	Niets om opnieuw in te stellen
De uitvoerende instantie wordt voor elke werkstroom opnieuw gemaakt (via een factory-methode)	No	Elke uitvoering krijgt een nieuw exemplaar met een schone status
De uitvoerder wordt gedeclareerd als over meerdere runs deelbaar (declareCrossRunShareable: true)	No	Deelbare uitvoerders voor meerdere uitvoeringen ondersteunen gelijktijdig gebruik zonder opnieuw in te zetten

Waarschuwing

Als een gedeelde stateful executor niet implementeert IResettableExecutor, genereert hergebruik van de werkstroom een InvalidOperationException:

"Cannot reuse Workflow with shared Executor instances that do not implement IResettableExecutor."

Hoe de runtime deze gebruikt

De werkstroomruntime beheert de levenscyclus voor opnieuw instellen automatisch. Je hoeft ResetAsync() niet zelf aan te roepen. De volgorde is:

1. Eigendom wordt verworven — wanneer een workflow-uitvoering wordt gestart, wordt de runtime eigenaar van het workflow-exemplaar en wordt geconstateerd welke uitvoerders moeten worden gereset.
2. Uitvoeringen worden uitgevoerd : uitvoerders verwerken berichten en kunnen de status verzamelen.
3. Eigendom vrijgegeven — wanneer de run is voltooid (of wordt verwijderd), geeft de runtime het eigendom vrij en worden alle gedeelde uitvoerderinstanties die IResettableExecutor implementeren aangeroepen met ResetAsync().
4. Gereed voor hergebruik : na een geslaagde reset kan de werkstroom worden gebruikt voor een nieuwe uitvoering.

Als een gedeelde uitvoerder niet opnieuw kan worden ingesteld (omdat de interface niet wordt geïmplementeerd), wordt de werkstroom gemarkeerd als niet-herbruikbaar en worden volgende uitvoeringen uitgevoerd.

Relatie met statusisolatie

IResettableExecutor vormt een aanvulling op het helper-methodepatroon dat wordt beschreven in State Management. De twee benaderingen dienen verschillende behoeften:

• Helpermethoden (het maken van nieuwe exemplaren per uitvoering) bieden de sterkste isolatiegaranties en worden aanbevolen als standaardbenadering.
• IResettableExecutor is handig in situaties waarin u uitvoerdersexemplaren wilt delen tussen verschillende uitvoeringen, bijvoorbeeld wanneer het maken van de uitvoerder veel kost of wanneer een werkstroom als agent wordt weergegeven en hergebruikt in meerdere aanroepen.

Kies de methode die het beste past bij uw scenario. Voor de meeste werkstromen zijn helpermethoden voldoende. Gebruik IResettableExecutor wanneer het delen van instanties een bewuste ontwerpkeuze is.

Dit concept is niet van toepassing op Python. Voor volledige statusisolatie bouwt u nieuwe workflow- en uitvoeringsinstanties voor iedere afzonderlijke uitvoering. Zie State Management voor patronen en voorbeelden.

Volgende stappen 

Statusbeheer