mirror of
https://github.com/apache/rocketmq.git
synced 2026-09-21 05:44:03 +08:00
Merge pull request #4490 from zhangjidi2016/fix_trace_lost
[ISSUE #4489]Some trace messages not being sent to the broker in time before producer shutdown.
This commit is contained in:
@@ -99,12 +99,12 @@ public class AsyncTraceDispatcher implements TraceDispatcher {
|
||||
this.traceTopicName = TopicValidator.RMQ_SYS_TRACE_TOPIC;
|
||||
}
|
||||
this.traceExecutor = new ThreadPoolExecutor(//
|
||||
10, //
|
||||
20, //
|
||||
1000 * 60, //
|
||||
TimeUnit.MILLISECONDS, //
|
||||
this.appenderQueue, //
|
||||
new ThreadFactoryImpl("MQTraceSendThread_"));
|
||||
10, //
|
||||
20, //
|
||||
1000 * 60, //
|
||||
TimeUnit.MILLISECONDS, //
|
||||
this.appenderQueue, //
|
||||
new ThreadFactoryImpl("MQTraceSendThread_"));
|
||||
traceProducer = getAndCreateTraceProducer(rpcHook);
|
||||
}
|
||||
|
||||
@@ -188,6 +188,11 @@ public class AsyncTraceDispatcher implements TraceDispatcher {
|
||||
// The maximum waiting time for refresh,avoid being written all the time, resulting in failure to return.
|
||||
long end = System.currentTimeMillis() + 500;
|
||||
while (System.currentTimeMillis() <= end) {
|
||||
synchronized (taskQueueByTopic) {
|
||||
for (TraceDataSegment taskInfo : taskQueueByTopic.values()) {
|
||||
taskInfo.sendAllData();
|
||||
}
|
||||
}
|
||||
synchronized (traceContextQueue) {
|
||||
if (traceContextQueue.size() == 0 && appenderQueue.size() == 0) {
|
||||
break;
|
||||
@@ -252,7 +257,7 @@ public class AsyncTraceDispatcher implements TraceDispatcher {
|
||||
while (System.currentTimeMillis() < endTime) {
|
||||
try {
|
||||
TraceContext traceContext = traceContextQueue.poll(
|
||||
endTime - System.currentTimeMillis(), TimeUnit.MILLISECONDS
|
||||
endTime - System.currentTimeMillis(), TimeUnit.MILLISECONDS
|
||||
);
|
||||
|
||||
if (traceContext != null && !traceContext.getTraceBeans().isEmpty()) {
|
||||
|
||||
Reference in New Issue
Block a user