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

1. 修改发送实现 

2. 修改部分配置
This commit is contained in:
zhangxin10 2015-11-13 14:56:41 +08:00
parent 0ac1e8f077
commit 4fd1b34624
14 changed files with 70 additions and 78 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -1,6 +1,7 @@
package com.ai.cloud.skywalking.buffer;
import com.ai.cloud.skywalking.conf.Config;
import com.ai.cloud.skywalking.context.Span;
import com.ai.cloud.skywalking.sender.DataSenderFactory;
@ -69,8 +70,14 @@ public class BufferGroup {
bool = true;
data.append(dataBuffer[i]);
dataBuffer[i] = null;
if (index++ == MAX_BUFFER_DATA_SIZE) {
DataSenderFactory.getSender().send(data.toString());
if (index++ == MAX_BUFFER_DATA_SIZE || data.length() >= Config.Sender.MAX_SEND_LENGTH) {
while (!DataSenderFactory.getSender().send(data.toString())) {
try {
Thread.sleep(50L);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
index = 0;
data = new StringBuilder();
}

21
skywalking-api/src/main/java/com/ai/cloud/skywalking/buffer/config/BufferConfig.java
View File

@ -1,21 +0,0 @@
package com.ai.cloud.skywalking.buffer.config;
public class BufferConfig {
// 最大消费线程
public static int MAX_WORKER = 5;
// 桶大小
public static int GROUP_MAX_SIZE = 1000;
// 桶数量
public static int POOL_MAX_SIZE = 30;
// 发送的最大长度
public static int MAX_LENGTH = 512 * 1024;
// 发送的最大条数
public static int SEND_MAX_SIZE = 1;
// 最大发送者的连接数阀值
public static int SEND_CONNECTION_THRESHOLD = 2;
}

4
skywalking-api/src/main/java/com/ai/cloud/skywalking/buriedpoint/LocalBuriedPointSender.java
View File

@ -15,14 +15,14 @@ public class LocalBuriedPointSender implements IBuriedPointSender {
public ContextData beforeSend(Identification id) {
Span spanData = ContextGenerator.generateContextFromThreadLocal(id);
// 3.将新创建的Context存放到ThreadLocal栈中。
Context.getOrCreate().append(spanData);
Context.append(spanData);
// 4 并将当前的Context返回回去
return new ContextData(spanData);
}
public void afterSend() {
// 弹出上下文的栈顶中的元素
Span spanData = Context.getOrCreate().removeLastSpan();
Span spanData = Context.removeLastSpan();
if (spanData == null) {
return;
}

4
skywalking-api/src/main/java/com/ai/cloud/skywalking/buriedpoint/RPCBuriedPointReceiver.java
View File

@ -13,7 +13,7 @@ public class RPCBuriedPointReceiver implements IBuriedPointReceiver {
public void afterReceived() {
// 获取上下文的栈顶中的元素
Span spanData = Context.getOrCreate().removeLastSpan();
Span spanData = Context.removeLastSpan();
// 填上必要信息
spanData.setCost(System.currentTimeMillis() - spanData.getStartDate());
// 存放到本地发送进程中
@ -27,7 +27,7 @@ public class RPCBuriedPointReceiver implements IBuriedPointReceiver {
System.out.println("viewpointId:" + spanData.getViewPointId() + "\tParentLevelId:" + spanData.
getParentLevel() + "\tLevelId:" + spanData.getLevelId());
}
Context.getOrCreate().append(spanData);
Context.append(spanData);
}
public void handleException(Throwable e) {

4
skywalking-api/src/main/java/com/ai/cloud/skywalking/buriedpoint/RPCBuriedPointSender.java
View File

@ -15,14 +15,14 @@ public class RPCBuriedPointSender implements IBuriedPointSender {
public ContextData beforeSend(Identification id) {
Span spanData = ContextGenerator.generateContextFromThreadLocal(id);
// 3.将新创建的Context存放到ThreadLocal栈中。
Context.getOrCreate().append(spanData);
Context.append(spanData);
// 4 并将当前的Context返回回去
return new ContextData(new Span(spanData.getTraceId(), spanData.getParentLevel()));
}
public void afterSend() {
// 获取上下文的栈顶中的元素
Span spanData = Context.getOrCreate().removeLastSpan();
Span spanData = Context.removeLastSpan();
if (spanData == null) {
return;
}

6
skywalking-api/src/main/java/com/ai/cloud/skywalking/buriedpoint/ThreadBuriedPointSender.java
View File

@ -18,7 +18,7 @@ public class ThreadBuriedPointSender implements IBuriedPointSender {
public ThreadBuriedPointSender(int threadSeqId) {
Span spanData;
// 从ThreadLocal中取出上下文
final Span parentSpanData = Context.getOrCreate().getLastSpan();
final Span parentSpanData = Context.getLastSpan();
if (parentSpanData == null) {
spanData = new Span(TraceIdGenerator.generate());
} else {
@ -37,12 +37,12 @@ public class ThreadBuriedPointSender implements IBuriedPointSender {
span.setStartDate(System.currentTimeMillis());
span.setViewPointId(id.getViewPoint());
span.setProcessNo(BuriedPointMachineUtil.getProcessNo());
Context.getOrCreate().append(span);
Context.append(span);
return new ContextData(span);
}
public void afterSend() {
Span span = Context.getOrCreate().removeLastSpan();
Span span = Context.removeLastSpan();
if (span == null) {
return;
}

4
skywalking-api/src/main/java/com/ai/cloud/skywalking/buriedpoint/ThreadFactoryBuriedPointSender.java
View File

@ -15,14 +15,14 @@ public class ThreadFactoryBuriedPointSender implements IBuriedPointSender {
public ContextData beforeSend(Identification id) {
Span spanData = ContextGenerator.generateContextFromThreadLocal(id);
// 3.将新创建的Context存放到ThreadLocal栈中。
Context.getOrCreate().append(spanData);
Context.append(spanData);
// 4 并将当前的Context返回回去
return new ContextData(spanData);
}
public void afterSend() {
// 获取上下文的栈顶中的元素
Span spanData = Context.getOrCreate().removeLastSpan();
Span spanData = Context.removeLastSpan();
if (spanData == null) {
return;
}

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

@ -2,7 +2,7 @@ package com.ai.cloud.skywalking.conf;
public class Config {
public static class BuriedPoint{
public static class BuriedPoint {
public static boolean PRINTF = false;
}
@ -26,12 +26,15 @@ public class Config {
public static int MAX_BUFFER_DATA_SIZE = 1;
// 最大发送者的连接数阀比例
public static int SEND_CONNECTION_THRESHOLD = 2;
public static int CONNECT_PERCENT = 50;
// 发送服务端配置
public static String SENDER_SERVERS;
public static String SERVERS_ADDR;
// 是否开启发送
public static boolean IS_OFF = false;
// 发送的最大长度
public static int MAX_SEND_LENGTH = 1800;
}
}

22
skywalking-api/src/main/java/com/ai/cloud/skywalking/context/Context.java
View File

@ -4,33 +4,31 @@ import java.util.Stack;
public class Context {
private static ThreadLocal<SpanNodeStack> nodes = new ThreadLocal<SpanNodeStack>();
private static Context context;
private Context() {
nodes.set(new SpanNodeStack());
}
public void append(Span span) {
public static void append(Span span) {
if (nodes.get() == null) {
nodes.set(new SpanNodeStack());
}
nodes.get().push(span);
}
public Span getLastSpan() {
public static Span getLastSpan() {
if (nodes.get() == null) {
return null;
}
return nodes.get().peek();
}
public Span removeLastSpan() {
public static Span removeLastSpan() {
if (nodes.get() == null)
return null;
return nodes.get().pop();
}
public static Context getOrCreate() {
if (context == null) {
context = new Context();
}
return context;
}
static class SpanNodeStack {
private Stack<SpanNode> spans = new Stack<SpanNode>();

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

@ -52,7 +52,6 @@ public class DataSender {
socketChannel.write(ByteBuffer.wrap(ProtocolBuilder.builder(data)));
return true;
} catch (IOException e) {
e.printStackTrace();
// 发送失败 认为不可连接
DataSenderFactory.unRegister(this);
return false;

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

@ -6,26 +6,26 @@ import com.ai.cloud.skywalking.util.StringUtil;
import java.io.IOException;
import java.net.InetSocketAddress;
import java.net.SocketAddress;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
import java.util.*;
import java.util.concurrent.ThreadLocalRandom;
import static com.ai.cloud.skywalking.conf.Config.Sender.SEND_CONNECTION_THRESHOLD;
import static com.ai.cloud.skywalking.conf.Config.Sender.CONNECT_PERCENT;
public class DataSenderFactory {
private static List<SocketAddress> socketAddresses = new ArrayList<SocketAddress>();
private static List<SocketAddress> unUsedSocketAddresses = new ArrayList<SocketAddress>();
private static Set<SocketAddress> socketAddresses = new HashSet<SocketAddress>();
private static Set<SocketAddress> unUsedSocketAddresses = new HashSet<SocketAddress>();
private static List<DataSender> availableSenders = new ArrayList<DataSender>();
private static DataSenderMaker dataSenderMaker;
private static Object lock = new Object();
static {
try {
if (StringUtil.isEmpty(Config.Sender.SENDER_SERVERS)) {
if (StringUtil.isEmpty(Config.Sender.SERVERS_ADDR)) {
throw new IllegalArgumentException("Collection service configuration error.");
}
for (String serverConfig : Config.Sender.SENDER_SERVERS.split(";")) {
for (String serverConfig : Config.Sender.SERVERS_ADDR.split(";")) {
String[] server = serverConfig.split(":");
if (server.length != 2)
throw new IllegalArgumentException("Collection service configuration error.");
@ -36,11 +36,12 @@ public class DataSenderFactory {
System.exit(-1);
}
new DataSenderMaker().start();
dataSenderMaker = new DataSenderMaker();
dataSenderMaker.start();
}
public static DataSender getSender() {
while(availableSenders.size() <= 0){
while (availableSenders.size() == 0) {
try {
Thread.sleep(2000L);
} catch (InterruptedException e) {
@ -52,9 +53,14 @@ public class DataSenderFactory {
static class DataSenderMaker extends Thread {
private int avaiableSize = (int) Math.ceil(socketAddresses.size() * 1.0 / SEND_CONNECTION_THRESHOLD);
private int avaiableSize;
public DataSenderMaker() {
if (CONNECT_PERCENT <= 0 || CONNECT_PERCENT > 100) {
System.err.println("CONNECT_PERCENT must between 1 and 100");
System.exit(-1);
}
avaiableSize = (int) Math.ceil(socketAddresses.size() * 1.0 * ((CONNECT_PERCENT / 100) % 100));
// 初始化DataSender
Iterator<SocketAddress> it = socketAddresses.iterator();
List<SocketAddress> usedSocketAddress = new ArrayList<SocketAddress>();
@ -71,7 +77,7 @@ public class DataSenderFactory {
}
}
unUsedSocketAddresses = new ArrayList<SocketAddress>(socketAddresses);
unUsedSocketAddresses = new HashSet<SocketAddress>(socketAddresses);
unUsedSocketAddresses.removeAll(usedSocketAddress);
}
@ -87,20 +93,21 @@ public class DataSenderFactory {
// 当前发送的地址还是不可用
}
}
try {
Thread.sleep(3000L);
Thread.sleep(200L);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
}
public static void unRegister(DataSender sender) {
availableSenders.remove(sender);
unUsedSocketAddresses.add(sender.getServerIp());
synchronized (lock) {
availableSenders.remove(sender);
unUsedSocketAddresses.add(sender.getServerIp());
}
}
}

15
skywalking-api/src/main/java/com/ai/cloud/skywalking/sender/protocol/ProtocolBuilder.java
View File

@ -14,13 +14,12 @@ public class ProtocolBuilder {
return des;
}
private static byte[] intToByteArray(final int integer) {
int byteNum = (40 - Integer.numberOfLeadingZeros(integer < 0 ? ~integer : integer)) / 8;
byte[] byteArray = new byte[4];
for (int n = 0; n < byteNum; n++)
byteArray[3 - n] = (byte) (integer >>> (n * 8));
return (byteArray);
private static byte[] intToByteArray(final int value) {
byte[] src = new byte[4];
src[0] = (byte) ((value >> 24) & 0xFF);
src[1] = (byte) ((value >> 16) & 0xFF);
src[2] = (byte) ((value >> 8) & 0xFF);
src[3] = (byte) (value & 0xFF);
return src;
}
}

2
skywalking-api/src/main/java/com/ai/cloud/skywalking/util/ContextGenerator.java
View File

@ -51,7 +51,7 @@ public final class ContextGenerator {
private static Span getContextFromThreadLocal() {
Span span;
// 1.获取Context，从ThreadLocal栈中获取中
final Span parentSpan = Context.getOrCreate().getLastSpan();
final Span parentSpan = Context.getLastSpan();
// 2 校验Context，Context是否存在
if (parentSpan == null) {
// 不存在，新创建一个Context

4
skywalking-api/src/main/java/com/ai/cloud/skywalking/util/ExceptionHandleUtil.java
View File

@ -22,11 +22,11 @@ public final class ExceptionHandleUtil {
expMessage.append("\n本地发送埋点关闭异常读入流异常，异常信息:");
expMessage.append(e1.getCause().getMessage());
}
return expMessage.toString();
return expMessage.toString().replace('\n','&');
}
public static void handleException(Throwable e) {
Span spanData = Context.getOrCreate().getLastSpan();
Span spanData = Context.getLastSpan();
// 设置错误信息
byte errorCode = 1;
spanData.setStatueCode(errorCode);