zoukankan html css js c++ java

【Azure 事件中心】Event Hub Client 连接超时(OperationTimeout)测试及解说

Azure Event Hub(Azure事件中心) 是大数据流式处理平台和事件引入服务。它可以每秒接收和处理数百万个事件。在我们的使用中，需要代码编写的是两个部分：事件生产者和事件接收者

事件生成者：向事件中心发送数据的所有实体。事件发布者可以使用 HTTPS、AMQP 1.0 或 Apache Kafka（1.0 和更高版本）发布事件。
事件接收者：从事件中心读取事件数据的所有实体。所有事件中心使用者通过 AMQP 1.0 会话进行连接。事件中心服务在事件变得可用时通过会话来提供事件。所有 Kafka 使用者都通过 Kafka 协议 1.0 及更高版本进行连接。

下图显示了事件中心流处理体系结构：

(其他组件说明：https://docs.azure.cn/zh-cn/event-hubs/event-hubs-about#key-architecture-components)

问题描述

在使用事件生产者或接收者时，都需要与Event Hub服务建立连接。那么只要建立了连接，都需要考虑连接空闲超时参数。当发生Timout后，EventHubClient的表现是什么呢？有什么日志输出呢？同时，当新的消息需要发送时，EventHubClient时如何重建的连接呢? 重建连接耗时多长呢？

问题分析

以上问题，都可以通过调试EventHubClient的代码日志来分析。

实验步骤：

准备条件：参考文档，构建EventHub的发送消息代码：

【旧版】使用 Java 向/从 Azure 事件中心 (azure-eventhubs) 发送/接收事件：https://docs.azure.cn/zh-cn/event-hubs/event-hubs-java-get-started-send-legacy

【新版】使用 Java 向/从 Azure 事件中心 (azure-messaging-eventhubs) 发送/接收事件：https://docs.azure.cn/zh-cn/event-hubs/event-hubs-java-get-started-send

1) 在初始化EventHubClient对象时，使用默认的OperationTimout时间（60秒）

2) 在发送第一条消息前，使线程睡眠65秒。验证EventHubClient对象超时后SDK的输出

3) 记录触发发送第一条消息前的时间，获取EventHubClient建立连接的日志。计算耗时时间

4) 根据测试中输出的日志，得出5点结论

1：在2021-03-23 15:17:37.505时候，Connection连接建立成功，由于线程睡眠，所以在2021-03-23 15:18:37.463 （60秒后）Connection关闭，理由是( errorDescription[The connection was inactive for more than the allowed 60000 milliseconds and is closed by container 'LinkTracker')

2021-03-23 15:17:37.505  INFO 104016 --- [pool-1-thread-1] c.m.a.eventhubs.impl.ConnectionHandler   : onConnectionRemoteOpen: hostname[lbtesteh.servicebus.chinacloudapi.cn:5671], remoteContainer[cc19381675a649259fafb0c0d7faebe0_G18]
2021-03-23T07:17:37.538384400Z: Start to sleep ... waiting the connection idle timeout
2021-03-23 15:18:37.463  INFO 104016 --- [pool-1-thread-1] c.m.a.eventhubs.impl.ConnectionHandler   : onConnectionRemoteClose: hostname[lbtesteh.servicebus.chinacloudapi.cn:5671], errorCondition[amqp:connection:forced], errorDescription[The connection was inactive for more than the allowed 60000 milliseconds and is closed by container 'LinkTracker'. TrackingId:cc19381675a649259fafb0c0d7faebe0_G18, SystemTracker:gateway7, Timestamp:2021-03-23T07:18:37]

2：当线程睡眠结束，在2021-03-23T15:18:42.559930300开始发送消息

3：EventHubClient马上开始重新创建 Connection， Session，Link。耗时近2秒（从2021-03-23 15:18:42.578到2021-03-23 15:18:44.592）

4：重建连接后第一条消息发送成功在2021-03-23T15:18:44.829493900

2021-03-23T07:18:42.538152400Z: End to sleep ...
Start: Send message 0 !2021-03-23T15:18:42.559930300
2021-03-23 15:18:42.578  INFO 104016 --- [pool-1-thread-1] c.m.azure.eventhubs.impl.MessageSender   : clientId[EventHubClientImplb2e9d6-InternalSender], path[receivemessage], operationTimeout[PT1M30S], creating a send link
2021-03-23 15:18:42.606  INFO 104016 --- [pool-1-thread-1] c.m.a.eventhubs.impl.ConnectionHandler   : onConnectionInit hostname[lbtesteh.servicebus.chinacloudapi.cn]
2021-03-23 15:18:42.607  INFO 104016 --- [pool-1-thread-1] c.m.azure.eventhubs.impl.SessionHandler  : onSessionLocalOpen entityName[cbs-session], condition[Error{condition=null, description='null', info=null}]
2021-03-23 15:18:42.608  INFO 104016 --- [pool-1-thread-1] c.m.a.eventhubs.impl.SendLinkHandler     : onLinkLocalOpen linkName[cbs:sender], localTarget[Target{address='$cbs', durable=NONE, expiryPolicy=SESSION_END, timeout=0, dynamic=false, dynamicNodeProperties=null, capabilities=null}]
2021-03-23 15:18:42.609  INFO 104016 --- [pool-1-thread-1] c.m.a.eventhubs.impl.ReceiveLinkHandler  : onLinkLocalOpen linkName[cbs:receiver], localSource[Source{address='$cbs', durable=NONE, expiryPolicy=SESSION_END, timeout=0, dynamic=false, dynamicNodeProperties=null, distributionMode=null, filter=null, defaultOutcome=null, outcomes=null, capabilities=null}]
2021-03-23 15:18:42.609  INFO 104016 --- [pool-1-thread-1] c.m.a.eventhubs.impl.ConnectionHandler   : onConnectionLocalOpen: hostname[lbtesteh.servicebus.chinacloudapi.cn:5671], errorCondition[null], errorDescription[null]     
2021-03-23 15:18:42.610  INFO 104016 --- [pool-1-thread-1] c.m.a.eventhubs.impl.ConnectionHandler   : onConnectionBound hostname[lbtesteh.servicebus.chinacloudapi.cn]
2021-03-23 15:18:43.830  INFO 104016 --- [pool-1-thread-1] c.m.a.eventhubs.impl.ConnectionHandler   : onConnectionRemoteOpen: hostname[lbtesteh.servicebus.chinacloudapi.cn:5671], remoteContainer[79f3573f29a44a1296940126b03be077_G20]
2021-03-23 15:18:43.832  INFO 104016 --- [pool-1-thread-1] c.m.azure.eventhubs.impl.SessionHandler  : onSessionRemoteOpen entityName[cbs-session], sessionIncCapacity[0], sessionOutgoingWindow[2147483647]
2021-03-23 15:18:43.833  INFO 104016 --- [pool-1-thread-1] c.m.a.eventhubs.impl.SendLinkHandler     : onLinkRemoteOpen linkName[cbs:sender], remoteTarget[Target{address='$cbs', durable=NONE, expiryPolicy=SESSION_END, timeout=0, dynamic=false, dynamicNodeProperties=null, capabilities=null}]
2021-03-23 15:18:43.835  INFO 104016 --- [pool-1-thread-1] c.m.a.eventhubs.impl.ReceiveLinkHandler  : onLinkRemoteOpen linkName[cbs:receiver], remoteSource[Source{address='$cbs', durable=NONE, expiryPolicy=SESSION_END, timeout=0, dynamic=false, dynamicNodeProperties=null, distributionMode=null, filter=null, defaultOutcome=null, outcomes=null, capabilities=null}]
2021-03-23 15:18:44.265  INFO 104016 --- [pool-1-thread-1] c.m.azure.eventhubs.impl.SessionHandler  : onSessionLocalOpen entityName[receivemessage], condition[Error{condition=null, description='null', info=null}]
2021-03-23 15:18:44.420  INFO 104016 --- [pool-1-thread-1] c.m.azure.eventhubs.impl.SessionHandler  : onSessionRemoteOpen entityName[receivemessage], sessionIncCapacity[0], sessionOutgoingWindow[2147483647]
2021-03-23 15:18:44.421  INFO 104016 --- [pool-1-thread-1] c.m.a.eventhubs.impl.SendLinkHandler     : onLinkLocalOpen linkName[37adea_077_G20_1616483924421], localTarget[Target{address='receivemessage', durable=NONE, expiryPolicy=SESSION_END, timeout=0, dynamic=false, dynamicNodeProperties=null, capabilities=null}]
2021-03-23 15:18:44.590  INFO 104016 --- [pool-1-thread-1] c.m.a.eventhubs.impl.SendLinkHandler     : onLinkRemoteOpen linkName[37adea_077_G20_1616483924421], remoteTarget[Target{address='receivemessage', durable=NONE, expiryPolicy=SESSION_END, timeout=0, dynamic=false, dynamicNodeProperties=null, capabilities=null}]
2021-03-23 15:18:44.592  INFO 104016 --- [pool-1-thread-1] c.m.azure.eventhubs.impl.MessageSender   : onOpenComplete - clientId[EventHubClientImplb2e9d6-InternalSender], sendPath[receivemessage], linkName[37adea_077_G20_1616483924421]
Completed: Send message 0 !2021-03-23T15:18:44.829493900

5：第二条，第N条发送的时间就非常的短暂，在300毫秒左右。

Completed: Send message 0 !2021-03-23T15:18:44.829493900
Start: Send message 1 !2021-03-23T15:18:44.830505
Completed: Send message 1 !2021-03-23T15:18:45.035403600
Start: Send message 2 !2021-03-23T15:18:45.035403600
Completed: Send message 2 !2021-03-23T15:18:45.249423400
Start: Send message 3 !2021-03-23T15:18:45.250395800
Completed: Send message 3 !2021-03-23T15:18:45.468455300
Start: Send message 4 !2021-03-23T15:18:45.470397200
Completed: Send message 4 !2021-03-23T15:18:45.769922900
Start: Send message 5 !2021-03-23T15:18:45.770920300
Completed: Send message 5 !2021-03-23T15:18:45.973769300
Start: Send message 6 !2021-03-23T15:18:45.975713700
Completed: Send message 6 !2021-03-23T15:18:46.168880300
Start: Send message 7 !2021-03-23T15:18:46.169844900
Completed: Send message 7 !2021-03-23T15:18:46.377778800

总结：EventHub Client在创建过程中的三个关键因素和日志中所打印出的事件顺序

EventHub Client使用AMQP协议的连接创建与关闭的过程: 它涉及到三个主要的部分：connection，session和link

三者关系图

创建连接全过程事件顺序

关闭连接全过程事件顺序

代码

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 https://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>
    <parent>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-parent</artifactId>
        <version>2.4.4</version>
        <relativePath/> <!-- lookup parent from repository -->
    </parent>
    <groupId>com.example</groupId>
    <artifactId>sendehmsg</artifactId>
    <version>0.0.1-SNAPSHOT</version>
    <name>sendehmsg</name>
    <description>Demo project for Spring Boot</description>
    <properties>
        <java.version>11</java.version>
    </properties>
    <dependencies>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
        </dependency>

        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-test</artifactId>
            <scope>test</scope>
        </dependency>

        <dependency>
            <groupId>com.microsoft.azure</groupId>
            <artifactId>azure-eventhubs</artifactId>
            <version>2.2.0</version>
        </dependency>
        
    </dependencies>

    <build>
        <plugins>
            <plugin>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-maven-plugin</artifactId>
            </plugin>
        </plugins>
    </build>

</project>

View Code

SendehmsgApplication.java

package com.example.sendehmsg;

import com.microsoft.azure.eventhubs.ConnectionStringBuilder;
import com.microsoft.azure.eventhubs.EventData;
import com.microsoft.azure.eventhubs.EventHubClient;
import com.microsoft.azure.eventhubs.EventHubException;

import java.io.IOException;
import java.net.URI;
import java.net.URISyntaxException;
import java.nio.charset.Charset;
import java.sql.Date;
import java.sql.Time;
import java.time.Duration;
import java.time.Instant;
import java.time.LocalDateTime;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;

import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;

@SpringBootApplication
public class SendehmsgApplication {

    public static void main(String[] args)
            throws EventHubException, IOException, InterruptedException, URISyntaxException {
        SpringApplication.run(SendehmsgApplication.class, args);


        final ConnectionStringBuilder connStr = new ConnectionStringBuilder().setNamespaceName("xxxxxxxxxxxxxx")
                .setEndpoint(new URI("sb://xxxxxxxxxxxxxx.servicebus.chinacloudapi.cn/")).setEventHubName("receivemessage")
                .setSasKeyName("all").setSasKey("xxxxxxxxxxxxxx")
                .setOperationTimeout(Duration.ofSeconds(90));

        // The Executor handles all asynchronous tasks and this is passed to the
        // EventHubClient instance.
        // This enables the user to segregate their thread pool based on the work load.
        // This pool can then be shared across multiple EventHubClient instances.
        // The following sample uses a single thread executor, as there is only one
        // EventHubClient instance,
        // handling different flavors of ingestion to Event Hubs here.
        final ScheduledExecutorService executorService = Executors.newScheduledThreadPool(1);

        // Each EventHubClient instance spins up a new TCP/TLS connection, which is
        // expensive.
        // It is always a best practice to reuse these instances. The following sample
        // shows this.
        // final EventHubClient ehClient = EventHubClient.createSync(connectionstr,
        // executorService);
        final EventHubClient ehClient = EventHubClient.createSync(connStr.toString(), executorService);

        // Send the First message
        EventData sendEvent = EventData.create("Message 0".getBytes());
        System.out.println("Start: Send message 0 !" + LocalDateTime.now().toString());
        ehClient.sendSync(sendEvent);
        System.out.println("Completed: Send message 0 !" + LocalDateTime.now().toString());

        try {
            for (int i = 1; i < 100; i++) {
                var timelong = 10000;
                System.out.println("@@@@@@@@@@@@@@@@" + LocalDateTime.now().toString() + ": [" + i
                        + "] Start to sleep ... waiting the connection idle timeout");
                for (int j = 0; j < i*30; j++) {
                    Thread.sleep(timelong);
                }
                System.out.println("@@@@@@@@@@@@@@@@" + LocalDateTime.now().toString() + ": [" + i
                        + "] End to sleep ..." + timelong);
                String payload = "Message " + Integer.toString(i);
                // byte[] payloadBytes =
                // gson.toJson(payload).getBytes(Charset.defaultCharset());
                sendEvent = EventData.create(payload.getBytes());

                // Send - not tied to any partition
                // Event Hubs service will round-robin the events across all Event Hubs
                // partitions.
                // This is the recommended & most reliable way to send to Event Hubs.

                System.out.println("Start: Send message " + i + " !" + LocalDateTime.now().toString());

                ehClient.sendSync(sendEvent);

                System.out.println("Completed: Send message " + i + " !" + LocalDateTime.now().toString());
            }

            System.out.println(Instant.now() + ": Send Complete...");
            System.out.println("Press Enter to stop.");

            // System.in.read();
        } finally {
            ehClient.closeSync();
            executorService.shutdown();
        }

    }

}