(Azure Service Bus服務總線的兩大類消息處理方式： 隊列Queue和主題Topic)
問題描述
使用Service Bus作為企業消息代理，當有大量的數據堆積再Queue或Topic中時，如何來優化接收端處理消息的能力呢？
詳細解釋
在接收端(Receover)的代碼中，有兩個屬性與處理消息的能力有關。一是maxConcurrentCalls(最大並發處理數)， 二是prefetchCount (預提取消息數)。 在Service Bus的SDK(azure-messaging-servicebus:7.0.0.0)中，他們的描述如下：
| maxConcurrentCalls |
接收端所定義的ServiceBusProcessorClient處理的最大並發消息數。
The max concurrent messages that should be processed by the processor. |
package com.azure.messaging.servicebus.implementation.models; ... ...
public final class ServiceBusProcessorClientOptions { |
| prefetchCount | 接收端要預先提取的消息數 The number of messages to prefetch |
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條消息
| Hello World! |
第二次實驗：處理消息的線程號有5個:21，21，23，24，25, 在10秒時間中處理42條消息
| Hello World! |
第三次實驗：處理消息的線程號有10個:21，21 ... 30, 在10秒時間中處理46條消息
| Hello World! |
三次測試的結論
- 在測試中，由於測試的時長只有10秒，所以無法得出一個合理的maxConcurrentCalls和prefetchCount值。至少maxCouncurrentCalls的值能大幅度提升接收端（Receiver）處理消息的能力。
- 在第三次的的測試中，我們發現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
向 Azure 服務總線隊列發送消息並從中接收消息 (Java):https://docs.azure.cn/zh-cn/service-bus-messaging/service-bus-java-how-to-use-queues