Configurare il buffer dei dati e la persistenza del disco per i flussi di dati

Quando un flusso di dati invia messaggi a un endpoint di destinazione, ad esempio Hub eventi di Azure, Microsoft Fabric Real-Time Intelligence, Kafka o un altro servizio cloud, la destinazione o la rete potrebbe non essere più disponibile. Operazioni di Azure IoT usa il modello di sottoscrizione del broker MQTT locale per memorizzare nel buffer i messaggi e riprovare il recapito.

Questo articolo illustra come i flussi di dati proteggono i messaggi durante le interruzioni di destinazione e come configurare il buffering broker e la persistenza del disco per una protezione più avanzata.

Modalità di memorizzazione dei messaggi nel buffer dei flussi di dati durante le interruzioni di destinazione

Quando si usa il broker MQTT locale come endpoint di origine in un flusso di dati, direttamente o indirettamente tramite asset che lo pubblicano, il flusso di dati riceve i messaggi dal broker MQTT come sottoscrittore. Il flusso di dati riconosce ogni messaggio di origine dopo che il messaggio è stato elaborato correttamente e recapitato alla destinazione oppure dopo l'eliminazione intenzionale del messaggio a causa di filtri, convalida dello schema o scadenza del messaggio.

Se l'endpoint di destinazione non è disponibile, il recapito non può essere completato. In questo caso, il flusso di dati non riconosce il messaggio di origine. Il broker MQTT mantiene il messaggio nella coda del sottoscrittore e il flusso di dati ritenta il recapito. Quando viene ripristinata la connettività, il flusso di dati invia messaggi in coda alla destinazione e li riconosce dopo il recapito riuscito.

Il percorso del messaggio è: editore o asset al broker MQTT, broker MQTT al flusso di dati e flusso di dati verso l'endpoint di destinazione.

1. Il broker MQTT recapita un messaggio al flusso di dati.
2. Il flusso di dati invia il messaggio all'endpoint di destinazione.
3. Se l'invio ha esito positivo, il flusso di dati riconosce il messaggio di origine e il broker lo rimuove dalla coda del sottoscrittore.
4. Se l'invio ha esito negativo, il flusso di dati non riconosce il messaggio di origine. Il broker lo mantiene in coda.
5. Il flusso di dati ritenta il recapito fino a quando non riesce, il messaggio scade o si applica un limite configurato.

Livelli di configurazione della protezione dei dati

Usare i livelli di configurazione seguenti per controllare il modo in cui i messaggi vengono memorizzati nel buffer e protetti durante le interruzioni di destinazione.

La coda degli abbonati del broker MQTT locale viene archiviata in memoria per impostazione predefinita. È possibile configurare il broker MQTT per l'uso del disco in due modi diversi:

• Buffer dei messaggi su disco: utilizza il disco come buffer di riserva quando le code superano la memoria disponibile. Questa impostazione è utile per gestire backlog temporanei più ampi, ma non è equivalente a una persistenza durevole tra i riavvii del broker.
• Persistenza broker MQTT: salva in modo permanente i dati broker selezionati, incluse le code dei sottoscrittori, su disco, in modo che i messaggi in coda possano sopravvivere ai riavvii o alla perdita di energia.

Per informazioni dettagliate sulla configurazione del broker, vedere:

• Pianificazione della distribuzione
• Buffer dei messaggi basato su disco
• Configurare la persistenza del broker MQTT
• Opzioni MQTT avanzate

Scegliere una configurazione di buffering

Scegliere una configurazione in base ai requisiti di durata e durabilità dell'interruzione per il carico di lavoro:

• Per brevi interruzioni temporanee del cloud o della rete, la coda predefinita dei sottoscrittori in memoria potrebbe essere sufficiente.
• Per un maggiore throughput o interruzioni temporanee più prolungate, è necessario configurare il buffer dei messaggi con supporto disco.
• Per la protezione dal riavvio o dalla perdita di energia, abilitare la persistenza del broker MQTT e la persistenza della coda del sottoscrittore, quindi abilitare requestDiskPersistence sul flusso di dati o sul grafico del flusso di dati.
• Per gli ambienti di archiviazione delimitati, configurare i limiti della coda dei sottoscrittori, la scadenza dei messaggi e il monitoraggio in modo che il broker applichi limiti di coda e elimina o rifiuta i messaggi in base ai criteri.

Esempio: Interruzione della destinazione

Si supponga di creare un flusso di dati usando il broker MQTT locale predefinito come endpoint di origine e Hub eventi di Azure come endpoint di destinazione. Se la connettività tra il flusso di dati e Hub eventi di Azure viene persa, il flusso di dati ritenta l'invio e non riconosce i messaggi di origine. Il broker MQTT accoda i messaggi non confermati. Con le impostazioni predefinite, la coda viene archiviata in memoria. Con il buffer dei messaggi supportato dal disco, può essere eseguito lo spill della coda su disco. Con la persistenza del broker e il flusso requestDiskPersistencedi dati , i messaggi in coda possono sopravvivere ai riavvii del broker, soggetti ai limiti di persistenza, scadenza e archiviazione configurati.

Abilitare la persistenza del disco per un flusso di dati

La persistenza del disco consente ai flussi di dati e ai grafici del flusso di dati di mantenere lo stato di elaborazione tra i riavvii. Quando si abilita questa funzionalità, il broker MQTT rende persistenti i dati, ad esempio i messaggi nella coda del sottoscrittore, su disco. Questo approccio consente di assicurarsi che l'origine dati del flusso di dati non perda i dati in coda durante le interruzioni di alimentazione o i riavvii del broker. Il broker mantiene prestazioni ottimali perché la persistenza è configurata per ogni flusso di dati, quindi solo i flussi di dati necessari usano questa funzionalità.

Il flusso di dati richiede la persistenza durante la sottoscrizione usando una proprietà utente MQTTv5. Questa funzionalità può essere usata solo se:

• Il flusso di dati usa il broker o l'asset MQTT come origine.
• Il broker MQTT ha la persistenza abilitata con la modalità di persistenza dinamica impostata su Enabled per il tipo di dati, ad esempio le code dei sottoscrittori.

Per informazioni dettagliate sulla configurazione della persistenza del broker MQTT, vedere Configurare la persistenza del broker MQTT.

L'impostazione accetta Enabled o Disabled. Disabled è l'impostazione predefinita.

Configurare un flusso di dati

{
    "mode": "Enabled",
    "requestDiskPersistence": "Enabled",
    "operations": [
        // ... your data flow operations
    ]
}

resource dataflow 'Microsoft.IoTOperations/instances/dataflowProfiles/dataflows@2026-03-01' = {
  parent: defaultDataflowProfile
  name: dataflowName
  extendedLocation: {
    name: customLocation.id
    type: 'CustomLocation'
  }
  properties: {
    mode: 'Enabled'
    requestDiskPersistence: 'Enabled'
    operations: [
      // ... your data flow operations
    ]
  }
}

apiVersion: connectivity.iotoperations.azure.com/v1
kind: Dataflow
metadata:
  name: <DATAFLOW_NAME>
  namespace: azure-iot-operations
spec:
  profileRef: default 
  mode: Enabled
  requestDiskPersistence: Enabled
  operations:
    # ... your data flow operations

Configurare per un grafico del flusso di dati

resource dataflowGraph 'Microsoft.IoTOperations/instances/dataflowProfiles/dataflowGraphs@2026-03-01' = {
  parent: dataflowProfile
  name: 'my-graph'
  extendedLocation: {
    name: customLocation.id
    type: 'CustomLocation'
  }
  properties: {
    requestDiskPersistence: 'Enabled'
    nodes: [
      // ... your graph nodes
    ]
    nodeConnections: [
      // ... your node connections
    ]
  }
}

apiVersion: connectivity.iotoperations.azure.com/v1
kind: DataflowGraph
metadata:
  name: <GRAPH_NAME>
  namespace: azure-iot-operations
spec:
  profileRef: default
  requestDiskPersistence: Enabled
  nodes:
    # ... your graph nodes
  nodeConnections:
    # ... your node connections

Contenuti correlati

• Creare un flusso di dati
• Panoramica dei grafici del flusso di dati
• Configurare la persistenza del broker MQTT
• Buffer dei messaggi basato su disco