Increase the idle check interval of the message queue to 200ms to reduce CPU usage under low load conditions (#13227)

Author: wu-sheng

docs/en/changes/changes.md

@@ -14,6 +14,7 @@
 * Support Flink monitoring.
 * BanyanDB: Support `@ShardingKey` for Measure tags and set to TopNAggregation group tag by default.
 * BanyanDB: Support cold stage data query for metrics/traces/logs.
+* Increase the idle check interval of the message queue to 200ms to reduce CPU usage under low load conditions.
 
 #### UI
 

oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/worker/MetricsAggregateWorker.java

@@ -73,7 +73,7 @@ public class MetricsAggregateWorker extends AbstractWorker<Metrics> {
             "MetricsAggregateWorker." + modelName, name, queueChannelSize, queueBufferSize, BufferStrategy.IF_POSSIBLE);
 
         BulkConsumePool.Creator creator = new BulkConsumePool.Creator(
-            name, BulkConsumePool.Creator.recommendMaxSize() * 2, 20);
+            name, BulkConsumePool.Creator.recommendMaxSize() * 2, 200);
         try {
             ConsumerPoolFactory.INSTANCE.createIfAbsent(name, creator);
         } catch (Exception e) {

oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/worker/MetricsPersistentWorker.java

@@ -127,7 +127,7 @@ public class MetricsPersistentWorker extends PersistenceWorker<Metrics> implemen
         if (size == 0) {
             size = 1;
         }
-        BulkConsumePool.Creator creator = new BulkConsumePool.Creator(name, size, 20);
+        BulkConsumePool.Creator creator = new BulkConsumePool.Creator(name, size, 200);
         try {
             ConsumerPoolFactory.INSTANCE.createIfAbsent(name, creator);
         } catch (Exception e) {

oap-server/server-library/library-datacarrier-queue/src/main/java/org/apache/skywalking/oap/server/library/datacarrier/DataCarrier.java

@@ -106,14 +106,14 @@ public DataCarrier consume(Class<? extends IConsumer<T>> consumerClass,
     }
 
     /**
-     * set consumeDriver to this Carrier. consumer begin to run when {@link DataCarrier#produce} begin to work with 20
+     * set consumeDriver to this Carrier. consumer begins to run when {@link DataCarrier#produce} begin to work with 200
      * millis consume cycle.
      *
      * @param consumerClass class of consumer
      * @param num           number of consumer threads
      */
     public DataCarrier consume(Class<? extends IConsumer<T>> consumerClass, int num) {
-        return this.consume(consumerClass, num, 20, new Properties());
+        return this.consume(consumerClass, num, 200, new Properties());
     }
 
     /**
@@ -132,14 +132,14 @@ public DataCarrier consume(IConsumer<T> consumer, int num, long consumeCycle) {
     }
 
     /**
-     * set consumeDriver to this Carrier. consumer begin to run when {@link DataCarrier#produce} begin to work with 20
+     * set consumeDriver to this Carrier. consumer begin to run when {@link DataCarrier#produce} begin to work with 200
      * millis consume cycle.
      *
      * @param consumer single instance of consumer, all consumer threads will all use this instance.
      * @param num      number of consumer threads
      */
     public DataCarrier consume(IConsumer<T> consumer, int num) {
-        return this.consume(consumer, num, 20);
+        return this.consume(consumer, num, 200);
     }
 
     /**