Build real-time streaming applications with Azure Event Hubs SDK for Java. Use when implementing event streaming, high-throughput data ingestion, or building event-driven architectures.
يبقى الأمر في سطر واحد. مرّر أفقيًا لمراجعته كاملًا قبل النسخ.
تفضّل نسخة محلية؟ نزّل الملفات المتاحة حاليًا لدى SkillsMP.
عرض SKILL.md
SKILL.md
تعليمات المصدر · معاينة للقراءة فقط
name
azure-eventhub-java
description
Build real-time streaming applications with Azure Event Hubs SDK for Java. Use when implementing event streaming, high-throughput data ingestion, or building event-driven architectures.
Build real-time streaming applications using the Azure Event Hubs SDK for Java.
Installation
<dependency><groupId>com.azure</groupId><artifactId>azure-messaging-eventhubs</artifactId><version>5.19.0</version></dependency><!-- For checkpoint store (production) --><dependency><groupId>com.azure</groupId><artifactId>azure-messaging-eventhubs-checkpointstore-blob</artifactId><version>1.20.0</version></dependency>
Client Creation
EventHubProducerClient
import com.azure.messaging.eventhubs.EventHubProducerClient;
import com.azure.messaging.eventhubs.EventHubClientBuilder;
// With connection stringEventHubProducerClientproducer=newEventHubClientBuilder()
.connectionString("<connection-string>", "<event-hub-name>")
.buildProducerClient();
// Full connection string with EntityPathEventHubProducerClientproducer= ()
.connectionString()
.buildProducerClient();
// Start from beginning
EventPosition.earliest()
// Start from end (new events only)
EventPosition.latest()
// From specific offset
EventPosition.fromOffset(12345L)
// From specific sequence number
EventPosition.fromSequenceNumber(100L)
// From specific time
EventPosition.fromEnqueuedTime(Instant.now().minus(Duration.ofHours(1)))
Error Handling
import com.azure.messaging.eventhubs.models.ErrorContext;
.processError(errorContext -> {
Throwableerror= errorContext.getThrowable();
StringpartitionId= errorContext.getPartitionContext().getPartitionId();
if (error instanceof AmqpException) {
AmqpExceptionamqpError= (AmqpException) error;
if (amqpError.isTransient()) {
System.out.println("Transient error, will retry");
}
}
System.err.printf("Error on partition %s: %s%n", partitionId, error.getMessage());
})
Resource Cleanup
// Always close clientstry {
producer.send(batch);
} finally {
producer.close();
}
// Or use try-with-resourcestry (EventHubProducerClientproducer=newEventHubClientBuilder()
.connectionString(connectionString, eventHubName)
.buildProducerClient()) {
producer.send(events);
}