[improve] optimize collect thread prevent cpu idling (#2468)

Signed-off-by: tomsun28 <tomsun28@outlook.com>

collector/src/main/java/org/apache/hertzbeat/collector/dispatch/MetricsCollectorQueue.java

@@ -18,7 +18,6 @@
 package org.apache.hertzbeat.collector.dispatch;
 
 import java.util.concurrent.PriorityBlockingQueue;
-import java.util.concurrent.TimeUnit;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.stereotype.Component;
 
@@ -40,7 +39,7 @@ public class MetricsCollectorQueue {
     }
 
     public MetricsCollect getJob() throws InterruptedException {
-        return jobQueue.poll(2, TimeUnit.SECONDS);
+        return jobQueue.take();
     }
 
 }

collector/src/test/java/org/apache/hertzbeat/collector/collect/redis/RedisClusterCollectImplTest.java

@@ -31,6 +31,7 @@ import java.util.List;
 import org.apache.hertzbeat.common.entity.job.Metrics;
 import org.apache.hertzbeat.common.entity.job.protocol.RedisProtocol;
 import org.apache.hertzbeat.common.entity.message.CollectRep;
+import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.extension.ExtendWith;
@@ -62,7 +63,11 @@ public class RedisClusterCollectImplTest {
     void setUp() {
     }
 
-
+    @AfterEach
+    void setDown() {
+        connection.close();
+        client.shutdown();
+    }
 
     @Test
     void testCollect(){

collector/src/test/java/org/apache/hertzbeat/collector/collect/redis/RedisSingleCollectImplTest.java

@@ -29,6 +29,7 @@ import java.util.List;
 import org.apache.hertzbeat.common.entity.job.Metrics;
 import org.apache.hertzbeat.common.entity.job.protocol.RedisProtocol;
 import org.apache.hertzbeat.common.entity.message.CollectRep;
+import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.extension.ExtendWith;
@@ -69,6 +70,12 @@ class RedisSingleCollectImplTest {
                 .pattern("1")
                 .build();
     }
+   
+    @AfterEach
+    void setDown() {
+        connection.close();
+        client.shutdown();
+    }
 
     @Test
     void getInstance() {
@@ -150,5 +157,6 @@ class RedisSingleCollectImplTest {
             assertEquals(row.getColumns(1), version);
         }
         clientMockedStatic.close();
+        client.shutdown();
     }
 }

collector/src/test/java/org/apache/hertzbeat/collector/dispatch/MetricsCollectorQueueTest.java

@@ -17,10 +17,11 @@
 
 package org.apache.hertzbeat.collector.dispatch;
 
+import java.util.concurrent.locks.ReentrantLock;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 import static org.junit.jupiter.api.Assertions.assertEquals;
-import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertThrows;
 import static org.mockito.Mockito.mock;
 
 /**
@@ -47,9 +48,21 @@ class MetricsCollectorQueueTest {
     }
 
     @Test
-    void testGetJobTimeout() throws InterruptedException {
-
-        assertNull(metricsCollectorQueue.getJob());
+    void testGetJobTimeout() {
+        ReentrantLock lock = new ReentrantLock();
+        Thread run = new Thread(() -> {
+            try {
+                metricsCollectorQueue.getJob();
+            } catch (Exception e) {
+                assertThrows(InterruptedException.class, () -> {
+                    throw e;
+                });
+                lock.unlock();
+            }
+        });
+        run.start();
+        run.interrupt();
+        lock.lock();
     }
 
 }