Azure-Persistenzfunktionen: Verarbeitungslisten

Frageninhalt

Ich habe eine in Python geschriebene Azure-Persistent-Funktion mit einem Koordinator und zwei aktiven Funktionen

orchestrator ruft die erste aktive Funktion auf und erhält im Gegenzug eine Listenvariable (die Namensliste und diese Liste können bei jeder Ausführung der Funktion dynamisch sein)

Der nächste Schritt besteht darin, die zweite Aktivitätsfunktion für jedes Listenelement aufzurufen (sequentielle Verarbeitung – aufgrund von API-Einschränkungen des zweiten Aktivitätsfunktionsaufrufs)

#dynamically gets generated by the first activity function
payload=[1,2,3,4]            

tasks = [context.call_activity("secondfunction",ps) for ps in payload]
output = yield context.task_all(tasks)

Ich verwende bei der Fanout-Methode etwas anderes als seriell, kann aber anscheinend keine Alternative zu dem finden, was ich versuche.

Außerdem habe ich in der Datei host.json versucht zu erzwingen, dass jeweils nur eine aktive Funktion ausgeführt werden kann, um eine Parallelverarbeitung zu vermeiden

"extensions": {
    "durableTask": {
      "maxConcurrentActivityFunctions": 1,
      "maxConcurrentOrchestratorFunctions": 1
    }
  }

Erwähnenswert ist auch, dass ich nicht die gesamte Liste an die Aktivitätsfunktion übergeben kann, da die Ausführung der Aktivitätsfunktion mehr als 5–10 Minuten dauert die Listenorchestrierungsfunktion

Aber die Ergebnisse sind nicht kontinuierlich

Vielen Dank für Ihr Feedback

Richtige Antwort

Sie können versuchen, die folgenden zwei Methoden zu verwenden, um Ihre Anforderung zu erfüllen:-

Methode 1:-

Meine function_app.py:-

import azure.functions as func
import azure.durable_functions as df

myapp = df.dfapp(http_auth_level=func.authlevel.anonymous)

# http starter
@myapp.route(route="orchestrators/{functionname}")
@myapp.durable_client_input(client_name="client")
async def http_start(req: func.httprequest, client):
    function_name = req.route_params.get('functionname')
    instance_id = await client.start_new(function_name, none)  # pass the functionname here
    response = client.create_check_status_response(req, instance_id)
    return response

# orchestrator
@myapp.orchestration_trigger(context_name="context")
def hello_orchestrator(context):
    cities = ["seattle", "tokyo", "london"]

    tasks = []
    for city in cities:
        tasks.append(context.call_activity("hello", city))

    # wait for all tasks to complete
    results = yield context.task_all(tasks)

    return results

# activity
@myapp.activity_trigger(input_name="city")
def hello(city: str):
    print(f"processing {city}...")
    # your activity function logic goes here
    result = f"hello {city}!"

    return result

Ausgabe:-

Funktions-URL:-

http://localhost:7071/api/orchestrators/hello_orchestrator

Methode 2:-

function_app.py:-

import azure.functions as func
import azure.durable_functions as df

myApp = df.DFApp(http_auth_level=func.AuthLevel.ANONYMOUS)

# HTTP Starter
@myApp.route(route="orchestrators/{functionName}")
@myApp.durable_client_input(client_name="client")
async def http_start(req: func.HttpRequest, client):
    function_name = req.route_params.get('functionName')
    instance_id = await client.start_new(function_name, None)  # Pass the functionName here
    response = client.create_check_status_response(req, instance_id)
    return response

# Orchestrator
@myApp.orchestration_trigger(context_name="context")
def hello_orchestrator(context):
    # Call the first activity to get a list of names
    names_list = yield context.call_activity("get_names")

    # Process each name sequentially using the second activity
    results = []
    for name in names_list:
        result = yield context.call_activity("process_name", name)
        results.append(result)

    return results

# First Activity
@myApp.activity_trigger
def get_names():
    # Your logic to retrieve a dynamic list of names goes here
    # For demonstration purposes, returning a hardcoded list
    return ["John", "Alice", "Bob"]

# Second Activity
@myApp.activity_trigger(input_name="name")
def process_name(name: str):
    print(f"Processing {name}...")
    # Your logic to process each name goes here
    result = f"Hello {name}!"

    return result

Das obige ist der detaillierte Inhalt vonAzure-Persistenzfunktionen: Verarbeitungslisten. Für weitere Informationen folgen Sie bitte anderen verwandten Artikeln auf der PHP chinesischen Website!