kazusa
/
skywalking-java
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

1. 完成Sender的负载均衡， 

实现原理：为了完成Sender的负载均衡，采用了定时切换发送的Sender的服务器，每个一个时间间隔就会将DataSender池中的某一个
DataSender替换成没有连接的服务器。
This commit is contained in:
zhangxin10 2015-12-07 18:13:02 +08:00
parent 1af6e1a865
commit eadb979ec0
5 changed files with 293 additions and 54 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
skywalking-api/src/main/java/com/ai/cloud/skywalking/buffer/BufferGroup.java
View File

@ -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) {

91
skywalking-api/src/main/java/com/ai/cloud/skywalking/conf/Config.java
View File

@ -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;
}
}

32
skywalking-api/src/main/java/com/ai/cloud/skywalking/sender/DataSender.java
View File

@ -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;
}
}

12
skywalking-api/src/main/java/com/ai/cloud/skywalking/sender/DataSenderFactory.java
View File

@ -21,8 +21,8 @@ public class DataSenderFactory {
private static Logger logger = Logger.getLogger(DataSenderFactory.class.getName());
private static List<SocketAddress> socketAddresses = new ArrayList<SocketAddress>();
private static List<SocketAddress> unUsedSocketAddresses = new ArrayList<SocketAddress>();
private static List<InetSocketAddress> socketAddresses = new ArrayList<InetSocketAddress>();
private static List<InetSocketAddress> unUsedSocketAddresses = new ArrayList<InetSocketAddress>();
private static List<DataSender> availableSenders = new ArrayList<DataSender>();
private static Object lock = new Object();
@ -32,7 +32,7 @@ public class DataSenderFactory {
throw new IllegalArgumentException("Collection service configuration error.");
}
//过滤重复地址
Set<SocketAddress> tmpSocktAddress = new HashSet<SocketAddress>();
Set<InetSocketAddress> tmpSocktAddress = new HashSet<InetSocketAddress>();
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<SocketAddress>(socketAddresses);
unUsedSocketAddresses = new ArrayList<InetSocketAddress>(socketAddresses);
unUsedSocketAddresses.removeAll(usedSocketAddress);
}
public void run() {
Iterator<SocketAddress> unUsedSocketAddressIterator;
SocketAddress tmpSocketAddress;
Iterator<InetSocketAddress> unUsedSocketAddressIterator;
InetSocketAddress tmpSocketAddress;
while (true) {
unUsedSocketAddressIterator = unUsedSocketAddresses.iterator();
while (unUsedSocketAddressIterator.hasNext()) {

209
skywalking-api/src/main/java/com/ai/cloud/skywalking/sender/DataSenderFactoryWithBalance.java Normal file
View File

@ -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<InetSocketAddress> unusedServerAddresses = new ArrayList<InetSocketAddress>();
private static List<DataSender> usingDataSender = new ArrayList<DataSender>();
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<InetSocketAddress> tmpInetSocketAddress = new HashSet<InetSocketAddress>();
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;
}
}