-
Notifications
You must be signed in to change notification settings - Fork 11.7k
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
[ISSUE #4099]Optimized the performance of sending traceMessage in AsyncTraceDispatcher
#4180
Conversation
Codecov Report
@@ Coverage Diff @@
## develop #4180 +/- ##
=============================================
+ Coverage 47.92% 48.00% +0.07%
- Complexity 5002 5016 +14
=============================================
Files 634 635 +1
Lines 42529 42508 -21
Branches 5573 5563 -10
=============================================
+ Hits 20381 20404 +23
+ Misses 19647 19616 -31
+ Partials 2501 2488 -13
Continue to review full report at Codecov.
|
AsyncTraceDispatcher
**
AsyncTraceDispatcher
**AsyncTraceDispatcher
} | ||
|
||
@Override | ||
public void run() { | ||
StringBuilder buffer = new StringBuilder(1024); |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
It is a good habit to catch the unchecked exception in runnable, if we do not get the future result, for the ThreadPool will swallow the exception, and we get no information if something unexpected happens.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Currently, it is enough since the code outside the try-catch is simple, but it may introduce vulnerabilities if someone else adds code to it in the future.
It is better to show the best practice if you have spare time.
@@ -358,7 +382,7 @@ private void flushData(List<TraceTransferBean> transBeanList, String dataTopic, | |||
* @param keySet the keyset in this batch(including msgId in original message not offsetMsgId) | |||
* @param data the message trace data in this batch | |||
*/ | |||
private void sendTraceDataByMQ(Set<String> keySet, final String data, String dataTopic, String regionId) { | |||
private void sendTraceDataByMQ(Set<String> keySet, final String data, String regionId) { | |||
String traceTopic = traceTopicName; |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Since the topic is generated by getTraceTopicName during the previous process.
Here is better to just use the topic generated before, and do not generate it again.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Thanks a Lot. I do want to use the topicName generated in previous process, but forget.
@dongeforever Thanks for your review, this pr is updated, please help to review. |
} | ||
|
||
@Override | ||
public void run() { | ||
StringBuilder buffer = new StringBuilder(1024); |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Currently, it is enough since the code outside the try-catch is simple, but it may introduce vulnerabilities if someone else adds code to it in the future.
It is better to show the best practice if you have spare time.
…in `AsyncTraceDispatcher` (apache#4180) * Optimized the performance of sending traceMessage * Optimized the performance of sending traceMessage * Optimized the performance of sending traceMessage * Optimized the performance of sending traceMessage
What is the purpose of the change
Details in #4099 (comment). Given the details for convenience.
Describe in Chinese
该 PR 目的是优化轨迹消息发送模块性能,详情讨论见 #4099 (comment) 。为 review 方便此处给出设计详情和背景。
当前逻辑
AsyncTraceDispatcher#append
向任务元数据队列traceContextQueue
添加TraceContext
;worker
线程循环从traceContextQueue
中获取TraceContext
、并包装为任务AsyncAppenderRequest
,由线程池traceExecutor
执行;AsyncAppenderRequest#sendTraceData
中将TraceContext
转换为可计量大小的TraceTransferBean
,并按照按照topic+regionId对TraceTransferBean
进行分组、获取结构为Map<String, List<TraceTransferBean>>
的数据;TraceTransferBean
按照maxMsgSize
进行分批次的发送,根据AsyncTraceDispatcher#accessChannel
决定发送到rmq_sys_TRACE_DATA_${regionId}
还是轨迹消息队列(默认为RMQ_SYS_TRACE_TOPIC
) 。当前逻辑的问题是
worker
通过AsyncAppenderRequest
对数据进行了不必要的分割List<TraceTransferBean>
,在AsyncAppenderRequest#sendTraceData
中对数据也进行了不必要的分组。优化思路
work
线程将数据按照topic分组、保存到HashMap<String, TraceDataSegment>
中,TraceDataSegment
中保存了TraceTransferBean列表、数据大小和当前列表第一个TraceTransferBean的保存时间;TraceDataSegment
中时会判断TraceTransferBean列表是否已经达到 maxMsgSize,是则发送数据、并恢复TraceDataSegment
的初始值;work
线程每pollingTimeMil
毫秒会遍历所有的TraceDataSegment
,检查其数据保存超过是否waitTimeThresholdMil
毫秒,是则发送所有数据、以此避免时间保存过久不发送。该算法的总体思路是在数据达到大小阈值
maxMsgSize
或者时间阈值时发送数据。可避免之前因为数据不恰当的分组导致过多的请求。