feat( #506 ): 新负载均衡相关代码文件初始化

txlcn-protocol/src/main/java/com/codingapi/txlcn/protocol/message/separate/AbsMessage.java

@@ -3,6 +3,7 @@ package com.codingapi.txlcn.protocol.message.separate;
 import com.codingapi.txlcn.protocol.Protocoler;
 import com.codingapi.txlcn.protocol.message.Connection;
 import com.codingapi.txlcn.protocol.message.Message;
+import lombok.Data;
 import org.springframework.context.ApplicationContext;
 
 /**
@@ -10,8 +11,11 @@ import org.springframework.context.ApplicationContext;
  * @date 2020/3/4
  * @description
  */
+@Data
 public abstract class AbsMessage implements Message {
 
+    private Boolean isFirstNode;
+
     @Override
     public void handle(ApplicationContext springContext,
                        Protocoler protocoler,

txlcn-tc/src/main/java/com/codingapi/txlcn/tc/reporter/TxManagerReporter.java

@@ -4,15 +4,14 @@ import com.codingapi.txlcn.protocol.ProtocolServer;
 import com.codingapi.txlcn.protocol.Protocoler;
 import com.codingapi.txlcn.protocol.message.Connection;
 import com.codingapi.txlcn.protocol.message.Message;
-import com.codingapi.txlcn.protocol.message.separate.TmNodeMessage;
 import com.codingapi.txlcn.protocol.message.separate.SnowflakeMessage;
+import com.codingapi.txlcn.protocol.message.separate.TmNodeMessage;
 import com.codingapi.txlcn.protocol.message.separate.TransactionMessage;
 import com.codingapi.txlcn.tc.config.TxConfig;
 import lombok.AllArgsConstructor;
 import org.springframework.util.Assert;
 
 import java.util.Collection;
-import java.util.Iterator;
 import java.util.UUID;
 
 /**
@@ -37,12 +36,11 @@ public class TxManagerReporter {
         this.connections = protocoler.getConnections();
     }
 
-    private void selectLeader(){
-        if(connections.size()>0){
-            Iterator<Connection> iterator = connections.iterator();
-            while (iterator.hasNext()) {
-                leader = iterator.next();
-                if(leader!=null){
+    private void selectLeader() {
+        if (connections.size() > 0) {
+            for (Connection connection : connections) {
+                leader = connection;
+                if (leader != null) {
                     break;
                 }
             }
@@ -56,7 +54,7 @@ public class TxManagerReporter {
 
     /**
      * 发送消息
-     * @param message
+     * @param message message
      */
     public void sendMsg(Message message){
         selectLeader();
@@ -66,10 +64,12 @@ public class TxManagerReporter {
 
     /**
      * 请求消息
-     * @param message message
+     * @param message TmNodeMessage
      */
     public TmNodeMessage requestMsg(TmNodeMessage message){
         message.setInstanceId(UUID.randomUUID().toString());
+        // 第一个接收到 TC 消息的节点
+        message.setIsFirstNode(true);
         selectLeader();
         checkLeader();
         return leader.request(message);
@@ -77,10 +77,12 @@ public class TxManagerReporter {
 
     /**
      * 请求消息
-     * @param message
+     * @param message TransactionMessage
      */
     public TransactionMessage requestMsg(TransactionMessage message){
         message.setModuleName(txConfig.getApplicationName());
+        // 第一个接收到 TC 消息的节点
+        message.setIsFirstNode(true);
         selectLeader();
         checkLeader();
         return leader.request(message);
@@ -93,6 +95,8 @@ public class TxManagerReporter {
      */
     public SnowflakeMessage requestMsg(SnowflakeMessage message) {
         message.setInstanceId(UUID.randomUUID().toString());
+        // 第一个接收到 TC 消息的节点
+        message.setIsFirstNode(true);
         selectLeader();
         checkLeader();
         return leader.request(message);

txlcn-tm/src/main/java/com/codingapi/txlcn/protocol/loadbalancer/EventInterceptor.java

+package com.codingapi.txlcn.protocol.loadbalancer;
+
+/**
+ * @author WhomHim
+ * @description
+ * @date Create in 2020/9/9 17:48
+ */
+@FunctionalInterface
+public interface EventInterceptor {
+
+    void intercept();
+}

txlcn-tm/src/main/java/com/codingapi/txlcn/protocol/loadbalancer/InterceptorConfiguration.java

+package com.codingapi.txlcn.protocol.loadbalancer;
+
+import org.springframework.context.annotation.Configuration;
+
+/**
+ * @author WhomHim
+ * @description
+ * @date Create in 2020/9/9 17:57
+ */
+@Configuration
+public class InterceptorConfiguration {
+
+
+}

txlcn-tm/src/main/java/com/codingapi/txlcn/protocol/loadbalancer/LoadBalancerInterceptor.java

+package com.codingapi.txlcn.protocol.loadbalancer;
+
+/**
+ * @author WhomHim
+ * @description
+ * @date Create in 2020/9/9 18:06
+ */
+public class LoadBalancerInterceptor implements EventInterceptor{
+
+    @Override
+    public void intercept() {
+
+    }
+}

txlcn-tm/src/main/java/com/codingapi/txlcn/tm/node/TmNode.java

@@ -84,6 +84,10 @@ public class TmNode {
         otherNodeList.forEach(iNetSocketAddress ->
                 protocolServer.connectTo(iNetSocketAddress.getHostString(), iNetSocketAddress.getPort()));
     }
+
+    public void sentToOtherNode(){
+
+    }
 }
 
 

txlcn-tm/src/main/java/com/codingapi/txlcn/tm/reporter/ReporterConfiguration.java

+package com.codingapi.txlcn.tm.reporter;
+
+import com.codingapi.txlcn.protocol.ProtocolServer;
+import com.codingapi.txlcn.tm.config.TmConfig;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+
+/**
+ * @author whohim
+ */
+@Configuration
+public class ReporterConfiguration {
+
+    @Bean
+    public TxManagerReporter txManagerProtocolHandler(ProtocolServer protocolServer, TmConfig tmConfig) {
+        return new TxManagerReporter(protocolServer, tmConfig);
+    }
+
+
+}

txlcn-tm/src/main/java/com/codingapi/txlcn/tm/reporter/TxManagerReporter.java

+package com.codingapi.txlcn.tm.reporter;
+
+import com.codingapi.txlcn.protocol.ProtocolServer;
+import com.codingapi.txlcn.protocol.Protocoler;
+import com.codingapi.txlcn.protocol.message.Connection;
+import com.codingapi.txlcn.protocol.message.Message;
+import com.codingapi.txlcn.protocol.message.separate.SnowflakeMessage;
+import com.codingapi.txlcn.protocol.message.separate.TmNodeMessage;
+import com.codingapi.txlcn.protocol.message.separate.TransactionMessage;
+import com.codingapi.txlcn.tm.config.TmConfig;
+import lombok.AllArgsConstructor;
+import org.springframework.util.Assert;
+
+import java.util.Collection;
+import java.util.UUID;
+
+/**
+ * @author whohim
+ */
+@AllArgsConstructor
+public class TxManagerReporter {
+
+    private Protocoler protocoler;
+
+    private Collection<Connection> connections;
+
+    private Connection leader;
+
+    private TmConfig tmConfig;
+
+    public TxManagerReporter(ProtocolServer protocolServer, TmConfig tmConfig) {
+        this.protocoler =  protocolServer.getProtocoler();
+        this.tmConfig = tmConfig;
+        this.connections = protocoler.getConnections();
+    }
+
+    private void selectLeader() {
+        if (connections.size() > 0) {
+            for (Connection connection : connections) {
+                leader = connection;
+                if (leader != null) {
+                    break;
+                }
+            }
+        }
+    }
+
+
+    private void checkLeader(){
+        Assert.notNull(leader,"没有可用的TM资源.");
+    }
+
+    /**
+     * 发送消息
+     * @param message message
+     */
+    public void sendMsg(Message message){
+        selectLeader();
+        checkLeader();
+        leader.send(message);
+    }
+
+    /**
+     * 请求消息
+     * @param message TmNodeMessage
+     */
+    public TmNodeMessage requestMsg(TmNodeMessage message){
+        message.setInstanceId(UUID.randomUUID().toString());
+        selectLeader();
+        checkLeader();
+        return leader.request(message);
+    }
+
+    /**
+     * 请求消息
+     * @param message TransactionMessage
+     */
+    public TransactionMessage requestMsg(TransactionMessage message){
+        message.setModuleName(tmConfig.getName());
+        selectLeader();
+        checkLeader();
+        return leader.request(message);
+    }
+
+    /**
+     * 请求消息
+     * @param message SnowFlakeMessage
+     * @return  SnowflakeMessage
+     */
+    public SnowflakeMessage requestMsg(SnowflakeMessage message) {
+        message.setInstanceId(UUID.randomUUID().toString());
+        selectLeader();
+        checkLeader();
+        return leader.request(message);
+    }
+
+
+
+}