Fix ConsumeDriver running status

Author: JoeCqupt

apm-commons/apm-datacarrier/src/main/java/org/apache/skywalking/apm/commons/datacarrier/consumer/ConsumeDriver.java

@@ -27,7 +27,7 @@
  * Pool of consumers <p> Created by wusheng on 2016/10/25.
  */
 public class ConsumeDriver<T> implements IDriver {
-    private boolean running;
+    private volatile boolean running;
     private ConsumerThread[] consumerThreads;
     private Channels<T> channels;
     private ReentrantLock lock;
@@ -88,6 +88,9 @@ public void begin(Channels channels) {
         }
         lock.lock();
         try {
+            if (running){
+                return;
+            }
             this.allocateBuffer2Thread();
             for (ConsumerThread consumerThread : consumerThreads) {
                 consumerThread.start();
@@ -124,8 +127,14 @@ private void allocateBuffer2Thread() {
 
     @Override
     public void close(Channels channels) {
+        if (!running) {
+            return;
+        }
         lock.lock();
         try {
+            if (!running) {
+                return;
+            }
             this.running = false;
             for (ConsumerThread consumerThread : consumerThreads) {
                 consumerThread.shutdown();