Skicka händelser till eller ta emot händelser från händelsehubbar med hjälp av Python

Den här snabbstarten visar hur du skickar händelser till och tar emot händelser från en händelsehubb med azure-eventhub Python-paketet.

Förutsättningar

Om du inte har använt Azure Event Hubs tidigare kan du läsa Översikt över Event Hubs innan du gör den här snabbstarten.

För att slutföra den här snabbstarten, behöver du följande förhandskrav:

• Microsoft Azure-prenumeration. Om du vill använda Azure-tjänster, inklusive Azure Event Hubs, behöver du en prenumeration. Om du inte har ett befintligt Azure-konto registrerar du dig för en kostnadsfri utvärderingsversion.
• Python 3.8 eller senare, med pip installerat och uppdaterat.
• Visual Studio Code (rekommenderas) eller någon annan integrerad utvecklingsmiljö (IDE).
• Skapa ett Event Hubs-namnområde och en händelsehubb. Det första steget är att använda Azure Portal för att skapa ett Event Hubs-namnområde och hämta de hanteringsuppgifter som programmet behöver för att kommunicera med händelsehubben. Om du behöver skapa ett namnområde och en händelsehubb följer du anvisningarna i den här artikeln.

Installera paketen för att skicka händelser

Om du vill installera Python-paketen för Event Hubs öppnar du en kommandotolk som har Python i sin sökväg. Ändra katalogen till den mapp där du vill behålla dina exempel.

pip install azure-eventhub
pip install azure-identity
pip install aiohttp

pip install azure-eventhub

Autentisera appen till Azure

Den här snabbstarten visar två sätt att ansluta till Azure Event Hubs: lösenordslösa och anslutningssträng. Det första alternativet visar hur du använder ditt säkerhetsobjekt i Microsoft Entra-ID och rollbaserad åtkomstkontroll (RBAC) för att ansluta till ett Event Hubs-namnområde. Du behöver inte bekymra dig om att ha hårdkodade anslutningssträng i koden eller i en konfigurationsfil eller i en säker lagring som Azure Key Vault. Det andra alternativet visar hur du använder en anslutningssträng för att ansluta till ett Event Hubs-namnområde. Om du inte har använt Azure tidigare kan det vara lättare att följa anslutningssträng alternativet. Vi rekommenderar att du använder det lösenordslösa alternativet i verkliga program och produktionsmiljöer. Mer information finns i Autentisering och auktorisering. Du kan också läsa mer om lösenordslös autentisering på översiktssidan.

az eventhubs namespace show -g '<your-event-hub-resource-group>' -n '<your-event-hub-name> --query id

az role assignment create --assignee "<user@domain>" \
--role "Azure Event Hubs Data Owner" \
--scope "<your-resource-id>"

Get-AzResource -ResourceGroupName "<your-event-hub-resource-group>" -Name "<your-event-hub-name>"

New-AzRoleAssignment -SignInName <user@domain> `
-RoleDefinitionName "Azure Event Hubs Data Owner" `
-Scope <yourResourceId>

Skicka händelser

I det här avsnittet skapar du ett Python-skript för att skicka händelser till händelsehubben som du skapade tidigare.

1. Öppna din favoritredigerare för Python, till exempel Visual Studio Code.

2. Skapa ett skript med namnet send.py. Det här skriptet skickar en batch händelser till den händelsehubb som du skapade tidigare.

3. Klistra in följande kod i send.py:

   • Lösenordsfri (rekommenderas)
   • Anslutningssträng

   I koden använder du verkliga värden för att ersätta följande platshållare:

   • EVENT_HUB_FULLY_QUALIFIED_NAMESPACE
   • EVENT_HUB_NAME

   import asyncio
   
   from azure.eventhub import EventData
   from azure.eventhub.aio import EventHubProducerClient
   from azure.identity.aio import DefaultAzureCredential
   
   EVENT_HUB_FULLY_QUALIFIED_NAMESPACE = "EVENT_HUB_FULLY_QUALIFIED_NAMESPACE"
   EVENT_HUB_NAME = "EVENT_HUB_NAME"
   
   credential = DefaultAzureCredential()
   
   async def run():
       # Create a producer client to send messages to the event hub.
       # Specify a credential that has correct role assigned to access
       # event hubs namespace and the event hub name.
       producer = EventHubProducerClient(
           fully_qualified_namespace=EVENT_HUB_FULLY_QUALIFIED_NAMESPACE,
           eventhub_name=EVENT_HUB_NAME,
           credential=credential,
       )
       async with producer:
           # Create a batch.
           event_data_batch = await producer.create_batch()
   
           # Add events to the batch.
           event_data_batch.add(EventData("First event "))
           event_data_batch.add(EventData("Second event"))
           event_data_batch.add(EventData("Third event"))
   
           # Send the batch of events to the event hub.
           await producer.send_batch(event_data_batch)
   
           # Close credential when no longer needed.
           await credential.close()
   
   asyncio.run(run())
   

   Kommentar

   Exempel på andra alternativ för att skicka händelser till händelsehubben asynkront med hjälp av en anslutningssträng finns på sidan GitHub send_async.py. De mönster som visas där gäller även för att skicka händelser utan lösenord.

Ta emot händelser

Den här snabbstarten använder Azure Blob Storage som kontrollpunktslager. Kontrollpunktsarkivet används för att bevara kontrollpunkter (d.s. de senast lästa positionerna).

Följ dessa rekommendationer när du använder Azure Blob Storage som kontrollpunktslager:

• Använd en separat container för varje konsumentgrupp. Du kan använda samma lagringskonto, men använda en container per grupp.
• Använd inte containern för något annat och använd inte lagringskontot för något annat.
• Lagringskontot ska finnas i samma region som det distribuerade programmet finns i. Om programmet är lokalt kan du försöka välja den region som är närmast.

På sidan Lagringskonto i Azure Portal i avsnittet Blob Service kontrollerar du att följande inställningar är inaktiverade.

• Hierarkisk namnrymd
• Mjuk borttagning av blob
• Versionshantering

Skapa ett Azure Storage-konto och en blobcontainer

Skapa ett Azure Storage-konto och en blobcontainer i det genom att göra följande:

1. Skapa ett Azure Storage-konto
2. Skapa en blobcontainer.
3. Autentisera till blobcontainern.

Se till att registrera anslutningssträng och containernamnet för senare användning i mottagarkoden.

az storage account show --resource-group '<your-resource-group-name>' --name '<your-storage-account-name>' --query id

az role assignment create --assignee "<user@domain>" \
    --role "Storage Blob Data Contributor" \
    --scope "<your-resource-id>"

Get-AzResource -ResourceGroupName "<yourResourceGroupname>" -Name "<yourStorageAccountName>"

New-AzRoleAssignment -SignInName <user@domain> `
    -RoleDefinitionName "Storage Blob Data Contributor" `
    -Scope <yourStorageAccountId>

Installera paketen för att ta emot händelser

För mottagarsidan måste du installera ett eller flera paket. I den här snabbstarten använder du Azure Blob Storage för att spara kontrollpunkter så att programmet inte läser de händelser som det redan har läst. Den utför metadatakontrollpunkter på mottagna meddelanden med jämna mellanrum i en blob. Den här metoden gör det enkelt att fortsätta ta emot meddelanden senare där du slutade.

pip install azure-eventhub-checkpointstoreblob-aio
pip install azure-identity

pip install azure-eventhub-checkpointstoreblob-aio

Skapa ett Python-skript för att ta emot händelser

I det här avsnittet skapar du ett Python-skript för att ta emot händelser från händelsehubben:

1. Öppna din favoritredigerare för Python, till exempel Visual Studio Code.

2. Skapa ett skript med namnet recv.py.

3. Klistra in följande kod i recv.py:

   • Lösenordsfri (rekommenderas)
   • Anslutningssträng

   I koden använder du verkliga värden för att ersätta följande platshållare:

   • BLOB_STORAGE_ACCOUNT_URL
   • BLOB_CONTAINER_NAME
   • EVENT_HUB_FULLY_QUALIFIED_NAMESPACE
   • EVENT_HUB_NAME

   import asyncio
   
   from azure.eventhub.aio import EventHubConsumerClient
   from azure.eventhub.extensions.checkpointstoreblobaio import (
       BlobCheckpointStore,
   )
   from azure.identity.aio import DefaultAzureCredential
   
   BLOB_STORAGE_ACCOUNT_URL = "BLOB_STORAGE_ACCOUNT_URL"
   BLOB_CONTAINER_NAME = "BLOB_CONTAINER_NAME"
   EVENT_HUB_FULLY_QUALIFIED_NAMESPACE = "EVENT_HUB_FULLY_QUALIFIED_NAMESPACE"
   EVENT_HUB_NAME = "EVENT_HUB_NAME"
   
   credential = DefaultAzureCredential()
   
   async def on_event(partition_context, event):
       # Print the event data.
       print(
           'Received the event: "{}" from the partition with ID: "{}"'.format(
               event.body_as_str(encoding="UTF-8"), partition_context.partition_id
           )
       )
   
       # Update the checkpoint so that the program doesn't read the events
       # that it has already read when you run it next time.
       await partition_context.update_checkpoint(event)
   
   
   async def main():
       # Create an Azure blob checkpoint store to store the checkpoints.
       checkpoint_store = BlobCheckpointStore(
           blob_account_url=BLOB_STORAGE_ACCOUNT_URL,
           container_name=BLOB_CONTAINER_NAME,
           credential=credential,
       )
   
       # Create a consumer client for the event hub.
       client = EventHubConsumerClient(
           fully_qualified_namespace=EVENT_HUB_FULLY_QUALIFIED_NAMESPACE,
           eventhub_name=EVENT_HUB_NAME,
           consumer_group="$Default",
           checkpoint_store=checkpoint_store,
           credential=credential,
       )
       async with client:
           # Call the receive method. Read from the beginning of the partition
           # (starting_position: "-1")
           await client.receive(on_event=on_event, starting_position="-1")
   
       # Close credential when no longer needed.
       await credential.close()
   
   if __name__ == "__main__":
       # Run the main method.
       asyncio.run(main())
   

   Kommentar

   Exempel på andra alternativ för att ta emot händelser från Händelsehubb asynkront med hjälp av en anslutningssträng finns på sidan GitHub recv_with_checkpoint_store_async.py. De mönster som visas där gäller även för att ta emot händelser utan lösenord.

Kör mottagarappen

Om du vill köra skriptet öppnar du en kommandotolk som har Python i sökvägen och kör sedan det här kommandot:

python recv.py

Kör avsändarappen

Om du vill köra skriptet öppnar du en kommandotolk som har Python i sökvägen och kör sedan det här kommandot:

python send.py

Mottagarfönstret ska visa de meddelanden som skickades till händelsehubben.

Felsökning

Om du inte ser händelser i mottagarfönstret eller om koden rapporterar ett fel kan du prova följande felsökningstips:

• Om du inte ser resultat från recy.py kör du send.py flera gånger.

• Om du ser fel om "coroutine" när du använder den lösenordslösa koden (med autentiseringsuppgifter) kontrollerar du att du använder import från azure.identity.aio.

• Om du ser "Avsluten klientsession" med lösenordslös kod (med autentiseringsuppgifter) kontrollerar du att du stänger autentiseringsuppgifterna när du är klar. Mer information finns i Async-autentiseringsuppgifter.

• Om du ser auktoriseringsfel med recv.py när du kommer åt lagringen kontrollerar du att du har följt stegen i Skapa ett Azure-lagringskonto och en blobcontainer och tilldelat rollen Storage Blob Data Contributor till tjänstens huvudnamn.

• Om du får händelser med olika partitions-ID:t förväntas det här resultatet. Partitioner är en mekanism för organisering av data som har att göra med vilken underordnad parallellitet som krävs i de program som används. Antalet partitioner i en händelsehubb är direkt kopplat till antalet samtidiga läsare som du förväntar dig. Mer information finns i Läs mer om partitioner.

Nästa steg

I den här snabbstarten har du skickat och tagit emot händelser asynkront. Om du vill veta hur du skickar och tar emot händelser synkront går du till GitHub-sync_samples-sidan.

För alla exempel (både synkrona och asynkrona) på GitHub går du till Azure Event Hubs-klientbiblioteket för Python-exempel.