Skip to main content Skills Marketplace Découvrez et explorez les compétences IA créées par la communauté.
Installer avec Codex ou Claude Copiez ce prompt, collez-le dans Codex, Claude ou un autre assistant, puis laissez-le vérifier la page du skill et l'installer pour vous.
Afficher les détails du prompt Une commande directe contourne le prompt de vérification. Examinez la source avant de l'exécuter.
npx skills add https://github.com/BEKO2210/Firstbrain --skill azure-eventhub-tsLa commande reste sur une seule ligne. Faites défiler horizontalement pour la vérifier avant de la copier.
Vous préférez une copie locale ? Téléchargez les fichiers actuellement disponibles dans SkillsMP.
Télécharger Zip Téléchargement... Métiers associés SOC
Basé sur la classification professionnelle SOC
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]] 