From eadb979ec088507cbb6ce3a13a3658c6ace542c4 Mon Sep 17 00:00:00 2001 From: zhangxin10 Date: Mon, 7 Dec 2015 18:13:02 +0800 Subject: [PATCH] =?UTF-8?q?1.=20=E5=AE=8C=E6=88=90Sender=E7=9A=84=E8=B4=9F?= =?UTF-8?q?=E8=BD=BD=E5=9D=87=E8=A1=A1=EF=BC=8C=20=E5=AE=9E=E7=8E=B0?= =?UTF-8?q?=E5=8E=9F=E7=90=86=EF=BC=9A=E4=B8=BA=E4=BA=86=E5=AE=8C=E6=88=90?= =?UTF-8?q?Sender=E7=9A=84=E8=B4=9F=E8=BD=BD=E5=9D=87=E8=A1=A1=EF=BC=8C?= =?UTF-8?q?=E9=87=87=E7=94=A8=E4=BA=86=E5=AE=9A=E6=97=B6=E5=88=87=E6=8D=A2?= =?UTF-8?q?=E5=8F=91=E9=80=81=E7=9A=84Sender=E7=9A=84=E6=9C=8D=E5=8A=A1?= =?UTF-8?q?=E5=99=A8=EF=BC=8C=E6=AF=8F=E4=B8=AA=E4=B8=80=E4=B8=AA=E6=97=B6?= =?UTF-8?q?=E9=97=B4=E9=97=B4=E9=9A=94=E5=B0=B1=E4=BC=9A=E5=B0=86DataSende?= =?UTF-8?q?r=E6=B1=A0=E4=B8=AD=E7=9A=84=E6=9F=90=E4=B8=80=E4=B8=AA=20DataS?= =?UTF-8?q?ender=E6=9B=BF=E6=8D=A2=E6=88=90=E6=B2=A1=E6=9C=89=E8=BF=9E?= =?UTF-8?q?=E6=8E=A5=E7=9A=84=E6=9C=8D=E5=8A=A1=E5=99=A8=E3=80=82?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../cloud/skywalking/buffer/BufferGroup.java | 3 +- .../com/ai/cloud/skywalking/conf/Config.java | 91 ++++---- .../cloud/skywalking/sender/DataSender.java | 32 ++- .../skywalking/sender/DataSenderFactory.java | 12 +- .../sender/DataSenderFactoryWithBalance.java | 209 ++++++++++++++++++ 5 files changed, 293 insertions(+), 54 deletions(-) create mode 100644 skywalking-api/src/main/java/com/ai/cloud/skywalking/sender/DataSenderFactoryWithBalance.java diff --git a/skywalking-api/src/main/java/com/ai/cloud/skywalking/buffer/BufferGroup.java b/skywalking-api/src/main/java/com/ai/cloud/skywalking/buffer/BufferGroup.java index 12e621287..a0c715e03 100644 --- a/skywalking-api/src/main/java/com/ai/cloud/skywalking/buffer/BufferGroup.java +++ b/skywalking-api/src/main/java/com/ai/cloud/skywalking/buffer/BufferGroup.java @@ -7,6 +7,7 @@ import com.ai.cloud.skywalking.protocol.Span; import com.ai.cloud.skywalking.selfexamination.HealthCollector; import com.ai.cloud.skywalking.selfexamination.HeathReading; import com.ai.cloud.skywalking.sender.DataSenderFactory; +import com.ai.cloud.skywalking.sender.DataSenderFactoryWithBalance; import java.util.concurrent.atomic.AtomicInteger; import java.util.logging.Level; @@ -82,7 +83,7 @@ public class BufferGroup { } if (data != null && data.length() > 0) { - while (!DataSenderFactory.getSender().send(data.toString())) { + while (!DataSenderFactoryWithBalance.getSender().send(data.toString())) { try { Thread.sleep(CONSUMER_FAIL_RETRY_WAIT_INTERVAL); } catch (InterruptedException e) { diff --git a/skywalking-api/src/main/java/com/ai/cloud/skywalking/conf/Config.java b/skywalking-api/src/main/java/com/ai/cloud/skywalking/conf/Config.java index 57c6e98fb..f293ce2b9 100644 --- a/skywalking-api/src/main/java/com/ai/cloud/skywalking/conf/Config.java +++ b/skywalking-api/src/main/java/com/ai/cloud/skywalking/conf/Config.java @@ -2,60 +2,71 @@ package com.ai.cloud.skywalking.conf; public class Config { - public static class SkyWalking { - public static String USER_ID; + public static class SkyWalking { + public static String USER_ID; - public static String APPLICATION_ID; - } + public static String APPLICATION_ID; + } - public static class BuriedPoint { - // 是否打印埋点信息 - public static boolean PRINTF = false; + public static class BuriedPoint { + // 是否打印埋点信息 + public static boolean PRINTF = false; - public static int MAX_EXCEPTION_STACK_LENGTH = 4000; + public static int MAX_EXCEPTION_STACK_LENGTH = 4000; - // Business Key 最大长度 - public static int BUSINESSKEY_MAX_LENGTH = 300; - } + // Business Key 最大长度 + public static int BUSINESSKEY_MAX_LENGTH = 300; + } - public static class Consumer { - // 最大消费线程数 - public static int MAX_CONSUMER = 2; - // 消费者最大等待时间 - public static long MAX_WAIT_TIME = 5L; + public static class Consumer { + // 最大消费线程数 + public static int MAX_CONSUMER = 2; + // 消费者最大等待时间 + public static long MAX_WAIT_TIME = 5L; - // - public static long CONSUMER_FAIL_RETRY_WAIT_INTERVAL = 50L; - } + // + public static long CONSUMER_FAIL_RETRY_WAIT_INTERVAL = 50L; + } - public static class Buffer { - // 每个Buffer的最大个数 - public static int BUFFER_MAX_SIZE = 20000; + public static class Buffer { + // 每个Buffer的最大个数 + public static int BUFFER_MAX_SIZE = 20000; - // Buffer池的最大长度 - public static int POOL_SIZE = 5; - } + // Buffer池的最大长度 + public static int POOL_SIZE = 5; + } - public static class Sender { - // 最大发送者的连接数阀比例 - public static int CONNECT_PERCENT = 50; + public static class Sender { + // 最大发送者的连接数阀比例 + public static int CONNECT_PERCENT = 50; - // 发送服务端配置 - public static String SERVERS_ADDR; + // 发送服务端配置 + public static String SERVERS_ADDR = "127.0.0.1:34000;127.0.0.1:34001;127.0.0.1:34002"; - // 是否开启发送 - public static boolean IS_OFF = false; + // 是否开启发送 + public static boolean IS_OFF = false; - // 发送的最大长度 - public static int MAX_SEND_LENGTH = 18500; + // 发送的最大长度 + public static int MAX_SEND_LENGTH = 18500; - public static long RETRY_GET_SENDER_WAIT_INTERVAL = 2000L; + public static long RETRY_GET_SENDER_WAIT_INTERVAL = 2000L; - } + //切换Sender的周期 + public static long SWITCH_SENDER_INTERVAL = 10 * 60 * 1000; + //public static long SWITCH_SENDER_INTERVAL = 10 * 1000; - public static class SenderChecker { + // 切换Sender之后，关闭Sender的倒计时 + public static long CLOSE_SENDER_COUNTDOWN = 3 * 1000; - // 检查周期时间 - public static long CHECK_POLLING_TIME = 200L; - } + // Checker线程处理完成等待周期 + public static long CHECKER_THREAD_WAIT_INTERVAL = 1000; + + public static long RETRY_FIND_CONNECTION_SENDER = 1000; + } + + public static class SenderChecker { + + // 检查周期时间 + public static long CHECK_POLLING_TIME = 200L; + } } \ No newline at end of file diff --git a/skywalking-api/src/main/java/com/ai/cloud/skywalking/sender/DataSender.java b/skywalking-api/src/main/java/com/ai/cloud/skywalking/sender/DataSender.java index a7382614c..7f4b26abb 100644 --- a/skywalking-api/src/main/java/com/ai/cloud/skywalking/sender/DataSender.java +++ b/skywalking-api/src/main/java/com/ai/cloud/skywalking/sender/DataSender.java @@ -4,7 +4,6 @@ import com.ai.cloud.skywalking.sender.protocol.ProtocolBuilder; import java.io.IOException; import java.net.InetSocketAddress; -import java.net.SocketAddress; import java.net.StandardSocketOptions; import java.nio.ByteBuffer; import java.nio.channels.SelectionKey; @@ -12,14 +11,14 @@ import java.nio.channels.Selector; import java.nio.channels.SocketChannel; public class DataSender { - private SocketChannel socketChannel; private Selector selector; - private SocketAddress socketAddress; + private InetSocketAddress socketAddress; + private SenderStatus status = SenderStatus.FAILED; public DataSender(String ip, int port) throws IOException { selector = Selector.open(); - SocketAddress isa = new InetSocketAddress(ip, port); + InetSocketAddress isa = new InetSocketAddress(ip, port); //调用open的静态方法创建连接指定的主机的SocketChannel socketChannel = SocketChannel.open(isa); //设置该sc已非阻塞的方式工作 @@ -27,9 +26,10 @@ public class DataSender { socketChannel.register(selector, SelectionKey.OP_CONNECT); socketChannel.setOption(StandardSocketOptions.SO_KEEPALIVE, true); this.socketAddress = isa; + status = SenderStatus.READY; } - public DataSender(SocketAddress address) throws IOException { + public DataSender(InetSocketAddress address) throws IOException { selector = Selector.open(); socketChannel = SocketChannel.open(address); //设置该sc已非阻塞的方式工作 @@ -37,6 +37,7 @@ public class DataSender { socketChannel.register(selector, SelectionKey.OP_CONNECT); socketChannel.setOption(StandardSocketOptions.SO_KEEPALIVE, true); this.socketAddress = address; + status = SenderStatus.READY; } /** @@ -53,13 +54,30 @@ public class DataSender { return true; } catch (IOException e) { // 发送失败 认为不可连接 - DataSenderFactory.unRegister(this); + DataSenderFactoryWithBalance.unRegister(this); return false; } } - public SocketAddress getServerIp() { + public InetSocketAddress getServerIp() { return this.socketAddress; } + public void closeConnect() throws IOException { + if (socketChannel != null) { + socketChannel.close(); + } + } + + public enum SenderStatus { + READY, FAILED, SWITCHING + } + + public SenderStatus getStatus() { + return status; + } + + public void setStatus(SenderStatus status) { + this.status = status; + } } diff --git a/skywalking-api/src/main/java/com/ai/cloud/skywalking/sender/DataSenderFactory.java b/skywalking-api/src/main/java/com/ai/cloud/skywalking/sender/DataSenderFactory.java index 7b1420433..39b80f215 100644 --- a/skywalking-api/src/main/java/com/ai/cloud/skywalking/sender/DataSenderFactory.java +++ b/skywalking-api/src/main/java/com/ai/cloud/skywalking/sender/DataSenderFactory.java @@ -21,8 +21,8 @@ public class DataSenderFactory { private static Logger logger = Logger.getLogger(DataSenderFactory.class.getName()); - private static List socketAddresses = new ArrayList(); - private static List unUsedSocketAddresses = new ArrayList(); + private static List socketAddresses = new ArrayList(); + private static List unUsedSocketAddresses = new ArrayList(); private static List availableSenders = new ArrayList(); private static Object lock = new Object(); @@ -32,7 +32,7 @@ public class DataSenderFactory { throw new IllegalArgumentException("Collection service configuration error."); } //过滤重复地址 - Set tmpSocktAddress = new HashSet(); + Set tmpSocktAddress = new HashSet(); for (String serverConfig : Config.Sender.SERVERS_ADDR.split(";")) { String[] server = serverConfig.split(":"); if (server.length != 2) @@ -87,13 +87,13 @@ public class DataSenderFactory { unUsedSocketAddresses.add(socketAddresses.get(index)); } } - unUsedSocketAddresses = new ArrayList(socketAddresses); + unUsedSocketAddresses = new ArrayList(socketAddresses); unUsedSocketAddresses.removeAll(usedSocketAddress); } public void run() { - Iterator unUsedSocketAddressIterator; - SocketAddress tmpSocketAddress; + Iterator unUsedSocketAddressIterator; + InetSocketAddress tmpSocketAddress; while (true) { unUsedSocketAddressIterator = unUsedSocketAddresses.iterator(); while (unUsedSocketAddressIterator.hasNext()) { diff --git a/skywalking-api/src/main/java/com/ai/cloud/skywalking/sender/DataSenderFactoryWithBalance.java b/skywalking-api/src/main/java/com/ai/cloud/skywalking/sender/DataSenderFactoryWithBalance.java new file mode 100644 index 000000000..784a54d28 --- /dev/null +++ b/skywalking-api/src/main/java/com/ai/cloud/skywalking/sender/DataSenderFactoryWithBalance.java @@ -0,0 +1,209 @@ +package com.ai.cloud.skywalking.sender; + +import com.ai.cloud.skywalking.conf.Config; +import com.ai.cloud.skywalking.util.StringUtil; + +import java.io.IOException; +import java.net.InetSocketAddress; +import java.util.ArrayList; +import java.util.HashSet; +import java.util.List; +import java.util.Set; +import java.util.concurrent.ThreadLocalRandom; +import java.util.logging.Level; +import java.util.logging.Logger; + +import static com.ai.cloud.skywalking.conf.Config.Sender.*; + +public class DataSenderFactoryWithBalance { + + private static Logger logger = Logger.getLogger(DataSenderFactoryWithBalance.class.getName()); + // unUsedServerAddress存放没有使用的服务器地址， + private static List unusedServerAddresses = new ArrayList(); + + private static List usingDataSender = new ArrayList(); + private static int maxKeepConnectingSenderSize; + private static Object lock = new Object(); + private static boolean NEED_ADD_SENDER_FLAG = false; + + private static int calculateMaxKeeperConnectingSenderSize(int allAddressSize) { + if (CONNECT_PERCENT <= 0 || CONNECT_PERCENT > 100) { + logger.log(Level.ALL, "CONNECT_PERCENT must between 1 and 100"); + System.exit(-1); + } + return (int) Math.ceil(allAddressSize * ((1.0 * CONNECT_PERCENT / 100) % 100)); + } + + // 初始化服务端的地址数据 + static { + // 获取数据 + if (StringUtil.isEmpty(Config.Sender.SERVERS_ADDR)) { + throw new IllegalArgumentException("Collection service configuration error."); + } + + // 初始化地址 + Set tmpInetSocketAddress = new HashSet(); + for (String serverConfig : Config.Sender.SERVERS_ADDR.split(";")) { + String[] server = serverConfig.split(":"); + if (server.length != 2) + throw new IllegalArgumentException("Collection service configuration error."); + tmpInetSocketAddress.add(new InetSocketAddress(server[0], Integer.valueOf(server[1]))); + } + + unusedServerAddresses.addAll(tmpInetSocketAddress); + + //根据配置的服务器集群的地址，来计算保持连接的Sender的数量 + maxKeepConnectingSenderSize = calculateMaxKeeperConnectingSenderSize(tmpInetSocketAddress.size()); + + + // 初始化的发送程序 + int index = 0; + while (usingDataSender.size() < maxKeepConnectingSenderSize) { + index = ThreadLocalRandom.current().nextInt(0, unusedServerAddresses.size()); + try { + usingDataSender.add(new DataSender(unusedServerAddresses.get(index))); + unusedServerAddresses.remove(index); + } catch (IOException e) { + // 服务器连接不上 + logger.log(Level.SEVERE, "Failed to connect server[" + + unusedServerAddresses.get(index).getHostName() + "]"); + continue; + } + } + + new DataSenderChecker().start(); + } + + // 获取连接 + + public static DataSender getSender() { + DataSender readySender = null; + while (true) { + int index = ThreadLocalRandom.current().nextInt(0, usingDataSender.size()); + if (usingDataSender.get(index).getStatus() == DataSender.SenderStatus.READY) { + readySender = usingDataSender.get(index); + break; + } + + if (readySender == null) { + try { + Thread.sleep(RETRY_GET_SENDER_WAIT_INTERVAL); + } catch (InterruptedException e) { + logger.log(Level.ALL, "Sleep failed"); + } + } + + } + + return readySender; + } + + // 定时Sender状态检查 + public static class DataSenderChecker extends Thread { + public DataSenderChecker() { + super("Data-Sender-Checker"); + } + + @Override + public void run() { + long sleepTime = 0; + while (true) { + // 检查是否需要新增 + // NEED_ADD_SENDER_FLAG 将会在unRegister方法修改值 + if (NEED_ADD_SENDER_FLAG) { + DataSender newSender; + for (int i = 0; i < usingDataSender.size(); i++) { + if (usingDataSender.get(i).getStatus() == DataSender.SenderStatus.FAILED) { + // 正在使用的Sender的数量 <= maxKeepConnectingSenderSize + // 剩余的服务器地址数量 = 总得服务器地址数量 - 正在使用的Sender的数量 + // 可替换的服务器数量 = 剩余服务器地址数量 + // 当剩余服务器地址数量 <= 0 时，可以替换的地址也不存在，替换操作就可以不执行，所以这里的while是这样的意思 + // 当剩余服务器地址数量 > 0 时， 就可以找到可以替换的地址，替换操作也就可以执行了，这里的就会跳出while循环 + while ((newSender = findReadySender()) == null) { + try { + Thread.sleep(RETRY_FIND_CONNECTION_SENDER); + } catch (InterruptedException e) { + logger.log(Level.ALL, "Sleep failed."); + } + } + // 找到可以替换的Sender + usingDataSender.set(i, newSender); + unusedServerAddresses.add(usingDataSender.get(i).getServerIp()); + if (usingDataSender.size() >= maxKeepConnectingSenderSize) { + NEED_ADD_SENDER_FLAG = false; + break; + } + } + } + + + } + + // 检查是否需要替换 + if (sleepTime >= SWITCH_SENDER_INTERVAL) { + DataSender toBeSwitchSender; + DataSender tmpSender; + while (true) { + int toBeSwitchIndex = ThreadLocalRandom.current().nextInt(0, usingDataSender.size() - 1); + toBeSwitchSender = usingDataSender.get(toBeSwitchIndex); + if (toBeSwitchSender.getStatus() == DataSender.SenderStatus.READY) { + tmpSender = findReadySender(); + // 找到可以替换的Sender + if (tmpSender != null) { + usingDataSender.set(toBeSwitchIndex, tmpSender); + try { + Thread.sleep(CLOSE_SENDER_COUNTDOWN); + } catch (InterruptedException e) { + logger.log(Level.ALL, "Sleep Failed"); + } + unusedServerAddresses.remove(tmpSender.getServerIp()); + unusedServerAddresses.add(toBeSwitchSender.getServerIp()); + } + break; + } + } + sleepTime = 0; + } + + // + sleepTime += CHECKER_THREAD_WAIT_INTERVAL; + try { + Thread.sleep(CHECKER_THREAD_WAIT_INTERVAL); + } catch (InterruptedException e) { + logger.log(Level.ALL, "Sleep failed"); + } + + } + } + + } + + private static DataSender findReadySender() { + DataSender result = null; + for (InetSocketAddress serverAddress : unusedServerAddresses) { + try { + result = new DataSender(serverAddress); + break; + } catch (IOException e) { + if (result != null) { + try { + result.closeConnect(); + } catch (IOException ex) { + logger.log(Level.ALL, "Failed to close socket[" + + serverAddress.getHostName() + "]"); + } + } + continue; + } + } + return result; + } + + public static void unRegister(DataSender socket) { + int index = usingDataSender.indexOf(socket); + if (index != -1) { + usingDataSender.get(index).setStatus(DataSender.SenderStatus.FAILED); + } + NEED_ADD_SENDER_FLAG = true; + } +}