Skip to main content Skills Marktplatz Entdecken und erkunden Sie KI-Skills, die von der Community erstellt wurden.
Mit Codex oder Claude installieren Kopieren Sie diesen Prompt, fügen Sie ihn in Codex, Claude oder einen anderen Assistant ein und lassen Sie die Skill-Seite prüfen und installieren.
Prompt-Details anzeigen Ein direkter Befehl überspringt den Prüf-Prompt. Prüfen Sie die Quelle, bevor Sie ihn ausführen.
npx skills add https://github.com/BEKO2210/Firstbrain --skill azure-eventhub-tsDer Befehl bleibt in einer Zeile. Scrollen Sie horizontal, um ihn vor dem Kopieren vollständig zu prüfen.
Sie bevorzugen eine lokale Kopie? Laden Sie die Dateien herunter, die SkillsMP derzeit vorliegen.
ZIP herunterladen Herunterladen... name azure-eventhub-ts description High-throughput event streaming and real-time data ingestion. type skill created 2026-02-27T00:00:00.000Z domain cloud-infrastructure category azure risk unknown source community tags ["skill","cloud-infrastructure","azure","eventhub"]
Azure Event Hubs SDK for TypeScript
High-throughput event streaming and real-time data ingestion.
Installation
npm install @azure/event-hubs @azure/identity
For checkpointing with consumer groups:
npm install @azure/eventhubs-checkpointstore-blob @azure/storage-blob
Environment Variables
EVENTHUB_NAMESPACE=<namespace>.servicebus.windows.net
EVENTHUB_NAME=my-eventhub
STORAGE_ACCOUNT_NAME=<storage-account>
STORAGE_CONTAINER_NAME=checkpoints
Authentication
import { EventHubProducerClient , EventHubConsumerClient } from "@azure/event-hubs" ;
import { DefaultAzureCredential } from "@azure/identity" ;
const fullyQualifiedNamespace = process.env .EVENTHUB_NAMESPACE !;
const eventHubName = process.env .EVENTHUB_NAME !;
const credential = new DefaultAzureCredential ();
const producer = new EventHubProducerClient (fullyQualifiedNamespace, eventHubName, credential);
const consumer = new EventHubConsumerClient (
"$Default" ,
fullyQualifiedNamespace,
eventHubName,
credential
);
Core Workflow
Send Events
const producer = new EventHubProducerClient (namespace, eventHubName, credential);
const
await
createBatch
tryAdd
body
temperature
72.5
deviceId
"sensor-1"
tryAdd
body
temperature
68.2
deviceId
"sensor-2"
await
sendBatch
await
close
Send to Specific Partition
const batch = await producer.createBatch ({ partitionId : "0" });
const batch = await producer.createBatch ({ partitionKey : "device-123" });
Receive Events (Simple) const consumer = new EventHubConsumerClient ("$Default" , namespace, eventHubName, credential);
const subscription = consumer.subscribe ({
processEvents : async (events, context) => {
for (const event of events) {
console .log (`Partition: ${context.partitionId} , Body: ${JSON .stringify(event.body)} ` );
}
},
processError : async (err, context) => {
console .error (`Error on partition ${context.partitionId} : ${err.message} ` );
},
});
setTimeout (async () => {
await subscription.close ();
await consumer.close ();
}, 60000 );
Receive with Checkpointing (Production) import { EventHubConsumerClient } from "@azure/event-hubs" ;
import { ContainerClient } from "@azure/storage-blob" ;
import { BlobCheckpointStore } from "@azure/eventhubs-checkpointstore-blob" ;
const containerClient = new ContainerClient (
`https://${storageAccount} .blob.core.windows.net/${containerName} ` ,
credential
);
const checkpointStore = new BlobCheckpointStore (containerClient);
const consumer = new EventHubConsumerClient (
"$Default" ,
namespace,
eventHubName,
credential,
checkpointStore
);
const subscription = consumer.subscribe ({
processEvents : async (events, context) => {
for (const event of events) {
console .log (`Processing: ${JSON .stringify(event.body)} ` );
}
if (events.length > 0 ) {
await context.updateCheckpoint (events[events.length - 1 ]);
}
},
processError : async (err, context) => {
console .error (`Error: ${err.message} ` );
},
});
Receive from Specific Position const subscription = consumer.subscribe ({
processEvents : async (events, context) => { },
processError : async (err, context) => { },
}, {
startPosition : {
"0" : { offset : "@earliest" },
"1" : { offset : "@latest" },
"2" : { offset : "12345" },
"3" : { enqueuedOn : new Date ("2024-01-01" ) },
},
});
Event Hub Properties
const hubProperties = await producer.getEventHubProperties ();
console .log (`Partitions: ${hubProperties.partitionIds} ` );
const partitionProperties = await producer.getPartitionProperties ("0" );
console .log (`Last sequence: ${partitionProperties.lastEnqueuedSequenceNumber} ` );
Batch Processing Options const subscription = consumer.subscribe (
{
processEvents : async (events, context) => { },
processError : async (err, context) => { },
},
{
maxBatchSize : 100 ,
maxWaitTimeInSeconds : 30 ,
}
);
Key Types import {
EventHubProducerClient ,
EventHubConsumerClient ,
EventData ,
ReceivedEventData ,
PartitionContext ,
Subscription ,
SubscriptionEventHandlers ,
CreateBatchOptions ,
EventPosition ,
} from "@azure/event-hubs" ;
import { BlobCheckpointStore } from "@azure/eventhubs-checkpointstore-blob" ;
Event Properties
const batch = await producer.createBatch ();
batch.tryAdd ({
body : { data : "payload" },
properties : {
eventType : "telemetry" ,
deviceId : "sensor-1" ,
},
contentType : "application/json" ,
correlationId : "request-123" ,
});
consumer.subscribe ({
processEvents : async (events, context) => {
for (const event of events) {
console .log (`Type: ${event.properties?.eventType} ` );
console .log (`Sequence: ${event.sequenceNumber} ` );
console .log (`Enqueued: ${event.enqueuedTimeUtc} ` );
console .log (`Offset: ${event.offset} ` );
}
},
});
Error Handling consumer.subscribe ({
processEvents : async (events, context) => {
try {
for (const event of events) {
await processEvent (event);
}
await context.updateCheckpoint (events[events.length - 1 ]);
} catch (error) {
console .error ("Processing failed:" , error);
}
},
processError : async (err, context) => {
if (err.name === "MessagingError" ) {
console .warn ("Transient error:" , err.message );
} else {
console .error ("Fatal error:" , err);
}
},
});
Best Practices
Use checkpointing - Always checkpoint in production for exactly-once processingBatch sends - Use createBatch() for efficient sendingPartition keys - Use partition keys to ensure ordering for related eventsConsumer groups - Use separate consumer groups for different processing pipelinesHandle errors gracefully - Don't checkpoint on processing failuresClose clients - Always close producer/consumer when doneMonitor lag - Track lastEnqueuedSequenceNumber vs processed sequence
When to Use This skill is applicable to execute the workflow or actions described in the overview.
Connections
Domain: [[Cloud & Infrastruktur]]Kategorie: [[Microsoft Azure]]Navigation: [[Skills Uebersicht]], [[Home]] Mehr aus diesem Repository
Verwandte Berufe SOC
Basierend auf der SOC-Berufsklassifikation
