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>