Szybki start: przetwarzanie zdarzeń w czasie rzeczywistym przy użyciu usługi Azure Functions

W tym artykule użyjesz interfejsu wiersza polecenia dla deweloperów platformy Azure (azd), aby utworzyć funkcję wyzwalacza usługi Event Hubs na potrzeby przetwarzania zdarzeń w czasie rzeczywistym w usłudze Azure Functions. Po zweryfikowaniu kodu lokalnie należy wdrożyć go w nowej aplikacji funkcji bezserwerowej uruchomionej w planie Flex Consumption na platformie Azure.

Źródło projektu używa azd do tworzenia aplikacji funkcji i powiązanych zasobów oraz wdrażania kodu w Azure. To wdrożenie jest zgodne z bieżącymi najlepszymi rozwiązaniami dotyczącymi bezpiecznych i skalowalnych wdrożeń Azure Functions.

Domyślnie plan Flex Consumption stosuje model rozliczeń płatności za to, co wykorzystujesz, co pozwala na ukończenie tego artykułu i poniesienie tylko niewielkiego kosztu kilku centów amerykańskich lub mniej na koncie platformy Azure.

Ten artykuł obsługuje wersję 4 modelu programowania Node.js dla usługi Azure Functions.

Ten artykuł obsługuje wersję 2 modelu programowania Python dla Azure Functions.

Wymagania wstępne

• SDK .NET 8

• zestaw deweloperów Java 21
  • Ustaw zmienną środowiskową JAVA_HOME na lokalizację instalacji poprawnej wersji zestawu Java Development Kit (JDK).
• Apache Maven 3.8.x

• Node.js 22 lub nowsze

• PowerShell 7.4

• Python 3.11 lub nowszy

• Emulator magazynu Azurite

• Interfejs wiersza polecenia dla deweloperów platformy Azure

• Podstawowe narzędzia usługi Azure Functions

• Interfejs wiersza polecenia platformy Azure

• Konto platformy Azure z aktywną subskrypcją. Utwórz bezpłatne konto.

Inicjowanie projektu

Użyj polecenia , azd init aby utworzyć lokalny projekt kodu usługi Azure Functions na podstawie szablonu.

W lokalnym terminalu lub wierszu polecenia uruchom to azd init polecenie w pustym folderze:

azd init --template functions-quickstart-dotnet-azd-eventhub -e eventhub-dotnet

To polecenie ściąga pliki projektu z repozytorium szablonów i inicjuje projekt w bieżącym folderze. Flaga -e ustawia nazwę bieżącego środowiska. W azd środowisko systemowe zapewnia unikatowy kontekst wdrażania dla twojej aplikacji i można zdefiniować więcej niż jeden z nich. Nazwa środowiska jest również używana w nazwie grupy zasobów utworzonej w Azure.

W lokalnym terminalu lub wierszu polecenia uruchom to azd init polecenie w pustym folderze:

azd init --template functions-quickstart-typescript-azd-eventhub -e eventhub-ts

To polecenie ściąga pliki projektu z repozytorium szablonów i inicjuje projekt w bieżącym folderze. Flaga -e ustawia nazwę bieżącego środowiska. W azd środowisko systemowe zapewnia unikatowy kontekst wdrażania dla twojej aplikacji i można zdefiniować więcej niż jeden z nich. Nazwa środowiska jest również używana w nazwie grupy zasobów utworzonej w Azure.

W lokalnym terminalu lub wierszu polecenia uruchom to azd init polecenie w pustym folderze:

azd init --template functions-quickstart-python-azd-eventhub -e eventhub-py

To polecenie ściąga pliki projektu z repozytorium szablonów i inicjuje projekt w bieżącym folderze. Flaga -e ustawia nazwę bieżącego środowiska. W azd środowisko systemowe zapewnia unikatowy kontekst wdrażania dla twojej aplikacji i można zdefiniować więcej niż jeden z nich. Nazwa środowiska jest również używana w nazwie grupy zasobów utworzonej w Azure.

Tworzenie i aktywowanie środowiska wirtualnego

W folderze głównym uruchom następujące polecenia, aby utworzyć i aktywować środowisko wirtualne o nazwie .venv:

Linux/macOS

python3 -m venv .venv
source .venv/bin/activate

Jeśli Python nie instaluje pakietu venv w dystrybucji systemu Linux, uruchom następujące polecenie:

sudo apt-get install python3-venv

Windows (bash)

py -m venv .venv
source .venv/scripts/activate

Windows (cmd)

py -m venv .venv
.venv\scripts\activate

W lokalnym terminalu lub wierszu polecenia uruchom to azd init polecenie w pustym folderze:

azd init --template functions-quickstart-java-azd-eventhub -e eventhub-java

To polecenie ściąga pliki projektu z repozytorium szablonów i inicjuje projekt w bieżącym folderze. Flaga -e ustawia nazwę bieżącego środowiska. W azd środowisko systemowe zapewnia unikatowy kontekst wdrażania dla twojej aplikacji i można zdefiniować więcej niż jeden z nich. Nazwa środowiska jest również używana w nazwie grupy zasobów utworzonej w Azure.

W lokalnym terminalu lub wierszu polecenia uruchom to azd init polecenie w pustym folderze:

azd init --template functions-quickstart-javascript-azd-eventhub -e eventhub-js

To polecenie ściąga pliki projektu z repozytorium szablonów i inicjuje projekt w bieżącym folderze. Flaga -e ustawia nazwę bieżącego środowiska. W azd środowisko systemowe zapewnia unikatowy kontekst wdrażania dla twojej aplikacji i można zdefiniować więcej niż jeden z nich. Nazwa środowiska jest również używana w nazwie grupy zasobów utworzonej w Azure.

W lokalnym terminalu lub wierszu polecenia uruchom to azd init polecenie w pustym folderze:

azd init --template functions-quickstart-powershell-azd-eventhub -e eventhub-ps

To polecenie ściąga pliki projektu z repozytorium szablonów i inicjuje projekt w bieżącym folderze. Flaga -e ustawia nazwę bieżącego środowiska. W azd środowisko systemowe zapewnia unikatowy kontekst wdrażania dla twojej aplikacji i można zdefiniować więcej niż jeden z nich. Nazwa środowiska jest również używana w nazwie grupy zasobów utworzonej w Azure.

Tworzenie zasobów platformy Azure

Zanim będzie można uruchomić funkcję lokalnie, musisz utworzyć przestrzeń nazw i centrum usługi Event Hubs na platformie Azure. Użyj azd provision polecenia , aby utworzyć te zasoby i skonfigurować ustawienia lokalne przez dodanie wymaganego pliku local.settings.json .

1. Uruchom następujące polecenia, aby zalogować się na platformie Azure:

   azd auth login
   

   Postępuj zgodnie z monitami, aby uwierzytelnić się przy użyciu konta platformy Azure.

2. W folderze głównym uruchom następujące polecenie, aby utworzyć zasoby platformy Azure:

   azd provision
   

3. Po wyświetleniu monitu podaj następujące wymagane parametry wdrożenia:

   Parametr	Opis
   Subskrypcja platformy Azure	Subskrypcja, w której tworzysz zasoby.
   Lokalizacja platformy Azure	Region platformy Azure, w którym ma zostać utworzona grupa zasobów zawierająca nowe zasoby platformy Azure. Wyświetlane są tylko regiony, które obecnie obsługują plan Flex Consumption.
   vnetEnabled	Użyj wartości , False aby uniknąć dodatkowych obciążeń związanych z tworzeniem zasobów sieci wirtualnej.

   Polecenie azd provision tworzy wymagane zasoby platformy Azure, w tym przestrzeń nazw i centrum w usłudze Event Hubs, aplikację funkcji Flex Consumption, usługę Application Insights i konto magazynowe. Konfiguruje również plik local.settings.json za pomocą informacji o połączeniu usługi Event Hubs.

Uruchamianie w środowisku lokalnym

1. W osobnym oknie terminalu uruchom emulator magazynu Azurite:

   azurite
   

   Lokalny proces hosta usługi Functions używa emulatora Azurite dla wewnętrznego połączenia magazynu (AzureWebJobsStorage) wymaganego przez środowisko uruchomieniowe.

2. Aby uruchomić aplikację funkcji, uruchom następujące polecenia w terminalu lub wierszu polecenia, aby przejść do src folderu projektu i uruchomić aplikację funkcji:

   cd src
   func start
   

2. Aby uruchomić aplikację funkcji, uruchom to polecenie w terminalu lub wierszu polecenia:

   func start
   

2. Aby skompilować i uruchomić aplikację funkcji, uruchom następujące polecenia w terminalu lub wierszu polecenia:

   mvn clean package
   mvn azure-functions:run
   

2. Aby zainstalować zależności i uruchomić aplikację funkcji, uruchom następujące polecenia w terminalu lub wierszu polecenia:

   npm install
   npm start  
   

2. Aby zainstalować zależności i uruchomić aplikację funkcji, uruchom następujące polecenia w terminalu lub wierszu polecenia:

   cd src
   npm install
   npm start  
   

2. Aby uruchomić aplikację funkcji, uruchom to polecenie w terminalu lub wierszu polecenia:

   func start
   

3. Jeśli pojawi się monit, zezwól narzędziom Core Tools (func.exe) na działanie przez zaporę.

4. Gdy host Functions zostanie uruchomiony w lokalnym folderze projektu, loguje informacje o Twoich funkcjach do konsoli.

   Ten przykład zawiera funkcję wyzwalacza czasomierza, która automatycznie generuje artykuły z wiadomościami co 10 sekund i wysyła je do usługi Event Hubs. Funkcja wyzwalająca Event Hubs przetwarza następnie te zdarzenia i wykonuje analizę sentymentu oraz śledzenie zaangażowania.

   Zobaczysz dane wyjściowe podobne do tego przykładu:

    [2026-03-02T22:37:30.151Z] Executing 'Functions.EventHubsTrigger'
    [2026-03-02T22:37:30.159Z] Trigger Details: PartitionId: 24, OffsetString: 0, EnqueueTimeUtc: 2026-03-02T22:37:29.1790000+00:00, SequenceNumber: 0, Count: 1, Offset: 0, PartionId: 24
    [2026-03-02T22:37:30.169Z] ⭐ High-engagement article NEWS-20260302-0580CB82 (Views: 6123, Sentiment: 0.57) featured!
    [2026-03-02T22:37:30.174Z] 🔥 Viral article: NEWS-20260302-0580CB82 - 6,123 views
    [2026-03-02T22:37:30.181Z] 🌟 Featured article: NEWS-20260302-0580CB82
    [2026-03-02T22:37:30.185Z] ✅ Successfully processed article NEWS-20260302-0580CB82 - 'Technology Breakthrough in Renewable Energy Technology' by Sarah Johnson
    [2026-03-02T22:37:30.191Z] 📰 Processed 1 news articles, 0 failed in batch of 1
    [2026-03-02T22:37:30.196Z] 📊 NEWS BATCH SUMMARY: 1 articles | Total Views: 6,123 | Avg Views: 6,123 | Avg Sentiment: 0.57 | Status: [Featured: 1]
    [2026-03-02T22:37:30.200Z] 📂 Top Categories: [Health: 1] | Top Sources: [Innovation Weekly: 1]
    [2026-03-02T22:37:30.204Z] 🔥 Viral articles in batch: 1
    [2026-03-02T22:37:30.207Z] Executed 'Functions.EventHubsTrigger' (Succeeded, Duration=55ms)
    

5. Po zakończeniu naciśnij Ctrl+C w oknie terminalu, aby zatrzymać proces func.exe hosta.

6. Zamknij okno, w którym działa Azurite.

7. Uruchom polecenie deactivate , aby zamknąć środowisko wirtualne.

Przejrzyj kod (opcjonalnie)

Możesz przejrzeć kod definiujący funkcję wyzwalacza usługi Event Hubs:

using System.Text.Json;
using Azure.Messaging.EventHubs;
using Microsoft.Azure.Functions.Worker;
using Microsoft.Extensions.Logging;

namespace function_app;

public class EventHubsTrigger
{
    private readonly ILogger<EventHubsTrigger> _logger;
    private readonly NewsProcessingService _newsService;

    public EventHubsTrigger(ILogger<EventHubsTrigger> logger, NewsProcessingService newsService)
    {
        _logger = logger;
        _newsService = newsService;
    }

    [Function(nameof(EventHubsTrigger))]
    public async Task Run([EventHubTrigger("news", Connection = "EventHubConnection")] EventData[] input)
    {
        var processedArticles = new List<NewsArticle>();
        var failedEvents = 0;
        
        foreach (var message in input)
        {
            try
            {
                var messageBody = message.EventBody.ToString();

                // Parse the news article event
                var article = ParseNewsArticleEvent(messageBody);

                if (article != null)
                {
                    processedArticles.Add(article);
                }
                else
                {
                    failedEvents++;
                }
            }
            catch (Exception ex)
            {
                failedEvents++;
                _logger.LogWarning($"Error processing message: {ex.Message}");
            }
        }

        // Log summary of this execution
        _logger.LogInformation($"📰 Processed {processedArticles.Count} news articles, {failedEvents} failed in batch of {input.Length}");

Możesz przejrzeć kompletny projekt szablonu here.

package com.function;

import com.microsoft.azure.functions.*;
import com.microsoft.azure.functions.annotation.*;

import java.time.Instant;
import java.util.*;
import java.util.logging.Logger;

/**
 * Azure Function that processes messages from an input Event Hub,
 * adds metadata, and sends processed messages to an output Event Hub.
 */
public class EventHubsTriggerFunction {

    @FunctionName("EventHubsTrigger")
    public void run(
            @EventHubTrigger(
                name = "messages",
                eventHubName = "%INPUT_EVENTHUB_NAME%",
                connection = "EventHubConnection",
                cardinality = Cardinality.MANY)
            List<String> messages,
            @EventHubOutput(
                name = "output",
                eventHubName = "%OUTPUT_EVENTHUB_NAME%",
                connection = "EventHubConnection")
            OutputBinding<List<String>> output,
            final ExecutionContext context) {

        Logger logger = context.getLogger();
        logger.info(String.format("🔄 Event hub function processing %d message(s)", messages.size()));

        List<String> processedMessages = new ArrayList<>();

        for (String message : messages) {
            try {
                logger.info("📨 Processing event: " + message);

                // Create processed message with additional metadata
                String processedMessage = String.format(
                    "{\"id\":\"%s\",\"message\":%s,\"timestamp\":\"%s\"}",
                    UUID.randomUUID().toString(),
                    message,
                    Instant.now().toString());

                processedMessages.add(processedMessage);
                logger.info("\u2728 Message processed: " + processedMessage);

            } catch (Exception e) {
                logger.severe("\u274C Error processing message: " + e.getMessage());
            }
        }

        // Send processed messages to output Event Hub
        if (!processedMessages.isEmpty()) {
            output.setValue(processedMessages);
            logger.info(String.format("📤 Sent %d message(s) to output Event Hub", processedMessages.size()));
        }
    }
}

Możesz przejrzeć kompletny projekt szablonu here.

const { app, output } = require("@azure/functions");

const eventHubOutput = output.eventHub({
    connection: 'EventHubConnection',
    eventHubName: '%OUTPUT_EVENTHUB_NAME%'
});

async function EventHubsTrigger(messages, context) {
    context.log(`🔄 Event hub function processing ${messages.length} message(s)`);
    
    const processedMessages = [];
    
    for (const message of messages) {
        try {
            // Parse the incoming message
            const eventData = typeof message === 'string' ? JSON.parse(message) : message;
            context.log('📨 Processing event:', eventData);
            
            // Create processed message with additional metadata
            const processedMessage = {
                id: eventData.id || crypto.randomUUID(),
                message: eventData.message || JSON.stringify(eventData),
                timestamp: new Date().toISOString()
            };
            
            processedMessages.push(processedMessage);
            context.log('✨ Message processed:', processedMessage);
            
        } catch (error) {
            context.error(`❌ Error processing message: ${error}`);
        }
    }
    
    // Send processed messages to output Event Hub
    if (processedMessages.length > 0) {
        context.extraOutputs.set(eventHubOutput, processedMessages);
        context.log(`📤 Sent ${processedMessages.length} message(s) to output Event Hub`);
    }
}

app.eventHub('EventHubsTrigger', {
    connection: 'EventHubConnection',
    eventHubName: '%INPUT_EVENTHUB_NAME%',
    cardinality: 'many',
    extraOutputs: [eventHubOutput],
    handler: EventHubsTrigger
});

Możesz przejrzeć kompletny projekt szablonu here.

import { app, InvocationContext, output } from "@azure/functions";

interface EventMessage {
    id: string;
    message: string;
    timestamp: string;
}

const eventHubOutput = output.eventHub({
    connection: 'EventHubConnection',
    eventHubName: '%OUTPUT_EVENTHUB_NAME%'
});

export async function EventHubsTrigger(messages: unknown[], context: InvocationContext): Promise<void> {
    context.log(`🔄 Event hub function processing ${messages.length} message(s)`);
    
    const processedMessages: EventMessage[] = [];
    
    for (const message of messages) {
        try {
            // Parse the incoming message
            const eventData = typeof message === 'string' ? JSON.parse(message) : message;
            context.log('📨 Processing event:', eventData);
            
            // Create processed message with additional metadata
            const processedMessage: EventMessage = {
                id: eventData.id || crypto.randomUUID(),
                message: eventData.message || JSON.stringify(eventData),
                timestamp: new Date().toISOString()
            };
            
            processedMessages.push(processedMessage);
            context.log('✨ Message processed:', processedMessage);
            
        } catch (error) {
            context.error(`❌ Error processing message: ${error}`);
        }
    }
    
    // Send processed messages to output Event Hub
    if (processedMessages.length > 0) {
        context.extraOutputs.set(eventHubOutput, processedMessages);
        context.log(`📤 Sent ${processedMessages.length} message(s) to output Event Hub`);
    }
}

app.eventHub('EventHubsTrigger', {
    connection: 'EventHubConnection',
    eventHubName: '%INPUT_EVENTHUB_NAME%',
    cardinality: 'many',
    extraOutputs: [eventHubOutput],
    handler: EventHubsTrigger
});

Możesz przejrzeć kompletny projekt szablonu here.

param($InputEvents, $TriggerMetadata)

Write-Host "🔄 Event hub function processing $($InputEvents.Count) message(s)"

$processedMessages = @()

foreach ($message in $InputEvents) {
    try {
        # Parse the incoming message
        $eventData = $message | ConvertFrom-Json -ErrorAction SilentlyContinue
        if (-not $eventData) {
            $eventData = @{ message = $message }
        }

        Write-Host "📨 Processing event: $($eventData | ConvertTo-Json -Compress)"

        # Create processed message with additional metadata
        $processedMessage = @{
            id        = if ($eventData.id) { $eventData.id } else { [guid]::NewGuid().ToString() }
            message   = if ($eventData.message) { $eventData.message } else { ($eventData | ConvertTo-Json -Compress) }
            timestamp = (Get-Date).ToUniversalTime().ToString('o')
        }

        $processedMessages += $processedMessage
        Write-Host "✨ Message processed: $($processedMessage | ConvertTo-Json -Compress)"
    }
    catch {
        Write-Error "❌ Error processing message: $_"
    }
}

# Send processed messages to output Event Hub
if ($processedMessages.Count -gt 0) {
    Push-OutputBinding -Name OutputEvents -Value ($processedMessages | ConvertTo-Json -Compress)
    Write-Host "📤 Sent $($processedMessages.Count) message(s) to output Event Hub"
}

Wyzwalacz jest definiowany w odpowiednich function.json.

Możesz przejrzeć kompletny projekt szablonu here.

import azure.functions as func
import logging
import json
from datetime import datetime, timezone
import random
import uuid
from typing import List

app = func.FunctionApp()

# News article data model
class NewsArticle:
            title=random.choice(titles_templates).format(topic=topic),
            content=f"Comprehensive coverage of the latest developments in {topic}. " * random.randint(10, 20),
            author=random.choice(authors),
            source=random.choice(sources),
            category=random.choice(categories),
            published_date=datetime.now(timezone.utc),
            view_count=random.randint(100, 10000),
            sentiment_score=round(random.uniform(-1.0, 1.0), 2),
            status=random.choice(["Published", "Featured"]),
            tags=[random.choice(topics) for _ in range(random.randint(3, 5))]
        )
        articles.append(article)
    
    # Send articles to Event Hub
    events_json = json.dumps([article.to_dict() for article in articles])
    event.set(events_json)
    
    logging.info(f'✅ HIGH-THROUGHPUT: Successfully generated {num_articles} news articles in ~10 seconds')


# Event Hub trigger - processes news articles
@app.event_hub_message_trigger(arg_name="events", event_hub_name="news",
                                connection="EventHubConnection")
def EventHubsTrigger(events: List[func.EventHubEvent]):
    """Process news articles from Event Hub with sentiment analysis and engagement tracking"""
    
    # Handle both single event and list of events
    if not isinstance(events, list):
        events = [events]
    
    batch_articles = []
    failed_count = 0
    
    for event in events:
        try:
            # Parse the event data
            event_data = json.loads(event.get_body().decode('utf-8'))
            
            # Handle both single article and array of articles
            articles = event_data if isinstance(event_data, list) else [event_data]
            
            for article_data in articles:
                # Process each article
                article_id = article_data.get('ArticleId')
                title = article_data.get('Title')
                author = article_data.get('Author')

Możesz przejrzeć kompletny projekt szablonu here.

Po zweryfikowaniu funkcji lokalnie nadszedł czas, aby opublikować ją na platformie Azure.

Wdrażanie na platformie Azure

Ten projekt jest skonfigurowany do używania polecenia azd up do wdrożenia kodu w nowej aplikacji funkcji w planie Flex Consumption na platformie Azure. Ponieważ zasoby zostały już przydzielone, to polecenie wdraża kod do istniejącej aplikacji funkcji.

Wskazówka

Ten projekt zawiera zestaw plików Bicep, których używa się azd do tworzenia bezpiecznego wdrożenia w planie Flex Consumption, zgodnego z najlepszymi praktykami.

W folderze głównym repozytorium uruchom następujące polecenie, aby wdrożyć projekt kodu w aplikacji funkcji na platformie Azure:

azd deploy

Pakiet wdrożeniowy pakuje Twój kod i wdraża go do aplikacji funkcji. Po zakończeniu działania polecenia zobaczysz linki do utworzonych zasobów.

Weryfikowanie wdrożenia

Po zakończeniu wdrażania funkcja wyzwalacza usługi Event Hubs automatycznie rozpoczyna przetwarzanie zdarzeń po ich nadejściu w centrum zdarzeń.

1. W portalu Azure przejdź do swojej nowej aplikacji funkcji.

2. Wybierz Strumień logów z menu po lewej stronie, aby monitorować wykonywanie funkcji w czasie rzeczywistym.

3. Zobaczysz wpisy dziennika, które pokazują, że funkcja wyzwalacza Event Hubs przetwarza zdarzenia wygenerowane przez wyzwalacz czasomierza.

Ponowne wdrażanie kodu

azd up Uruchom polecenie tyle razy, ile potrzebujesz, aby aprowizować zasoby platformy Azure i wdrażać aktualizacje kodu w aplikacji funkcji.

Uwaga / Notatka

Wdrożone pliki kodu są zawsze zastępowane przez najnowszy pakiet wdrożeniowy.

Początkowe odpowiedzi na azd monity i wszystkie zmienne środowiskowe wygenerowane przez azd są przechowywane lokalnie w środowisku o podanej nazwie. Użyj polecenia , azd env get-values aby przejrzeć wszystkie zmienne w środowisku, które zostały użyte podczas tworzenia zasobów platformy Azure.

Uprzątnij zasoby

Po zakończeniu pracy z aplikacją funkcji i powiązanymi zasobami użyj tego polecenia, aby usunąć aplikację funkcji i powiązane z nią zasoby z platformy Azure i uniknąć ponoszenia dodatkowych kosztów:

azd down --no-prompt

Uwaga / Notatka

Opcja --no-prompt powoduje azd usunięcie grupy zasobów bez potwierdzenia.

To polecenie nie ma wpływu na lokalny projekt kodu.

Powiązane artykuły

• Wyzwalacz usługi Event Hubs dla usługi Azure Functions
• Scenariusze dla Azure Functions
• Elastyczny Plan Konsumpcji
• CLI dla deweloperów platformy Azure (azd)
• azd referencja
• Dokumentacja podstawowych narzędzi usługi Azure Functions
• Kodowanie i testowanie usługi Azure Functions lokalnie