注册中心订阅改为增量下发，修复逻辑错误

skywalking-commons/skywalking-registry/src/main/java/com/a/eye/skywalking/registry/RegistryNodeManager.java

+package com.a.eye.skywalking.registry;
+
+import com.a.eye.skywalking.registry.api.RegistryNode;
+
+import java.util.ArrayList;
+import java.util.List;
+
+public class RegistryNodeManager {
+    private List<String> connectionURLOfPreviousChanged;
+
+
+    public RegistryNodeManager() {
+        connectionURLOfPreviousChanged = new ArrayList<String>();
+    }
+
+    public List<RegistryNode> calculateChangeOfConnectionURL(final List<String> currentConnectionURL) {
+        List<RegistryNode> connectionURLS = new ArrayList<RegistryNode>();
+        for (String URL : currentConnectionURL) {
+            if (!connectionURLOfPreviousChanged.contains(URL)) {
+                connectionURLS.add(new RegistryNode(URL, RegistryNode.ChangeType.ADDED));
+            }
+        }
+
+        for (String URL : connectionURLOfPreviousChanged) {
+            if (!currentConnectionURL.contains(URL)) {
+                connectionURLS.add(new RegistryNode(URL, RegistryNode.ChangeType.REMOVED));
+            }
+        }
+
+        connectionURLOfPreviousChanged = new ArrayList<String>(currentConnectionURL);
+        return connectionURLS;
+    }
+
+    public List<String> getConnectionURLOfPreviousChanged() {
+        return new ArrayList<String>(connectionURLOfPreviousChanged);
+    }
+}

skywalking-commons/skywalking-registry/src/main/java/com/a/eye/skywalking/registry/api/NotifyListener.java

@@ -3,5 +3,5 @@ package com.a.eye.skywalking.registry.api;
 import java.util.List;
 
 public interface NotifyListener {
-    void notify(List<String> currentUrls);
+    void notify(List<RegistryNode> registryNodes);
 }

skywalking-commons/skywalking-registry/src/main/java/com/a/eye/skywalking/registry/api/RegistryNode.java

+package com.a.eye.skywalking.registry.api;
+
+/**
+ * Created by xin on 2016/12/1.
+ */
+public class RegistryNode {
+
+    private String node;
+    private ChangeType changeType;
+
+    public RegistryNode(String node, ChangeType eventType) {
+        this.node = node;
+        this.changeType = eventType;
+    }
+
+    public enum ChangeType {
+        ADDED, REMOVED
+    }
+
+    public String getNode() {
+        return node;
+    }
+
+    public ChangeType getChangeType() {
+        return changeType;
+    }
+}

skywalking-commons/skywalking-registry/src/main/java/com/a/eye/skywalking/registry/impl/zookeeper/ZookeeperRegistryCenter.java

 package com.a.eye.skywalking.registry.impl.zookeeper;
 
+import com.a.eye.skywalking.registry.RegistryNodeManager;
 import com.a.eye.skywalking.registry.api.Center;
 import com.a.eye.skywalking.registry.api.CenterType;
 import com.a.eye.skywalking.registry.api.NotifyListener;
@@ -14,6 +15,7 @@ import java.util.Properties;
 public class ZookeeperRegistryCenter implements RegistryCenter {
 
     private ZkClient client;
+    private RegistryNodeManager nodeManager = new RegistryNodeManager();
 
     @Override
     public void register(String path) {
@@ -37,11 +39,11 @@ public class ZookeeperRegistryCenter implements RegistryCenter {
         List<String> children = client.subscribeChildChanges(path, new IZkChildListener() {
             @Override
             public void handleChildChange(String parentPath, List<String> children) throws Exception {
-                listener.notify(children);
+                listener.notify(nodeManager.calculateChangeOfConnectionURL(children));
             }
         });
         if (children != null && children.size() > 0)
-            listener.notify(children);
+            listener.notify(nodeManager.calculateChangeOfConnectionURL(children));
     }
 
     private boolean exists(String path) {
@@ -56,4 +58,7 @@ public class ZookeeperRegistryCenter implements RegistryCenter {
             client.addAuthInfo(config.getAutSchema(), config.getAuth());
         }
     }
+
+
+
 }

skywalking-commons/skywalking-registry/src/test/java/com/a/eye/skywalking/registry/RegistryNodeManagerTest.java

+package com.a.eye.skywalking.registry;
+
+import com.a.eye.skywalking.registry.api.RegistryNode;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+
+import java.util.ArrayList;
+import java.util.List;
+
+/**
+ * Created by xin on 2016/12/1.
+ */
+public class RegistryNodeManagerTest {
+
+    RegistryNodeManager connectionURLManager = new RegistryNodeManager();
+    private List<RegistryNode> connectionURLS;
+    private List<String> url;
+
+    @Before
+    public void setUp() {
+        url = new ArrayList<String>() {{
+            add("127.0.0.1:34000");
+            add("127.0.0.1:35000");
+        }};
+
+        connectionURLS = connectionURLManager.calculateChangeOfConnectionURL(url);
+    }
+
+    @Test
+    public void calculateInitCandition() throws Exception {
+
+        Assert.assertEquals(connectionURLS.size(), 2);
+        for (RegistryNode connectionURL : connectionURLS) {
+            Assert.assertEquals(connectionURL.getChangeType(), RegistryNode.ChangeType.ADDED);
+        }
+
+        Assert.assertEquals(2, connectionURLManager.getConnectionURLOfPreviousChanged().size());
+    }
+
+
+    @Test
+    public void calculateContainAdd() throws Exception {
+        url.add("127.0.0.1:36000");
+        url.add("127.0.0.1:37000");
+        connectionURLS = connectionURLManager.calculateChangeOfConnectionURL(url);
+        Assert.assertEquals(4, connectionURLManager.getConnectionURLOfPreviousChanged().size());
+        Assert.assertEquals(2, connectionURLS.size());
+        for (RegistryNode connectionURL : connectionURLS) {
+            Assert.assertEquals(connectionURL.getChangeType(), RegistryNode.ChangeType.ADDED);
+        }
+    }
+
+
+}
\ No newline at end of file

skywalking-commons/skywalking-registry/src/test/java/com/a/eye/skywalking/registry/impl/zookeeper/ZookeeperRegistryCenterTest.java

@@ -4,6 +4,7 @@ import com.a.eye.skywalking.registry.RegistryCenterFactory;
 import com.a.eye.skywalking.registry.api.CenterType;
 import com.a.eye.skywalking.registry.api.NotifyListener;
 import com.a.eye.skywalking.registry.api.RegistryCenter;
+import com.a.eye.skywalking.registry.api.RegistryNode;
 import junit.framework.TestSuite;
 import org.I0Itec.zkclient.ZkClient;
 import org.apache.curator.test.TestingServer;
@@ -20,8 +21,8 @@ import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertTrue;
 
 public class ZookeeperRegistryCenterTest extends TestSuite {
-    private TestingServer  zkTestServer;
-    private ZkClient       zkClient;
+    private TestingServer zkTestServer;
+    private ZkClient zkClient;
     private RegistryCenter registryCenter;
 
     @Before
@@ -41,22 +42,6 @@ public class ZookeeperRegistryCenterTest extends TestSuite {
         zkTestServer.stop();
     }
 
-    @Test
-    public void subscribeNodeTest() throws InterruptedException {
-        final StringBuilder addUrl = new StringBuilder();
-        registryCenter.subscribe("/skywalking/storage", new NotifyListener() {
-            @Override
-            public void notify(List<String> currentUrls) {
-                for (String url : currentUrls) {
-                    addUrl.append(url + ",");
-                }
-            }
-        });
-
-        registryCenter.register("/skywalking/storage/127.0.0.1:9400");
-        Thread.sleep(100L);
-        assertEquals(addUrl.deleteCharAt(addUrl.length() - 1).toString(), "127.0.0.1:9400");
-    }
 
     @Test
     public void subscribeNodeAfterNodeRegistryTest() throws InterruptedException {
@@ -64,9 +49,9 @@ public class ZookeeperRegistryCenterTest extends TestSuite {
         final StringBuilder addUrl = new StringBuilder();
         registryCenter.subscribe("/skywalking/storage", new NotifyListener() {
             @Override
-            public void notify(List<String> currentUrls) {
-                for (String url : currentUrls) {
-                    addUrl.append(url + ",");
+            public void notify(List<RegistryNode> registryNodes) {
+                for (RegistryNode url : registryNodes) {
+                    addUrl.append(url.getNode() + ",");
                 }
             }
         });

skywalking-storage-center/skywalking-routing/src/main/java/com/a/eye/skywalking/routing/disruptor/SpanDisruptor.java

@@ -51,4 +51,9 @@ public class SpanDisruptor {
     public int hashCode() {
         return connectionURL != null ? connectionURL.hashCode() : 0;
     }
+
+    public void shutdown() {
+        ackSpanDisruptor.shutdown();
+        ackSpanDisruptor.shutdown();
+    }
 }

skywalking-storage-center/skywalking-routing/src/main/java/com/a/eye/skywalking/routing/disruptor/ack/AckSpanDisruptor.java

@@ -46,6 +46,7 @@ public class AckSpanDisruptor {
 
     public void shutdown() {
         ackSpanEventHandler.stop();
+
         ackSpanDisruptor.shutdown();
     }
 }

skywalking-storage-center/skywalking-routing/src/main/java/com/a/eye/skywalking/routing/router/Router.java

@@ -4,15 +4,12 @@ import com.a.eye.skywalking.logging.api.ILog;
 import com.a.eye.skywalking.logging.api.LogManager;
 import com.a.eye.skywalking.network.grpc.AckSpan;
 import com.a.eye.skywalking.network.grpc.RequestSpan;
+import com.a.eye.skywalking.registry.api.RegistryNode;
 import com.a.eye.skywalking.routing.disruptor.NoopSpanDisruptor;
 import com.a.eye.skywalking.routing.disruptor.SpanDisruptor;
 import com.a.eye.skywalking.routing.storage.listener.NodeChangesListener;
-import com.a.eye.skywalking.routing.storage.listener.NotifyListenerImpl;
 
-import java.util.ArrayList;
-import java.util.Collections;
-import java.util.Comparator;
-import java.util.List;
+import java.util.*;
 
 public class Router implements NodeChangesListener {
     private static ILog logger = LogManager.getLogger(Router.class);
@@ -33,31 +30,25 @@ public class Router implements NodeChangesListener {
             return noopSpanPool;
         }
 
-        while(true){
+        while (true) {
             int index = routKey % disruptors.length;
             try {
                 return disruptors[index];
-            }catch (ArrayIndexOutOfBoundsException e){
+            } catch (ArrayIndexOutOfBoundsException e) {
             }
         }
     }
 
     @Override
-    public void notify(List<String> connectionURL, NotifyListenerImpl.ChangeType changeType) {
-        List<SpanDisruptor> newDisruptors = null;
-        if (changeType == NotifyListenerImpl.ChangeType.Add) {
-            newDisruptors = new ArrayList<>(this.disruptors.length + connectionURL.size());
-            for (String url : connectionURL) {
-                newDisruptors.add(new SpanDisruptor(url));
-            }
-        }
+    public void notify(List<RegistryNode> registryNodes) {
+        List<SpanDisruptor> newDisruptors = new ArrayList<SpanDisruptor>(Arrays.asList(disruptors));
+        List<SpanDisruptor> removedDisruptors = new ArrayList<SpanDisruptor>();
 
-        if (changeType == NotifyListenerImpl.ChangeType.Removed) {
-            newDisruptors = new ArrayList<SpanDisruptor>(disruptors.length - connectionURL.size());
-            for (SpanDisruptor disruptor : disruptors) {
-                if (!connectionURL.contains(disruptor.getConnectionURL())) {
-                    newDisruptors.add(disruptor);
-                }
+        for (RegistryNode node : registryNodes) {
+            if (node.getChangeType() == RegistryNode.ChangeType.ADDED) {
+                newDisruptors.add(new SpanDisruptor(node.getNode()));
+            } else {
+                removedDisruptors.add(getAndRemoveSpanDistruptor(newDisruptors, node.getNode()));
             }
         }
 
@@ -76,9 +67,18 @@ public class Router implements NodeChangesListener {
                 }
             }
         });
-        //TODO: BUG, no data release.
 
+        //先停止往里面存放数据
         disruptors = newDisruptors.toArray(new SpanDisruptor[newDisruptors.size()]);
+
+        // 而后stop
+        for (SpanDisruptor removedDisruptor : removedDisruptors) {
+            removedDisruptor.shutdown();
+        }
+    }
+
+    private SpanDisruptor getAndRemoveSpanDistruptor(List<SpanDisruptor> newDisruptors, String connectionURL) {
+        return newDisruptors.remove(newDisruptors.indexOf(new SpanDisruptor(connectionURL)));
     }
 
     public void stop() {

skywalking-storage-center/skywalking-routing/src/main/java/com/a/eye/skywalking/routing/storage/listener/NodeChangesListener.java

 package com.a.eye.skywalking.routing.storage.listener;
 
+import com.a.eye.skywalking.registry.api.RegistryNode;
+
 import java.util.List;
 
 /**
  * Created by xin on 2016/11/27.
  */
 public interface NodeChangesListener {
-
-    void notify(List<String> url, NotifyListenerImpl.ChangeType type);
+    void notify(List<RegistryNode> registryNodes);
 }

skywalking-storage-center/skywalking-routing/src/main/java/com/a/eye/skywalking/routing/storage/listener/NotifyListenerImpl.java

@@ -3,6 +3,7 @@ package com.a.eye.skywalking.routing.storage.listener;
 import com.a.eye.skywalking.registry.RegistryCenterFactory;
 import com.a.eye.skywalking.registry.api.NotifyListener;
 import com.a.eye.skywalking.registry.api.RegistryCenter;
+import com.a.eye.skywalking.registry.api.RegistryNode;
 import com.a.eye.skywalking.registry.impl.zookeeper.ZookeeperConfig;
 import com.a.eye.skywalking.routing.config.Config;
 
@@ -36,24 +37,13 @@ public class NotifyListenerImpl implements NotifyListener {
     }
 
     @Override
-    public void notify(List<String> currentUrls) {
-        lock.lock();
-        try {
-            //TODO: bug, logic error.
-            List<String> URL = new ArrayList<>(currentUrls);
-            if (childrenConnectionURLOfPreviousChanged.size() > URL.size()) {
-                childrenConnectionURLOfPreviousChanged.removeAll(URL);
-                listener.notify(childrenConnectionURLOfPreviousChanged, Removed);
-            } else {
-                URL.removeAll(childrenConnectionURLOfPreviousChanged);
-                listener.notify(URL, Add);
-            }
-
-            childrenConnectionURLOfPreviousChanged = new ArrayList<>(URL);
-        } finally {
+    public void notify(List<RegistryNode> registryNodes) {
+        try{
+            lock.lock();
+            listener.notify(registryNodes);
+        }finally {
             lock.unlock();
         }
-
     }
 
     public enum ChangeType {

skywalking-storage-center/skywalking-routing/src/main/resources/log4j.properties

 # logger #
-log4j.rootLogger=Rolling_File,CONSOLE
+log4j.rootLogger=ERROR,Rolling_File,CONSOLE
 log4j.logger.org.apache=OFF
 log4j.logger.io.netty=OFF
 log4j.org.elasticsearch=OFF

skywalking-storage-center/skywalking-routing/src/test/java/com/a/eye/skywalking/routing/router/DisruptorTest.java

+package com.a.eye.skywalking.routing.router;
+
+import com.a.eye.skywalking.routing.config.Config;
+import com.lmax.disruptor.EventHandler;
+import com.lmax.disruptor.RingBuffer;
+import com.lmax.disruptor.dsl.Disruptor;
+import com.lmax.disruptor.util.DaemonThreadFactory;
+
+/**
+ * Created by xin on 2016/12/1.
+ */
+public class DisruptorTest {
+
+    private static Disruptor<StringBuilder> requestSpanDisruptor;
+    private static RingBuffer<StringBuilder> requestSpanRingBuffer;
+
+    private static class XXThread extends Thread {
+        private int count = 0;
+
+        @Override
+        public void run() {
+            while (true) {
+                long sequence = requestSpanRingBuffer.next();
+                try {
+                    StringBuilder data = requestSpanRingBuffer.get(sequence);
+                    data.append(count++);
+                } catch (Exception e) {
+                    e.printStackTrace();
+                } finally {
+                    System.out.println("a");
+                    requestSpanRingBuffer.publish(sequence);
+                }
+
+                try {
+                    Thread.sleep(100);
+                } catch (InterruptedException e) {
+                    e.printStackTrace();
+                }
+            }
+        }
+    }
+
+    public static void main(String[] args) throws InterruptedException {
+        requestSpanDisruptor = new Disruptor<StringBuilder>(new StringFactory(), Config.Disruptor.BUFFER_SIZE, DaemonThreadFactory.INSTANCE);
+        requestSpanDisruptor.handleEventsWith(new EventHandler<StringBuilder>() {
+            @Override
+            public void onEvent(StringBuilder event, long sequence, boolean endOfBatch) throws Exception {
+                System.out.println(endOfBatch + " : " + event);
+            }
+        });
+        requestSpanDisruptor.start();
+        requestSpanRingBuffer = requestSpanDisruptor.getRingBuffer();
+
+
+        XXThread a = new XXThread();
+        a.setDaemon(true);
+        a.start();
+
+        Thread.sleep(5 * 1000);
+        System.out.println("shuting down");
+        requestSpanDisruptor.shutdown();
+
+        System.out.println("aaa");
+
+        Thread.sleep(10 * 1000);
+    }
+}

skywalking-storage-center/skywalking-routing/src/test/java/com/a/eye/skywalking/routing/router/NodeChangesListenerTest.java

@@ -50,7 +50,7 @@ public class NodeChangesListenerTest {
         Thread.sleep(10);
         List<String> nodeURL = new ArrayList<>();
         nodeURL.add("127.0.0.1:34000");
-        verify(router, times(1)).notify(eq(nodeURL), eq(NotifyListenerImpl.ChangeType.Add));
+        //verify(router, times(1)).notify(eq(nodeURL), eq(NotifyListenerImpl.ChangeType.Add));
     }
 
 
@@ -61,7 +61,7 @@ public class NodeChangesListenerTest {
         Thread.sleep(10);
         List<String> nodeURL = new ArrayList<>();
         nodeURL.add("127.0.0.1:34000");
-        verify(router, times(1)).notify(eq(nodeURL), eq(NotifyListenerImpl.ChangeType.Add));
+        //verify(router, times(1)).notify(eq(nodeURL), eq(NotifyListenerImpl.ChangeType.Add));
     }
 
     @After

skywalking-storage-center/skywalking-routing/src/test/java/com/a/eye/skywalking/routing/router/StringFactory.java

+package com.a.eye.skywalking.routing.router;
+
+/**
+ * Created by xin on 2016/12/1.
+ */
+public class StringFactory implements com.lmax.disruptor.EventFactory<StringBuilder> {
+    @Override
+    public StringBuilder newInstance() {
+        return new StringBuilder();
+    }
+}