【Azure 服务总线】详解Azure Service Bus SDK中接收消息时设置的maxConcurrentCalls，prefetchCount参数

int concall=1;
        int prefetch =1;
        // Create an instance of the processor through the ServiceBusClientBuilder
        ServiceBusProcessorClient processorClient = new ServiceBusClientBuilder().connectionString(connectionString)
                .processor().topicName(topicName).subscriptionName(subName).processMessage(messageProcessor)
                .processError(errorHandler).maxConcurrentCalls(concall).prefetchCount(prefetch).buildProcessorClient();
                System.out.println("Starting the processor");
        System.out.println("Set Processor: maxConcurrentCalls = "+concall+", prefetchCount = "+prefetch);
        
        processorClient.start();

Consumer<ServiceBusReceivedMessageContext> messageProcessor = context -> {
            ServiceBusReceivedMessage message = context.getMessage();
            ordernumber++;
            System.out.println(ordernumber + " Message ID:" + message.getMessageId() + ",Current Delivery Count:"
                    + message.getDeliveryCount() + ",Current Thread ID:" + Thread.currentThread().getId());
        };

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
  xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
  <modelVersion>4.0.0</modelVersion>
  <groupId>testgroupid</groupId>
  <artifactId>testartifactid</artifactId>
  <version>1.0-SNAPSHOT</version>
  <name>testartifactid</name>
  <!-- FIXME change it to the project's website -->
  <url>http://www.example.com</url>
  <properties>
    <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
    <maven.compiler.source>1.8</maven.compiler.source>
    <maven.compiler.target>1.8</maven.compiler.target>
  </properties>
  <dependencies>
    <dependency>
      <groupId>junit</groupId>
      <artifactId>junit</artifactId>
      <version>4.11</version>
      <scope>test</scope>
    </dependency>
    <dependency>
      <groupId>com.azure</groupId>
      <artifactId>azure-messaging-servicebus</artifactId>
      <version>7.0.0</version>
  </dependency>
  
    <dependency>
      <groupId>com.azure</groupId>
      <artifactId>azure-messaging-servicebus</artifactId>
      <version>7.0.0-beta.7</version>
    </dependency>
  </dependencies>
  <build>
    <pluginManagement><!-- lock down plugins versions to avoid using Maven defaults (may be moved to parent pom) -->
      <plugins>
        <!-- clean lifecycle, see https://maven.apache.org/ref/current/maven-core/lifecycles.html#clean_Lifecycle -->
        <plugin>
          <artifactId>maven-clean-plugin</artifactId>
          <version>3.1.0</version>
        </plugin>
        <!-- default lifecycle, jar packaging: see https://maven.apache.org/ref/current/maven-core/default-bindings.html#Plugin_bindings_for_jar_packaging -->
        <plugin>
          <artifactId>maven-resources-plugin</artifactId>
          <version>3.0.2</version>
        </plugin>
        <plugin>
          <artifactId>maven-compiler-plugin</artifactId>
          <version>3.8.0</version>
        </plugin>
        <plugin>
          <artifactId>maven-surefire-plugin</artifactId>
          <version>2.22.1</version>
        </plugin>
        <plugin>
          <artifactId>maven-jar-plugin</artifactId>
          <version>3.0.2</version>
        </plugin>
        <plugin>
          <artifactId>maven-install-plugin</artifactId>
          <version>2.5.2</version>
        </plugin>
        <plugin>
          <artifactId>maven-deploy-plugin</artifactId>
          <version>2.8.2</version>
        </plugin>
        <!-- site lifecycle, see https://maven.apache.org/ref/current/maven-core/lifecycles.html#site_Lifecycle -->
        <plugin>
          <artifactId>maven-site-plugin</artifactId>
          <version>3.7.1</version>
        </plugin>
        <plugin>
          <artifactId>maven-project-info-reports-plugin</artifactId>
          <version>3.0.0</version>
        </plugin>
      </plugins>
    </pluginManagement>
  </build>
</project>

package com.servicebus.test;
import com.azure.messaging.servicebus.*;
import com.azure.messaging.servicebus.models.*;
import java.util.concurrent.TimeUnit;
import java.util.function.Consumer;
import java.sql.Date;
import java.util.Arrays;
import java.util.List;
/**
 * Hello world!
 *
 */
public class App {
    static String connectionString = "Endpoint=sb://xxxxxxxx.servicebus.chinacloudapi.cn/;SharedAccessKeyName=RootManageSharedAccessKey;SharedAccessKey=xxxxxxxxxxxxxxxxxxxxxxxxxxxxx";
    static String topicName = "thisistest";
    static String subName = "lubusb1";
    static int ordernumber = 0;
    public static void main(String[] args) throws InterruptedException {
        System.out.println("Hello World!");
        // sendMessage();
        // sendMessageBatch();
        receiveMessages();
        System.out.println("Done World!");
    }
    // handles received messages
    static void receiveMessages() throws InterruptedException {
        // Consumer that processes a single message received from Service Bus
        Consumer<ServiceBusReceivedMessageContext> messageProcessor = context -> {
            ServiceBusReceivedMessage message = context.getMessage();
            ordernumber++;
            System.out.println(ordernumber + " Message ID:" + message.getMessageId() + ",Current Delivery Count:"
                    + message.getDeliveryCount() + ",Current Thread ID:" + Thread.currentThread().getId());
        };
        // Consumer that handles any errors that occur when receiving messages
        Consumer<Throwable> errorHandler = throwable -> {
            System.out.println("Error when receiving messages: " + throwable.getMessage());
            if (throwable instanceof ServiceBusReceiverException) {
                ServiceBusReceiverException serviceBusReceiverException = (ServiceBusReceiverException) throwable;
                System.out.println("Error source: " + serviceBusReceiverException.getErrorSource());
            }
        };
        int concall=10;
        int prefetch =30;
        // Create an instance of the processor through the ServiceBusClientBuilder
        ServiceBusProcessorClient processorClient = new ServiceBusClientBuilder().connectionString(connectionString)
                .processor().topicName(topicName).subscriptionName(subName).processMessage(messageProcessor)
                .processError(errorHandler).maxConcurrentCalls(concall).prefetchCount(prefetch).buildProcessorClient();
                System.out.println("Starting the processor");
        System.out.println("Set Processor: maxConcurrentCalls = "+concall+", prefetchCount = "+prefetch);
        
        processorClient.start();
               
        TimeUnit.SECONDS.sleep(10);
        System.out.println("Total Process Message Count = "+ordernumber+" in 10 seconds.");
        System.out.println("Stopping and closing the processor");
        processorClient.close();
    }
    static void sendMessage() {
        // create a Service Bus Sender client for the queue
        ServiceBusSenderClient senderClient = new ServiceBusClientBuilder().connectionString(connectionString).sender()
                .topicName(topicName).buildClient();
        // send one message to the topic
        senderClient.sendMessage(new ServiceBusMessage("Hello, World!"));
        System.out.println("Sent a single message to the topic: " + topicName);
    }
    static List<ServiceBusMessage> createMessages() {
        // create a list of messages and return it to the caller
        ServiceBusMessage[] messages = { new ServiceBusMessage("First message"),
                new ServiceBusMessage("Second message"), new ServiceBusMessage("Third message") };
        return Arrays.asList(messages);
    }
    static void sendMessageBatch() {
        // create a Service Bus Sender client for the topic
        ServiceBusSenderClient senderClient = new ServiceBusClientBuilder().connectionString(connectionString).sender()
                .topicName(topicName).buildClient();
        // Creates an ServiceBusMessageBatch where the ServiceBus.
        ServiceBusMessageBatch messageBatch = senderClient.createMessageBatch();
        // create a list of messages
        List<ServiceBusMessage> listOfMessages = createMessages();
        // We try to add as many messages as a batch can fit based on the maximum size
        // and send to Service Bus when
        // the batch can hold no more messages. Create a new batch for next set of
        // messages and repeat until all
        // messages are sent.
        for (ServiceBusMessage message : listOfMessages) {
            if (messageBatch.tryAddMessage(message)) {
                continue;
            }
            // The batch is full, so we create a new batch and send the batch.
            senderClient.sendMessages(messageBatch);
            System.out.println("Sent a batch of messages to the topic: " + topicName);
            // create a new batch
            messageBatch = senderClient.createMessageBatch();
            // Add that message that we couldn't before.
            if (!messageBatch.tryAddMessage(message)) {
                System.err.printf("Message is too large for an empty batch. Skipping. Max size: %s.",
                        messageBatch.getMaxSizeInBytes());
            }
        }
        if (messageBatch.getCount() > 0) {
            senderClient.sendMessages(messageBatch);
            System.out.println("Sent a batch of messages to the topic: " + topicName);
        }
        // close the client
        senderClient.close();
    }
}