(Azure Service Bus服務匯流排的兩大類訊息處理方式： 佇列Queue和主題Topic)

問題描述

使用Service Bus作為企業訊息代理，當有大量的資料堆積再Queue或Topic中時，如何來優化接收端處理訊息的能力呢？

詳細解釋

在接收端(Receover)的程式碼中，有兩個屬性與處理訊息的能力有關。一是maxConcurrentCalls(最大併發處理數)， 二是prefetchCount (預提取訊息數)。 在Service Bus的SDK(azure-messaging-servicebus:7.0.0.0)中，他們的描述如下：

package com.azure.messaging.servicebus.implementation.models;

... ...

public final class ServiceBusProcessorClientOptions {

... ...

    /**
     * The max concurrent messages that should be processed by the processor.
     * @return The max concurrent message that should be processed by the processor.
     */
    public int getMaxConcurrentCalls() {
        return maxConcurrentCalls;
    }

package com.azure.messaging.servicebus;
... ...

public final class ServiceBusClientBuilder { 
... ...

// Using 0 pre-fetch count for both receive modes, to avoid message lock lost exceptions in application 
// receiving messages at a slow rate. Applications can set it to a higher value if they need better performance. 
private static final int DEFAULT_PREFETCH_COUNT = 1;

在初始化ServiceBusProcessorClient物件時，可以設定maxConcurrentCalls和prefetchCount的值。如

// Create an instance of the processor through the ServiceBusClientBuilder
ServiceBusProcessorClient processorClient = new ServiceBusClientBuilder().processError(errorHandler).maxConcurrentCalls(5).prefetchCount(10).buildProcessorClient();

實驗驗證

在本次的實驗中，如何來驗證maxConcurrentCalls值啟作用了呢？如何判斷prefetchCount是否獲取了訊息呢？ 

• 針對maxConcurrentCalls，可以在處理訊息的程式碼中[processMessage(messageProcessor)]打印出當前執行緒的ID[Thread.currentThread().getId()]。
• 針對prefetchCount，可以從側面來驗證，即獲取message的DeliveryCount來判斷已經預提取了多少次

本次實驗的程式碼參考Azure Service Bus的快速入門文件所編寫，文末包含全部的程式碼和POM.XML檔案。

首先在程式碼中設定concall和prefetch值。預設情況下為1.本次實驗也從1開始，在設定的10秒鐘之內檢視消費訊息的數量。

        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();

然後再處理訊息的物件中，打印出當前處理訊息的次序，訊息ID，Delivery次數，處理訊息的執行緒ID。

        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());
        };

第一次實驗：處理訊息的執行緒號只有一個:21, 在10秒時間中處理23條訊息

第二次實驗：處理訊息的執行緒號有5個:21，21，23，24，25, 在10秒時間中處理42條訊息

第三次實驗：處理訊息的執行緒號有10個:21，21 ... 30, 在10秒時間中處理46條訊息

三次測試的結論

1. 在測試中，由於測試的時長只有10秒，所以無法得出一個合理的maxConcurrentCalls和prefetchCount值。至少maxCouncurrentCalls的值能大幅度提升接收端（Receiver）處理訊息的能力。
2. 在第三次的的測試中，我們發現Delivery Count的計數變為了1，這是因為在第二次測試中，我們設定的預提取數量為10，每次提取的數量大於了接收端能處理的數量。在10秒鐘的測試中，並沒有完全處理完所有提取出來的訊息，以致於在第三次測試中，這些訊息的Delivery次數從0變成了1。

優化建議

預提取可加快訊息流程，方法是在應用程式請求訊息時及請求訊息前，準備好訊息用於本地檢索。

• 通過 ReceiveAndDelete 接收模式，預提取快取區獲取的所有訊息在佇列中不再可用，僅保留在記憶體中預提取快取區，直到應用程式通過 Receive/ReceiveAsync 或 OnMessage/OnMessageAsync API 接收到它們 。 如果在應用程式接收到訊息前終止應用程式，這些訊息將丟失，且不可恢復。

• 在 PeekLock 接收模式下，提取到預提取快取區的訊息將以鎖定狀態進入快取區，並且將超時時鐘用於鎖定計時。 如果預提取快取區很大，且處理所需時間過長，以致訊息鎖定在駐留於預提取快取區，甚至應用程式還在處理訊息時就到期，可能出現一些令人困惑的事件要應用程式處理。

如果訊息處理需要高度的可靠性，且處理需要大量精力和時間，則建議謹慎使用或者絲毫不用預提取功能。

如果需要較高吞吐量且訊息處理通常比較便宜，則預提取會產生顯著的吞吐量優勢。

需要均衡對佇列或訂閱配置的最大預提取數和鎖定持續時間，以便鎖定超時至少超出最大預提取快取區大小外加一條訊息的累積預期訊息處理時間。 同時，鎖定超時不應過長，防止訊息在被意外丟棄後超出其最大 TimeToLive，因此需要訊息的鎖定在重新傳送訊息前到期。

附錄一：使用Service Bus Explorer工具快速生成大量訊息

附錄二：測試例項pom.xml內容

<?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>

附錄三：App.java程式碼

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();
    }
}

參考資料

Service Bus Explorer：https://github.com/paolosalvatori/ServiceBusExplorer

預提取 Azure 服務匯流排訊息：https://docs.azure.cn/zh-cn/service-bus-messaging/service-bus-prefetch#if-it-is-faster-why-is-prefetch-not-the-default-option

預提取：https://docs.azure.cn/zh-cn/service-bus-messaging/service-bus-performance-improvements?tabs=net-standard-sdk-2#prefetching

向 Azure 服務匯流排佇列傳送訊息並從中接收訊息 (Java):https://docs.azure.cn/zh-cn/service-bus-messaging/service-bus-java-how-to-use-queues