no message

skywalking-collector/skywalking-collector-cluster/src/main/java/com/a/eye/skywalking/collector/actor/AbstractAsyncMember.java

+package com.a.eye.skywalking.collector.actor;
+
+import akka.actor.ActorRef;
+import com.a.eye.skywalking.collector.queue.MessageHolder;
+import com.lmax.disruptor.RingBuffer;
+
+/**
+ * @author pengys5
+ */
+public abstract class AbstractAsyncMember extends AbstractMember {
+
+    private RingBuffer<MessageHolder> ringBuffer;
+
+    public AbstractAsyncMember(RingBuffer<MessageHolder> ringBuffer, ActorRef actorRef) {
+        super(actorRef);
+        this.ringBuffer = ringBuffer;
+    }
+
+    public void onEvent(MessageHolder event, long sequence, boolean endOfBatch) throws Exception {
+        Object message = event.getMessage();
+        event.reset();
+        receive(message);
+    }
+
+    public void beTold(Object message) throws Exception {
+        long sequence = ringBuffer.next();
+        try {
+            ringBuffer.get(sequence).setMessage(message);
+        } finally {
+            ringBuffer.publish(sequence);
+        }
+    }
+}

skywalking-collector/skywalking-collector-cluster/src/main/java/com/a/eye/skywalking/collector/actor/AbstractASyncMemberProvider.java → skywalking-collector/skywalking-collector-cluster/src/main/java/com/a/eye/skywalking/collector/actor/AbstractAsyncMemberProvider.java

@@ -3,7 +3,7 @@ package com.a.eye.skywalking.collector.actor;
 import akka.actor.ActorRef;
 import com.a.eye.skywalking.collector.queue.DaemonThreadFactory;
 import com.a.eye.skywalking.collector.queue.MessageHolder;
-import com.lmax.disruptor.EventFactory;
+import com.a.eye.skywalking.collector.queue.MessageHolderFactory;
 import com.lmax.disruptor.EventHandler;
 import com.lmax.disruptor.RingBuffer;
 import com.lmax.disruptor.dsl.Disruptor;
@@ -13,49 +13,29 @@ import java.lang.reflect.Constructor;
 /**
  * @author pengys5
  */
-public abstract class AbstractASyncMemberProvider<T extends EventHandler> {
+public abstract class AbstractAsyncMemberProvider<T extends EventHandler> extends AbstractMemberProvider<T> {
 
-    private RingBuffer<MessageHolder> ringBuffer;
+    public abstract int queueSize();
 
-    public abstract Class memberClass();
-
-    public T createWorker(EventFactory eventFactory, ActorRef actorRef) throws Exception {
+    @Override
+    public T createWorker(ActorRef actorRef) throws Exception {
         if (memberClass() == null) {
             throw new IllegalArgumentException("cannot createInstance() with nothing obtained from memberClass()");
         }
 
-        Constructor memberConstructor = memberClass().getDeclaredConstructor(new Class[]{ActorRef.class});
-        memberConstructor.setAccessible(true);
-        T member = (T) memberConstructor.newInstance(actorRef);
-
         // Specify the size of the ring buffer, must be power of 2.
-        int bufferSize = 1024;
+        int bufferSize = queueSize();
         // Construct the Disruptor
-        Disruptor<MessageHolder> disruptor = new Disruptor<MessageHolder>(eventFactory, bufferSize, DaemonThreadFactory.INSTANCE);
-        // Connect the handler
-        disruptor.handleEventsWith(member);
+        Disruptor<MessageHolder> disruptor = new Disruptor<MessageHolder>(MessageHolderFactory.INSTANCE, bufferSize, DaemonThreadFactory.INSTANCE);
         // Start the Disruptor, starts all threads running
-        disruptor.start();
-        // Get the ring buffer from the Disruptor to be used for publishing.
-        ringBuffer = disruptor.getRingBuffer();
-        return member;
-    }
+        RingBuffer<MessageHolder> ringBuffer = disruptor.start();
 
-    public void onData(MessageHolder message) {
-        long sequence = ringBuffer.next();
-        try {
-            ringBuffer.get(sequence).setMessage(message);
-        } finally {
-            ringBuffer.publish(sequence);
-        }
-    }
+        Constructor memberConstructor = memberClass().getDeclaredConstructor(new Class<?>[]{RingBuffer.class, ActorRef.class});
+        memberConstructor.setAccessible(true);
+        T member = (T) memberConstructor.newInstance(ringBuffer, actorRef);
 
-    /**
-     * Use {@link #memberClass()} method returned class's simple name as a role name.
-     *
-     * @return is role of Worker
-     */
-    protected String roleName() {
-        return memberClass().getSimpleName();
+        // Connect the handler
+        disruptor.handleEventsWith(member);
+        return member;
     }
 }

skywalking-collector/skywalking-collector-cluster/src/main/java/com/a/eye/skywalking/collector/actor/AbstractMember.java

@@ -13,13 +13,13 @@ import java.util.List;
 /**
  * @author pengys5
  */
-public abstract class AbstractMember<T> implements EventHandler<MessageHolder<T>> {
+public abstract class AbstractMember implements EventHandler<MessageHolder> {
 
     private Logger logger = LogManager.getFormatterLogger(AbstractMember.class);
 
     private ActorRef actorRef;
 
-    public ActorRef getSelf() {
+    private ActorRef getSelf() {
         return actorRef;
     }
 
@@ -27,6 +27,8 @@ public abstract class AbstractMember<T> implements EventHandler<MessageHolder<T>
         this.actorRef = actorRef;
     }
 
+    public abstract void beTold(Object message) throws Exception;
+
     /**
      * Receive the message to analyse.
      *
@@ -35,12 +37,6 @@ public abstract class AbstractMember<T> implements EventHandler<MessageHolder<T>
      */
     public abstract void receive(Object message) throws Exception;
 
-    public void onEvent(MessageHolder<T> event, long sequence, boolean endOfBatch) throws Exception {
-        T message = event.getMessage();
-        event.reset();
-        receive(message);
-    }
-
     /**
      * Send analysed data to next Worker.
      *
@@ -49,7 +45,7 @@ public abstract class AbstractMember<T> implements EventHandler<MessageHolder<T>
      * @param message              is the data used to send to next worker.
      * @throws Exception
      */
-    public void tell(AbstractWorkerProvider targetWorkerProvider, WorkerSelector selector, T message) throws Exception {
+    public void tell(AbstractWorkerProvider targetWorkerProvider, WorkerSelector selector, Object message) throws Exception {
         logger.debug("worker provider: %s ,role name: %s", targetWorkerProvider.getClass().getName(), targetWorkerProvider.roleName());
         List<WorkerRef> availableWorks = WorkersRefCenter.INSTANCE.availableWorks(targetWorkerProvider.roleName());
         selector.select(availableWorks, message).tell(message, getSelf());

skywalking-collector/skywalking-collector-cluster/src/main/java/com/a/eye/skywalking/collector/actor/AbstractMemberProvider.java

+package com.a.eye.skywalking.collector.actor;
+
+import akka.actor.ActorRef;
+
+/**
+ * @author pengys5
+ */
+public abstract class AbstractMemberProvider<T> {
+
+    public abstract Class memberClass();
+
+    public abstract T createWorker(ActorRef actorRef) throws Exception;
+}

skywalking-collector/skywalking-collector-cluster/src/main/java/com/a/eye/skywalking/collector/actor/AbstractSyncMember.java

+package com.a.eye.skywalking.collector.actor;
+
+import akka.actor.ActorRef;
+import com.a.eye.skywalking.collector.queue.MessageHolder;
+
+/**
+ * @author pengys5
+ */
+public abstract class AbstractSyncMember extends AbstractMember {
+
+    public AbstractSyncMember(ActorRef actorRef) {
+        super(actorRef);
+    }
+
+    @Override
+    public void onEvent(MessageHolder event, long sequence, boolean endOfBatch) throws Exception {
+    }
+
+    @Override
+    public void beTold(Object message) throws Exception {
+        receive(message);
+    }
+}

skywalking-collector/skywalking-collector-cluster/src/main/java/com/a/eye/skywalking/collector/actor/AbstractSyncMemberProvider.java

@@ -7,27 +7,17 @@ import java.lang.reflect.Constructor;
 /**
  * @author pengys5
  */
-public abstract class AbstractSyncMemberProvider<T> {
-
-    public abstract Class memberClass();
+public abstract class AbstractSyncMemberProvider<T> extends AbstractMemberProvider<T> {
 
+    @Override
     public T createWorker(ActorRef actorRef) throws Exception {
         if (memberClass() == null) {
             throw new IllegalArgumentException("cannot createInstance() with nothing obtained from memberClass()");
         }
 
-        Constructor memberConstructor = memberClass().getDeclaredConstructor(new Class[]{ActorRef.class});
+        Constructor memberConstructor = memberClass().getDeclaredConstructor(new Class<?>[]{ActorRef.class});
         memberConstructor.setAccessible(true);
         T member = (T) memberConstructor.newInstance(actorRef);
         return member;
     }
-
-    /**
-     * Use {@link #memberClass()} method returned class's simple name as a role name.
-     *
-     * @return is role of Worker
-     */
-    protected String roleName() {
-        return memberClass().getSimpleName();
-    }
 }

skywalking-collector/skywalking-collector-cluster/src/main/java/com/a/eye/skywalking/collector/actor/AbstractWorker.java

@@ -38,7 +38,7 @@ import java.util.List;
  * }
  * }}}
  */
-public abstract class AbstractWorker<T> extends UntypedActor {
+public abstract class AbstractWorker extends UntypedActor {
 
     private Logger logger = LogManager.getFormatterLogger(AbstractWorker.class);
 
@@ -96,7 +96,7 @@ public abstract class AbstractWorker<T> extends UntypedActor {
      * @param message              is the data used to send to next worker.
      * @throws Throwable
      */
-    public void tell(AbstractWorkerProvider targetWorkerProvider, WorkerSelector selector, T message) throws Throwable {
+    public void tell(AbstractWorkerProvider targetWorkerProvider, WorkerSelector selector, Object message) throws Throwable {
         List<WorkerRef> availableWorks = WorkersRefCenter.INSTANCE.availableWorks(targetWorkerProvider.roleName());
         selector.select(availableWorks, message).tell(message, getSelf());
     }

skywalking-collector/skywalking-collector-cluster/src/main/java/com/a/eye/skywalking/collector/actor/selector/WorkerSelector.java

@@ -13,7 +13,7 @@ import java.util.List;
  *
  * @author wusheng
  */
-public interface WorkerSelector<T> {
+public interface WorkerSelector {
     /**
      * select a {@link WorkerRef} from a {@link WorkerRef} list.
      *
@@ -21,5 +21,5 @@ public interface WorkerSelector<T> {
      * @param message the {@link AbstractWorker} is going to send.
      * @return the selected {@link WorkerRef}
      */
-    WorkerRef select(List<WorkerRef> members, T message);
+    WorkerRef select(List<WorkerRef> members, Object message);
 }

skywalking-collector/skywalking-collector-cluster/src/main/java/com/a/eye/skywalking/collector/queue/MessageHolder.java

@@ -3,14 +3,14 @@ package com.a.eye.skywalking.collector.queue;
 /**
  * @author pengys5
  */
-public class MessageHolder<T> {
-    private T message;
+public class MessageHolder {
+    private Object message;
 
-    public T getMessage() {
+    public Object getMessage() {
         return message;
     }
 
-    public void setMessage(T message) {
+    public void setMessage(Object message) {
         this.message = message;
     }
 

skywalking-collector/skywalking-collector-cluster/src/main/java/com/a/eye/skywalking/collector/queue/MessageHolderFactory.java

+package com.a.eye.skywalking.collector.queue;
+
+import com.lmax.disruptor.EventFactory;
+
+/**
+ * @author pengys5
+ */
+public class MessageHolderFactory implements EventFactory<MessageHolder> {
+
+    public static MessageHolderFactory INSTANCE = new MessageHolderFactory();
+
+    public MessageHolder newInstance() {
+        return new MessageHolder();
+    }
+}

skywalking-collector/skywalking-collector-worker/src/main/java/com/a/eye/skywalking/collector/worker/PersistenceMember.java

 package com.a.eye.skywalking.collector.worker;
 
 import akka.actor.ActorRef;
+import com.a.eye.skywalking.collector.actor.AbstractAsyncMember;
 import com.a.eye.skywalking.collector.actor.AbstractMember;
+import com.a.eye.skywalking.collector.queue.MessageHolder;
 import com.a.eye.skywalking.collector.worker.storage.EsClient;
 import com.google.gson.JsonObject;
+import com.lmax.disruptor.RingBuffer;
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
 import org.elasticsearch.action.bulk.BulkRequestBuilder;
@@ -16,7 +19,7 @@ import java.util.Map;
 /**
  * @author pengys5
  */
-public abstract class PersistenceMember<T> extends AbstractMember<T> {
+public abstract class PersistenceMember extends AbstractAsyncMember {
 
     private Logger logger = LogManager.getFormatterLogger(PersistenceMember.class);
 
@@ -24,8 +27,8 @@ public abstract class PersistenceMember<T> extends AbstractMember<T> {
 
     private Map<String, JsonObject> persistenceData = new HashMap();
 
-    public PersistenceMember(ActorRef actorRef) {
-        super(actorRef);
+    public PersistenceMember(RingBuffer<MessageHolder> ringBuffer, ActorRef actorRef) {
+        super(ringBuffer, actorRef);
     }
 
     public abstract String esIndex();
@@ -54,7 +57,6 @@ public abstract class PersistenceMember<T> extends AbstractMember<T> {
         if (message instanceof PersistenceCommand) {
             persistence(false);
         } else {
-            logger.debug("receive message");
             analyse(message);
         }
     }
@@ -62,25 +64,11 @@ public abstract class PersistenceMember<T> extends AbstractMember<T> {
     private void persistence(boolean dataFull) {
         long now = System.currentTimeMillis();
         if (now - lastPersistenceTimestamp > 5000 || dataFull) {
-            boolean success = saveToEs();
+            boolean success = EsClient.saveToEs(esIndex(), esType(), persistenceData);
             if (success) {
                 persistenceData.clear();
                 lastPersistenceTimestamp = now;
             }
         }
     }
-
-    private boolean saveToEs() {
-        TransportClient client = EsClient.client();
-        BulkRequestBuilder bulkRequest = client.prepareBulk();
-
-        for (Map.Entry<String, JsonObject> entry : persistenceData.entrySet()) {
-            String id = entry.getKey();
-            JsonObject data = entry.getValue();
-            bulkRequest.add(client.prepareIndex(esIndex(), esType(), id).setSource(data.toString()));
-        }
-
-        BulkResponse bulkResponse = bulkRequest.execute().actionGet();
-        return !bulkResponse.hasFailures();
-    }
 }

skywalking-collector/skywalking-collector-worker/src/main/java/com/a/eye/skywalking/collector/worker/PersistenceWorker.java

@@ -5,9 +5,6 @@ import com.a.eye.skywalking.collector.worker.storage.EsClient;
 import com.google.gson.JsonObject;
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
-import org.elasticsearch.action.bulk.BulkRequestBuilder;
-import org.elasticsearch.action.bulk.BulkResponse;
-import org.elasticsearch.client.transport.TransportClient;
 
 import java.util.HashMap;
 import java.util.Map;
@@ -15,7 +12,7 @@ import java.util.Map;
 /**
  * @author pengys5
  */
-public abstract class PersistenceWorker<T> extends AbstractWorker<T> {
+public abstract class PersistenceWorker extends AbstractWorker {
 
     private Logger logger = LogManager.getFormatterLogger(PersistenceWorker.class);
 
@@ -57,25 +54,11 @@ public abstract class PersistenceWorker<T> extends AbstractWorker<T> {
     private void persistence(boolean dataFull) {
         long now = System.currentTimeMillis();
         if (now - lastPersistenceTimestamp > 5000 || dataFull) {
-            boolean success = saveToEs();
+            boolean success = EsClient.saveToEs(esIndex(), esType(), persistenceData);
             if (success) {
                 persistenceData.clear();
                 lastPersistenceTimestamp = now;
             }
         }
     }
-
-    private boolean saveToEs() {
-        TransportClient client = EsClient.client();
-        BulkRequestBuilder bulkRequest = client.prepareBulk();
-
-        for (Map.Entry<String, JsonObject> entry : persistenceData.entrySet()) {
-            String id = entry.getKey();
-            JsonObject data = entry.getValue();
-            bulkRequest.add(client.prepareIndex(esIndex(), esType(), id).setSource(data.toString()));
-        }
-
-        BulkResponse bulkResponse = bulkRequest.execute().actionGet();
-        return !bulkResponse.hasFailures();
-    }
 }

skywalking-collector/skywalking-collector-worker/src/main/java/com/a/eye/skywalking/collector/worker/WorkerConfig.java

@@ -37,4 +37,9 @@ public class WorkerConfig extends ClusterConfig {
         }
     }
 
+    public static class Queue {
+        public static class TraceSegmentRecordMember {
+            public static int Size = 32;
+        }
+    }
 }

skywalking-collector/skywalking-collector-worker/src/main/java/com/a/eye/skywalking/collector/worker/application/ApplicationMember.java

 package com.a.eye.skywalking.collector.worker.application;
 
 import akka.actor.ActorRef;
-import com.a.eye.skywalking.collector.actor.AbstractMember;
+import com.a.eye.skywalking.collector.actor.AbstractSyncMember;
 import com.a.eye.skywalking.collector.actor.AbstractSyncMemberProvider;
 import com.a.eye.skywalking.collector.actor.selector.RollingSelector;
 import com.a.eye.skywalking.collector.worker.application.metric.TraceSegmentRecordMember;
@@ -19,14 +19,15 @@ import org.apache.logging.log4j.Logger;
 /**
  * @author pengys5
  */
-public class ApplicationMember extends AbstractMember {
+public class ApplicationMember extends AbstractSyncMember {
 
     private Logger logger = LogManager.getFormatterLogger(ApplicationMember.class);
 
-    private TraceSegmentRecordMember traceSegmentRecordMember = TraceSegmentRecordMember.Factory.INSTANCE.createWorker(TraceSegmentRecordMember.MessageFactory.INSTANCE, getSelf());
+    private TraceSegmentRecordMember recordMember;
 
     public ApplicationMember(ActorRef actorRef) throws Exception {
         super(actorRef);
+        recordMember = TraceSegmentRecordMember.Factory.INSTANCE.createWorker(actorRef);
     }
 
     @Override
@@ -34,7 +35,7 @@ public class ApplicationMember extends AbstractMember {
         if (message instanceof TraceSegment) {
             logger.debug("begin translate TraceSegment Object to JsonObject");
             TraceSegment traceSegment = (TraceSegment) message;
-            traceSegmentRecordMember.receive(traceSegment);
+            recordMember.beTold(traceSegment);
 
             sendToDAGNodePersistence(traceSegment);
             sendToNodeInstancePersistence(traceSegment);

skywalking-collector/skywalking-collector-worker/src/main/java/com/a/eye/skywalking/collector/worker/application/metric/TraceSegmentRecordMember.java

@@ -2,18 +2,18 @@ package com.a.eye.skywalking.collector.worker.application.metric;
 
 
 import akka.actor.ActorRef;
-import com.a.eye.skywalking.collector.actor.AbstractASyncMemberProvider;
-import com.a.eye.skywalking.collector.actor.AbstractMember;
+import com.a.eye.skywalking.collector.actor.AbstractAsyncMemberProvider;
 import com.a.eye.skywalking.collector.actor.selector.RollingSelector;
 import com.a.eye.skywalking.collector.queue.MessageHolder;
 import com.a.eye.skywalking.collector.worker.PersistenceMember;
+import com.a.eye.skywalking.collector.worker.WorkerConfig;
 import com.a.eye.skywalking.collector.worker.application.persistence.TraceSegmentRecordPersistence;
 import com.a.eye.skywalking.trace.Span;
 import com.a.eye.skywalking.trace.TraceSegment;
 import com.a.eye.skywalking.trace.TraceSegmentRef;
 import com.google.gson.JsonArray;
 import com.google.gson.JsonObject;
-import com.lmax.disruptor.EventFactory;
+import com.lmax.disruptor.RingBuffer;
 
 import java.util.List;
 import java.util.Map;
@@ -33,8 +33,8 @@ public class TraceSegmentRecordMember extends PersistenceMember {
         return "trace_segment";
     }
 
-    public TraceSegmentRecordMember(ActorRef actorRef) throws Throwable {
-        super(actorRef);
+    public TraceSegmentRecordMember(RingBuffer<MessageHolder> ringBuffer, ActorRef actorRef) {
+        super(ringBuffer, actorRef);
     }
 
     @Override
@@ -47,16 +47,13 @@ public class TraceSegmentRecordMember extends PersistenceMember {
         }
     }
 
-    public static class MessageFactory implements EventFactory<MessageHolder> {
-        public static MessageFactory INSTANCE = new MessageFactory();
+    public static class Factory extends AbstractAsyncMemberProvider<TraceSegmentRecordMember> {
+        public static Factory INSTANCE = new Factory();
 
-        public MessageHolder newInstance() {
-            return new MessageHolder();
+        @Override
+        public int queueSize() {
+            return WorkerConfig.Queue.TraceSegmentRecordMember.Size;
         }
-    }
-
-    public static class Factory extends AbstractASyncMemberProvider<TraceSegmentRecordMember> {
-        public static Factory INSTANCE = new Factory();
 
         @Override
         public Class memberClass() {

skywalking-collector/skywalking-collector-worker/src/main/java/com/a/eye/skywalking/collector/worker/application/persistence/DAGNodePersistence.java

@@ -12,7 +12,7 @@ import java.io.Serializable;
 /**
  * @author pengys5
  */
-public class DAGNodePersistence extends PersistenceWorker<DAGNodePersistence.Metric> {
+public class DAGNodePersistence extends PersistenceWorker {
 
     private Logger logger = LogManager.getFormatterLogger(DAGNodePersistence.class);
 

skywalking-collector/skywalking-collector-worker/src/main/java/com/a/eye/skywalking/collector/worker/application/persistence/NodeInstancePersistence.java

@@ -10,7 +10,7 @@ import org.apache.logging.log4j.Logger;
 /**
  * @author pengys5
  */
-public class NodeInstancePersistence extends PersistenceWorker<NodeInstancePersistence.Metric> {
+public class NodeInstancePersistence extends PersistenceWorker {
 
     private Logger logger = LogManager.getFormatterLogger(NodeInstancePersistence.class);
 

skywalking-collector/skywalking-collector-worker/src/main/java/com/a/eye/skywalking/collector/worker/application/persistence/ResponseCostPersistence.java

@@ -12,7 +12,7 @@ import java.io.Serializable;
 /**
  * @author pengys5
  */
-public class ResponseCostPersistence extends PersistenceWorker<ResponseCostPersistence.Metric> {
+public class ResponseCostPersistence extends PersistenceWorker {
 
     private Logger logger = LogManager.getFormatterLogger(ResponseCostPersistence.class);
 

skywalking-collector/skywalking-collector-worker/src/main/java/com/a/eye/skywalking/collector/worker/application/persistence/ResponseSummaryPersistence.java

@@ -12,7 +12,7 @@ import java.io.Serializable;
 /**
  * @author pengys5
  */
-public class ResponseSummaryPersistence extends PersistenceWorker<ResponseSummaryPersistence.Metric> {
+public class ResponseSummaryPersistence extends PersistenceWorker {
 
     private Logger logger = LogManager.getFormatterLogger(ResponseSummaryPersistence.class);
 

skywalking-collector/skywalking-collector-worker/src/main/java/com/a/eye/skywalking/collector/worker/applicationref/presistence/DAGNodeRefPersistence.java

@@ -10,7 +10,7 @@ import java.io.Serializable;
 /**
  * @author pengys5
  */
-public class DAGNodeRefPersistence extends PersistenceWorker<DAGNodeRefPersistence.Metric> {
+public class DAGNodeRefPersistence extends PersistenceWorker {
 
     @Override
     public String esIndex() {

skywalking-collector/skywalking-collector-worker/src/main/java/com/a/eye/skywalking/collector/worker/receiver/TraceSegmentReceiver.java

@@ -16,11 +16,14 @@ public class TraceSegmentReceiver extends AbstractWorker {
 
     private Logger logger = LogManager.getFormatterLogger(TraceSegmentReceiver.class);
 
-    private ApplicationMember applicationMember = ApplicationMember.Factory.INSTANCE.createWorker(getSelf());
+    private ApplicationMember applicationMember;
+
+    private ApplicationRefMember applicationRefMember;
 
-    private ApplicationRefMember applicationRefMember = ApplicationRefMember.Factory.INSTANCE.createWorker(getSelf());
 
     public TraceSegmentReceiver() throws Exception {
+        applicationMember = ApplicationMember.Factory.INSTANCE.createWorker(getSelf());
+        applicationRefMember = ApplicationRefMember.Factory.INSTANCE.createWorker(getSelf());
     }
 
     @Override
@@ -29,8 +32,8 @@ public class TraceSegmentReceiver extends AbstractWorker {
             TraceSegment traceSegment = (TraceSegment) message;
             logger.debug("receive message instanceof TraceSegment, traceSegmentId is %s", traceSegment.getTraceSegmentId());
 
-            applicationMember.receive(traceSegment);
-            applicationRefMember.receive(traceSegment);
+            applicationMember.beTold(traceSegment);
+            applicationRefMember.beTold(traceSegment);
         }
     }
 

skywalking-collector/skywalking-collector-worker/src/main/java/com/a/eye/skywalking/collector/worker/storage/EsClient.java

 package com.a.eye.skywalking.collector.worker.storage;
 
+import com.google.gson.JsonObject;
+import org.elasticsearch.action.bulk.BulkRequestBuilder;
+import org.elasticsearch.action.bulk.BulkResponse;
 import org.elasticsearch.client.transport.TransportClient;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.transport.InetSocketTransportAddress;
@@ -7,6 +10,7 @@ import org.elasticsearch.transport.client.PreBuiltTransportClient;
 
 import java.net.InetAddress;
 import java.net.UnknownHostException;
+import java.util.Map;
 
 /**
  * @author pengys5
@@ -24,7 +28,16 @@ public class EsClient {
                 .addTransportAddress(new InetSocketTransportAddress(InetAddress.getByName("host1"), 9300));
     }
 
-    public static TransportClient client() {
-        return client;
+    public static boolean saveToEs(String esIndex, String esType, Map<String, JsonObject> persistenceData) {
+        BulkRequestBuilder bulkRequest = client.prepareBulk();
+
+        for (Map.Entry<String, JsonObject> entry : persistenceData.entrySet()) {
+            String id = entry.getKey();
+            JsonObject data = entry.getValue();
+            bulkRequest.add(client.prepareIndex(esIndex, esType, id).setSource(data.toString()));
+        }
+
+        BulkResponse bulkResponse = bulkRequest.execute().actionGet();
+        return !bulkResponse.hasFailures();
     }
 }