Azure Service Bus Java客户端：实现高效并发消息处理

import com.azure.identity.DefaultAzureCredentialBuilder;
import com.azure.messaging.servicebus.ServiceBusClientBuilder;
import com.azure.messaging.servicebus.ServiceBusReceivedMessage;
import com.azure.messaging.servicebus.ServiceBusReceiverAsyncClient;
import com.azure.core.credential.DefaultAzureCredential;
import reactor.core.Disposable;

public class AsyncMessageProcessor {

    private static final String CONNECTION_STRING = "YOUR_SERVICE_BUS_CONNECTION_STRING";
    private static final String QUEUE_NAME = "YOUR_QUEUE_NAME";

    public static void main(String[] args) {
        DefaultAzureCredential credential = new DefaultAzureCredentialBuilder().build();

        ServiceBusReceiverAsyncClient asyncClient = new ServiceBusClientBuilder()
            .credential(credential)
            .connectionString(CONNECTION_STRING)
            .receiver()
            .queueName(QUEUE_NAME)
            .buildAsyncClient();

        System.out.println("Starting to receive messages...");

        Disposable subscription = asyncClient.receiveMessages()
            .subscribe(
                AsyncMessageProcessor::processMessage,
                AsyncMessageProcessor::processError,
                () -> System.out.println("Receiving complete.")
            );

        // Keep the main thread alive to continue receiving messages
        // In a real application, you might manage this lifecycle differently
        try {
            Thread.sleep(Long.MAX_VALUE);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            System.err.println("Main thread interrupted.");
        } finally {
            subscription.dispose(); // Clean up the subscription
            asyncClient.close(); // Close the client
            System.out.println("Client closed.");
        }
    }

    private static void processMessage(ServiceBusReceivedMessage message) {
        System.out.println("Processing message. Thread: " + Thread.currentThread().getName());
        System.out.printf("Processed message. Session: %s, Sequence #: %s. Contents: %s%n",
            message.getMessageId(), message.getSequenceNumber(), message.getBody());

        // Simulate some work that takes time
        try {
            Thread.sleep(100); // For demonstration, actual processing might be longer
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
        // In a real scenario, you would complete/abandon/dead-letter the message
        // asyncClient.complete(message).subscribe(); // This would be done asynchronously
    }

    private static void processError(Throwable error) {
        System.err.println("Error occurred: " + error.getMessage());
    }
}

import com.azure.identity.DefaultAzureCredentialBuilder;
import com.azure.messaging.servicebus.ServiceBusClientBuilder;
import com.azure.messaging.servicebus.ServiceBusReceivedMessage;
import com.azure.messaging.servicebus.ServiceBusProcessorClient;
import com.azure.core.credential.DefaultAzureCredential;

import java.util.concurrent.TimeUnit;

public class ConcurrentMessageProcessor {

    private static final String CONNECTION_STRING = "YOUR_SERVICE_BUS_CONNECTION_STRING";
    private static final String QUEUE_NAME = "YOUR_QUEUE_NAME";

    public static void main(String[] args) throws InterruptedException {
        DefaultAzureCredential credential = new DefaultAzureCredentialBuilder().build();

        // Create an instance of the processor through the ServiceBusClientBuilder
        ServiceBusProcessorClient processorClient = new ServiceBusClientBuilder()
            .credential(credential)
            .connectionString(CONNECTION_STRING)
            .processor()
            .queueName(QUEUE_NAME)
            // 配置并发处理的最大消息数量
            .maxConcurrentCalls(5) // 示例：同时处理5条消息
            .processMessage(ConcurrentMessageProcessor::processMessage)
            .processError(ConcurrentMessageProcessor::processError)
            .buildProcessorClient();

        System.out.println("Starting the processor...");
        processorClient.start(); // 启动处理器，开始接收和处理消息

        // Keep the main thread alive for a duration to allow processing
        // In a real application, manage the lifecycle appropriately
        TimeUnit.SECONDS.sleep(60); // Allow processing for 60 seconds

        System.out.println("Stopping the processor...");
        processorClient.stop(); // 停止处理器
        System.out.println("Processor stopped.");
    }

    private static void processMessage(ServiceBusReceivedMessage message) {
        System.out.println("Processing message. Thread: " + Thread.currentThread().getName());
        System.out.printf("Processed message. Session: %s, Sequence #: %s. Contents: %s%n",
            message.getMessageId(), message.getSequenceNumber(), message.getBody());

        // Simulate some work that takes time
        try {
            Thread.sleep(500); // Simulate longer processing time
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }

        // 自动完成消息（如果未手动配置）或在此处手动完成
        // processorClient会自动处理消息的完成、放弃、死信等操作，
        // 除非你在processMessage方法中抛出异常或显式调用context.abandon/deadLetter
    }

    private static void processError(Throwable error) {
        System.err.println("Error occurred: " + error.getMessage());
    }
}