TOOD:调度

src/main/java/com/yuandian/dataflow/Server.java

@@ -4,6 +4,7 @@ package com.yuandian.dataflow;
 import com.yuandian.dataflow.statemachine.RaftClosure;
 import com.yuandian.dataflow.statemachine.StateMachine;
 
+import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.boot.SpringApplication;
 import org.springframework.boot.SpringBootConfiguration;
 import org.springframework.boot.autoconfigure.SpringBootApplication;
@@ -51,6 +52,14 @@ import com.alipay.sofa.jraft.option.ReplicatorGroupOptions;
 @SpringBootApplication
 public class Server {
     
+    @Autowired
+    public static Node node;
+    
+    public static Node GetNode()  {
+        return node;
+    }   
+ 
+
     public static void main(String[] args) {
 
         String[] peers = new String[]{"localhost:4440","localhost:4441","localhost:4442"};
@@ -82,20 +91,13 @@ public class Server {
         nodeOptions.setFsm(new StateMachine());
 
         RaftGroupService cluster = new RaftGroupService(groupId, serverId, nodeOptions);
-        Node node = cluster.start();
-
-        Closure done = new RaftClosure();
-        Task task = new Task();
-        task.setData(ByteBuffer.wrap("hello".getBytes()));
-        task.setDone(done);
-        node.apply(task);
-
+        node = cluster.start();
 
         System.setProperty("server.port", sprPort);
         SpringApplication.run(Server.class, args);
  
  
-
+        Closure done = new RaftClosure();
         node.shutdown(done);
     }
 }

src/main/java/com/yuandian/dataflow/controller/TaskLog.java

@@ -1,25 +1,53 @@
 package com.yuandian.dataflow.controller;
 
-import com.google.gson.JsonObject;
+import java.nio.ByteBuffer;
 
+import com.alipay.sofa.jraft.Closure;
+import com.alipay.sofa.jraft.entity.Task;
+import com.google.gson.JsonObject;
+import com.yuandian.dataflow.Server;
+import com.yuandian.dataflow.statemachine.RaftClosure;
+
+import org.apache.commons.lang.ObjectUtils.Null;
+import org.apache.commons.logging.Log;
+import org.apache.ratis.thirdparty.org.checkerframework.common.reflection.qual.GetMethod;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 import org.springframework.http.HttpStatus;
 import org.springframework.http.MediaType;
 import org.springframework.http.ResponseEntity;
 import org.springframework.stereotype.Controller;
+import org.springframework.web.bind.annotation.GetMapping;
 import org.springframework.web.bind.annotation.PostMapping;
 import org.springframework.web.bind.annotation.RequestBody;
 
 @Controller
 public class TaskLog {
 
-    @PostMapping(path = "/test", produces = MediaType.APPLICATION_JSON_VALUE)
-	public ResponseEntity<JsonObject> greeting(@RequestBody JsonObject input) {
-	 
-		String message = (String) input.get("message").getAsString();
-		System.out.println(message);
+	private static final Logger LOG = LoggerFactory.getLogger(TaskLog.class);
+
+    @GetMapping(path = "/test")
+	public ResponseEntity<JsonObject> greeting() {
+
+		Task task = new Task();
+        Closure done = new RaftClosure();
+        task.setData(ByteBuffer.wrap("hello".getBytes()));
+        task.setDone( done);
+		LOG.info( done.toString());
+		LOG.info(Server.GetNode().toString());
 		JsonObject response  = new JsonObject();
+
+		if (Server.GetNode() == null) {
+			return new ResponseEntity<JsonObject>(response, HttpStatus.OK);
+		}
+        Server.GetNode()
+		.apply(task);
+	 
+ 
+		
        
 		response.addProperty("status", "success");
+		response.addProperty("apply", "hello");
 		return new ResponseEntity<JsonObject>(response, HttpStatus.OK);
 	}
 }

src/main/java/com/yuandian/dataflow/statemachine/RaftClosure.java

@@ -3,14 +3,19 @@ package com.yuandian.dataflow.statemachine;
 import com.alipay.sofa.jraft.Closure;
 import com.alipay.sofa.jraft.Status;
 
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
 public class RaftClosure implements Closure {
 
+	private static final Logger LOG = LoggerFactory.getLogger(StateMachine.class);
+
 	@Override
 	public void run(Status status) {
 			
-		System.out.println("Task completed with status"+status.getCode());
-		System.out.println("Task completed with "+status.getErrorMsg());
-		System.out.println("Task completed with "+status.getRaftError());
+		LOG.info("Task completed with status"+status.getCode());
+		LOG.info("Task completed with "+status.getErrorMsg());
+		LOG.info("Task completed with "+status.getRaftError());
 	}
 
 //	@Override

src/main/java/com/yuandian/dataflow/statemachine/StateMachine.java

@@ -1,7 +1,5 @@
 package com.yuandian.dataflow.statemachine;
 
- 
-
 import java.io.File;
 import java.io.IOException;
 import java.nio.ByteBuffer;
@@ -25,20 +23,20 @@ import com.alipay.sofa.jraft.util.Utils;
  *
  * @author boyan (boyan@alibaba-inc.com)
  *
- * 2018-Apr-09 4:52:31 PM
+ *         2018-Apr-09 4:52:31 PM
  */
 public class StateMachine extends StateMachineAdapter {
 
-    private static final Logger LOG        = LoggerFactory.getLogger(StateMachine.class);
+    private static final Logger LOG = LoggerFactory.getLogger(StateMachine.class);
 
     /**
      * Counter value
      */
-    private final AtomicLong    value      = new AtomicLong(0);
+    private final AtomicLong value = new AtomicLong(0);
     /**
      * Leader term
      */
-    private final AtomicLong    leaderTerm = new AtomicLong(-1);
+    private final AtomicLong leaderTerm = new AtomicLong(-1);
 
     public boolean isLeader() {
         return this.leaderTerm.get() > 0;
@@ -54,15 +52,17 @@ public class StateMachine extends StateMachineAdapter {
     @Override
     public void onApply(final Iterator iter) {
         while (iter.hasNext()) {
-         
+
             if (iter.done() != null) {
-                // This task is applied by this node, get value from closure to avoid additional parsing.
-                
+                // This task is applied by this node, get value from closure to avoid additional
+                // parsing.
+        
+                LOG.debug("done:%s",iter.getData().toString());
             } else {
                 // Have to parse FetchAddRequest from this user log.
-              
+                LOG.debug("null:%s",iter.getData().toString());
             }
-      
+
             iter.next();
         }
     }
@@ -79,7 +79,7 @@ public class StateMachine extends StateMachineAdapter {
 
     @Override
     public boolean onSnapshotLoad(final SnapshotReader reader) {
-        
+
         return true;
     }
 

stop.sh

@@ -0,0 +1,3 @@
+screen -S raft-0 -X quit
+screen -S raft-1 -X quit
+screen -S raft-2 -X quit