增加 TransportStrategy 功能

src/META-INF/application-template.xml

@@ -37,6 +37,7 @@
             threads： 线程总数， 默认: <group>节点数*CPU核数*8
             bufferCapacity: ByteBuffer的初始化大小， 默认: 8K; 
             bufferPoolSize： ByteBuffer池的大小，默认: <group>节点数*CPU核数*8
+            strategy: 远程请求的负载均衡策略， 必须是org.redkale.net.TransportStrategy的实现类
         -->
         <transport bufferCapacity="8K" bufferPoolSize="32" threads="32"/>
         

src/org/redkale/boot/Application.java

@@ -232,12 +232,14 @@ public final class Application {
         }
         this.logger = Logger.getLogger(this.getClass().getSimpleName());
         this.serversLatch = new CountDownLatch(config.getAnyValues("server").length + 1);
+        this.classLoader = new RedkaleClassLoader(Thread.currentThread().getContextClassLoader());
         logger.log(Level.INFO, "------------------------------- Redkale " + Redkale.getDotedVersion() + " -------------------------------");
         //------------------配置 <transport> 节点 ------------------
         ObjectPool<ByteBuffer> transportPool = null;
         ExecutorService transportExec = null;
         AsynchronousChannelGroup transportGroup = null;
         final AnyValue resources = config.getAnyValue("resources");
+        TransportStrategy strategy = null;
         if (resources != null) {
             AnyValue transportConf = resources.getAnyValue("transport");
             int groupsize = resources.getAnyValues("group").length;
@@ -257,6 +259,10 @@ public final class Application {
                     });
                 //-----------transportChannelGroup--------------
                 try {
+                    final String strategyClass = transportConf.getValue("strategy");
+                    if (strategyClass != null && !strategyClass.isEmpty()) {
+                        strategy = (TransportStrategy) classLoader.loadClass(strategyClass).newInstance();
+                    }
                     final AtomicInteger counter = new AtomicInteger();
                     transportExec = Executors.newFixedThreadPool(threads, (Runnable r) -> {
                         Thread t = new Thread(r);
@@ -271,8 +277,7 @@ public final class Application {
                 logger.log(Level.INFO, Transport.class.getSimpleName() + " configure bufferCapacity = " + bufferCapacity + "; bufferPoolSize = " + bufferPoolSize + "; threads = " + threads + ";");
             }
         }
-        this.transportFactory = new TransportFactory(transportExec, transportPool, transportGroup);
+        this.transportFactory = new TransportFactory(transportExec, transportPool, transportGroup, strategy);
-        this.classLoader = new RedkaleClassLoader(Thread.currentThread().getContextClassLoader());
         Thread.currentThread().setContextClassLoader(this.classLoader);
         this.serverClassLoader = new RedkaleClassLoader(this.classLoader);
     }

src/org/redkale/net/Transport.java

@@ -56,15 +56,20 @@ public final class Transport {
 
     protected final ObjectPool<ByteBuffer> bufferPool;
 
+    //负载均衡策略
+    protected final TransportStrategy strategy;
+
     protected final ConcurrentHashMap<SocketAddress, BlockingQueue<AsyncConnection>> connPool = new ConcurrentHashMap<>();
 
     public Transport(String name, String subprotocol, final ObjectPool<ByteBuffer> transportBufferPool,
-        final AsynchronousChannelGroup transportChannelGroup, final InetSocketAddress clientAddress, final Collection<InetSocketAddress> addresses) {
+        final AsynchronousChannelGroup transportChannelGroup, final InetSocketAddress clientAddress,
-        this(name, DEFAULT_PROTOCOL, subprotocol, transportBufferPool, transportChannelGroup, clientAddress, addresses);
+        final Collection<InetSocketAddress> addresses, final TransportStrategy strategy) {
+        this(name, DEFAULT_PROTOCOL, subprotocol, transportBufferPool, transportChannelGroup, clientAddress, addresses, strategy);
     }
 
     public Transport(String name, String protocol, String subprotocol, final ObjectPool<ByteBuffer> transportBufferPool,
-        final AsynchronousChannelGroup transportChannelGroup, final InetSocketAddress clientAddress, final Collection<InetSocketAddress> addresses) {
+        final AsynchronousChannelGroup transportChannelGroup, final InetSocketAddress clientAddress,
+        final Collection<InetSocketAddress> addresses, final TransportStrategy strategy) {
         this.name = name;
         this.subprotocol = subprotocol == null ? "" : subprotocol.trim();
         this.protocol = protocol;
@@ -72,6 +77,7 @@ public final class Transport {
         this.group = transportChannelGroup;
         this.bufferPool = transportBufferPool;
         this.clientAddress = clientAddress;
+        this.strategy = strategy;
         updateRemoteAddresses(addresses);
     }
 
@@ -132,6 +138,10 @@ public final class Transport {
         return remoteAddres;
     }
 
+    public ConcurrentHashMap<SocketAddress, BlockingQueue<AsyncConnection>> getAsyncConnectionPool() {
+        return connPool;
+    }
+
     @Override
     public String toString() {
         return Transport.class.getSimpleName() + "{name = " + name + ", protocol = " + protocol + ", clientAddress = " + clientAddress + ", remoteAddres = " + Arrays.toString(remoteAddres) + "}";
@@ -158,6 +168,7 @@ public final class Transport {
     }
 
     public AsyncConnection pollConnection(SocketAddress addr) {
+        if (this.strategy != null) return strategy.pollConnection(addr, this);
         if (addr == null && remoteAddres.length == 1) addr = remoteAddres[0];
         final boolean rand = addr == null;
         if (rand && remoteAddres.length < 1) throw new RuntimeException("Transport (" + this.name + ") have no remoteAddress list");

src/org/redkale/net/TransportFactory.java

@@ -43,10 +43,19 @@ public class TransportFactory {
 
     protected final List<WeakReference<Service>> services = new CopyOnWriteArrayList<>();
 
-    public TransportFactory(ExecutorService executor, ObjectPool<ByteBuffer> bufferPool, AsynchronousChannelGroup channelGroup) {
+    //负载均衡策略
+    protected final TransportStrategy strategy;
+
+    public TransportFactory(ExecutorService executor, ObjectPool<ByteBuffer> bufferPool, AsynchronousChannelGroup channelGroup,
+        final TransportStrategy strategy) {
         this.executor = executor;
         this.bufferPool = bufferPool;
         this.channelGroup = channelGroup;
+        this.strategy = strategy;
+    }
+
+    public TransportFactory(ExecutorService executor, ObjectPool<ByteBuffer> bufferPool, AsynchronousChannelGroup channelGroup) {
+        this(executor, bufferPool, channelGroup, null);
     }
 
     public String findGroupName(InetSocketAddress addr) {
@@ -127,14 +136,14 @@ public class TransportFactory {
         }
         if (info == null) return null;
         if (sncpAddress != null) addresses.remove(sncpAddress);
-        return new Transport(groups.stream().sorted().collect(Collectors.joining(";")), info.protocol, info.subprotocol, this.bufferPool, this.channelGroup, sncpAddress, addresses);
+        return new Transport(groups.stream().sorted().collect(Collectors.joining(";")), info.protocol, info.subprotocol, this.bufferPool, this.channelGroup, sncpAddress, addresses, this.strategy);
     }
 
     private Transport loadTransport(final String groupName, InetSocketAddress sncpAddress) {
         if (groupName == null) return null;
         TransportGroupInfo info = groupInfos.get(groupName);
         if (info == null) return null;
-        return new Transport(groupName, info.protocol, info.subprotocol, this.bufferPool, this.channelGroup, sncpAddress, info.addresses);
+        return new Transport(groupName, info.protocol, info.subprotocol, this.bufferPool, this.channelGroup, sncpAddress, info.addresses, this.strategy);
     }
 
     public ExecutorService getExecutor() {

src/org/redkale/net/TransportStrategy.java

@@ -0,0 +1,21 @@
+/*
+ * To change this license header, choose License Headers in Project Properties.
+ * To change this template file, choose Tools | Templates
+ * and open the template in the editor.
+ */
+package org.redkale.net;
+
+import java.net.SocketAddress;
+
+/**
+ * 远程请求的负载均衡策略
+ *
+ * <p>
+ * 详情见: https://redkale.org
+ *
+ * @author zhangjx
+ */
+public interface TransportStrategy {
+
+    public AsyncConnection pollConnection(SocketAddress addr, Transport transport);
+}