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 ea5b3f4cd2b6c36a77ca70c6073e43197d30dfa2..7b142043362be77c1f23e6b3b64e4279adb94551 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 Set socketAddresses = new HashSet(); - private static Set unUsedSocketAddresses = new HashSet(); + private static List socketAddresses = new ArrayList(); + private static List unUsedSocketAddresses = new ArrayList(); private static List availableSenders = new ArrayList(); private static Object lock = new Object(); @@ -31,13 +31,17 @@ public class DataSenderFactory { if (StringUtil.isEmpty(Config.Sender.SERVERS_ADDR)) { throw new IllegalArgumentException("Collection service configuration error."); } - + //过滤重复地址 + Set tmpSocktAddress = 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."); - socketAddresses.add(new InetSocketAddress(server[0], Integer.valueOf(server[1]))); + tmpSocktAddress.add(new InetSocketAddress(server[0], Integer.valueOf(server[1]))); } + + socketAddresses.addAll(tmpSocktAddress); + } catch (Exception e) { logger.log(Level.ALL, "Collection service configuration error.", e); System.exit(-1); @@ -72,18 +76,18 @@ public class DataSenderFactory { // 初始化DataSender List usedSocketAddress = new ArrayList(); - for (SocketAddress socketAddress : socketAddresses) { - if (availableSenders.size() >= availableSize) { - break; - } + int index; + while (availableSenders.size() < availableSize) { + // 随机获取服务器地址 + index = ThreadLocalRandom.current().nextInt(socketAddresses.size()); try { - availableSenders.add(new DataSender(socketAddress)); - usedSocketAddress.add(socketAddress); + availableSenders.add(new DataSender(socketAddresses.get(index))); + usedSocketAddress.add(socketAddresses.get(index)); } catch (IOException e) { - unUsedSocketAddresses.add(socketAddress); + unUsedSocketAddresses.add(socketAddresses.get(index)); } } - unUsedSocketAddresses = new HashSet(socketAddresses); + unUsedSocketAddresses = new ArrayList(socketAddresses); unUsedSocketAddresses.removeAll(usedSocketAddress); }