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

数据上报的关键组件和流程

This commit is contained in:
kazusa 2025-07-28 22:24:39 +08:00
parent 12e30dfe34
commit 243b8bfc26
11 changed files with 35 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

8
apm-commons/apm-datacarrier/src/main/java/org/apache/skywalking/apm/commons/datacarrier/DataCarrier.java
View File

@ -31,6 +31,10 @@ import org.apache.skywalking.apm.commons.datacarrier.partition.SimpleRollingPart
/**
* DataCarrier main class. use this instance to set Producer/Consumer Model.
*/
/**
* 数据缓冲层
*/
public class DataCarrier<T> {
private Channels<T> channels;
private IDriver driver;
@ -50,8 +54,11 @@ public class DataCarrier<T> {
public DataCarrier(String name, String envPrefix, int channelSize, int bufferSize, BufferStrategy strategy) {
this.name = name;
// 每个分区的内存大小
bufferSize = EnvUtil.getInt(envPrefix + "_BUFFER_SIZE", bufferSize);
// 缓存的数据分区,提高并发度
channelSize = EnvUtil.getInt(envPrefix + "_CHANNEL_SIZE", channelSize);
// 数据分区
channels = new Channels<>(channelSize, bufferSize, new SimpleRollingPartitioner<T>(), strategy);
}
@ -126,6 +133,7 @@ public class DataCarrier<T> {
if (driver != null) {
driver.close(channels);
}
// 创建数据采集线程
driver = new ConsumeDriver<T>(this.name, this.channels, consumer, num, consumeCycle);
driver.begin(channels);
return this;

1
apm-commons/apm-datacarrier/src/main/java/org/apache/skywalking/apm/commons/datacarrier/buffer/Channels.java
View File

@ -33,6 +33,7 @@ public class Channels<T> {
public Channels(int channelSize, int bufferSize, IDataPartitioner<T> partitioner, BufferStrategy strategy) {
this.dataPartitioner = partitioner;
this.strategy = strategy;
// 数据分区
bufferChannels = new QueueBuffer[channelSize];
for (int i = 0; i < channelSize; i++) {
if (BufferStrategy.BLOCKING.equals(strategy)) {

6
apm-commons/apm-datacarrier/src/main/java/org/apache/skywalking/apm/commons/datacarrier/consumer/ConsumerThread.java
View File

@ -50,6 +50,7 @@ public class ConsumerThread<T> extends Thread {
final List<T> consumeList = new ArrayList<T>(1500);
while (running) {
// 消费缓冲数据
if (!consume(consumeList)) {
try {
Thread.sleep(consumeCycle);
@ -67,11 +68,13 @@ public class ConsumerThread<T> extends Thread {
private boolean consume(List<T> consumeList) {
for (DataSource dataSource : dataSources) {
// 将数据读取到consumeList
dataSource.obtain(consumeList);
}
if (!consumeList.isEmpty()) {
try {
// 数据消费
consumer.consume(consumeList);
} catch (Throwable t) {
consumer.onError(consumeList, t);
@ -91,6 +94,9 @@ public class ConsumerThread<T> extends Thread {
/**
* DataSource is a reference to {@link Buffer}.
*/
/**
* 存储数据
*/
class DataSource {
private QueueBuffer<T> sourceBuffer;

1
apm-sniffer/apm-agent-core/src/main/java/org/apache/skywalking/apm/agent/core/boot/ServiceManager.java
View File

@ -62,6 +62,7 @@ public enum ServiceManager {
private Map<Class, BootService> loadAllServices() {
Map<Class, BootService> bootedServices = new LinkedHashMap<>();
List<BootService> allServices = new LinkedList<>();
// 使用spi方式加载BootService
load(allServices);
for (final BootService bootService : allServices) {
Class<? extends BootService> bootServiceClass = bootService.getClass();

1
apm-sniffer/apm-agent-core/src/main/java/org/apache/skywalking/apm/agent/core/context/TracingContext.java
View File

@ -467,6 +467,7 @@ public class TracingContext implements AbstractTracerContext {
}
AgentSo11y.measureTracingContextCompletion(false);
TraceSegment finishedSegment = segment.finish(limitMechanismWorking);
// 通知TraceSegmentServiceClient上报数据
TracingContext.ListenerManager.notifyFinish(finishedSegment);
running = false;
}

3
apm-sniffer/apm-agent-core/src/main/java/org/apache/skywalking/apm/agent/core/context/trace/AbstractTracingSpan.java
View File

@ -331,6 +331,9 @@ public abstract class AbstractTracingSpan implements AbstractSpan {
return this;
}
/**
* 异步存储span数据
*/
@Override
public AbstractSpan asyncFinish() {
if (!isInAsyncMode) {

4
apm-sniffer/apm-agent-core/src/main/java/org/apache/skywalking/apm/agent/core/jvm/JVMService.java
View File

@ -42,6 +42,10 @@ import org.apache.skywalking.apm.util.RunnableWithExceptionProtection;
* The <code>JVMService</code> represents a timer, which collectors JVM cpu, memory, memorypool, gc, thread and class info,
* and send the collected info to Collector through the channel provided by {@link GRPCChannelManager}
*/
/**
* 开始 JVM 指标收集
*/
@DefaultImplementor
public class JVMService implements BootService, Runnable {
private static final ILog LOGGER = LogManager.getLogger(JVMService.class);

4
apm-sniffer/apm-agent-core/src/main/java/org/apache/skywalking/apm/agent/core/remote/GRPCChannelManager.java
View File

@ -45,6 +45,9 @@ import org.apache.skywalking.apm.util.StringUtil;
import static org.apache.skywalking.apm.agent.core.conf.Config.Collector.IS_RESOLVE_DNS_PERIODICALLY;
/**
* 建立OAP的gRPC连接
*/
@DefaultImplementor
public class GRPCChannelManager implements BootService, Runnable {
private static final ILog LOGGER = LogManager.getLogger(GRPCChannelManager.class);
@ -74,6 +77,7 @@ public class GRPCChannelManager implements BootService, Runnable {
connectCheckFuture = Executors.newSingleThreadScheduledExecutor(
new DefaultNamedThreadFactory("GRPCChannelManager")
).scheduleAtFixedRate(
// 心跳检查,skywalking通过随机选择一个oap节点并保持长连接的方式来代替负载均衡,如果连接不通重新选择一个节点
new RunnableWithExceptionProtection(
this,
t -> LOGGER.error("unexpected exception.", t)

3
apm-sniffer/apm-agent-core/src/main/java/org/apache/skywalking/apm/agent/core/remote/ServiceManagementClient.java
View File

@ -44,6 +44,9 @@ import org.apache.skywalking.apm.util.RunnableWithExceptionProtection;
import static org.apache.skywalking.apm.agent.core.conf.Config.Collector.GRPC_UPSTREAM_TIMEOUT;
/**
* 注册服务实例
*/
@DefaultImplementor
public class ServiceManagementClient implements BootService, Runnable, GRPCChannelListener {
private static final ILog LOGGER = LogManager.getLogger(ServiceManagementClient.class);

4
apm-sniffer/apm-agent-core/src/main/java/org/apache/skywalking/apm/agent/core/remote/TraceSegmentServiceClient.java
View File

@ -44,6 +44,9 @@ import static org.apache.skywalking.apm.agent.core.conf.Config.Buffer.BUFFER_SIZ
import static org.apache.skywalking.apm.agent.core.conf.Config.Buffer.CHANNEL_SIZE;
import static org.apache.skywalking.apm.agent.core.remote.GRPCChannelStatus.CONNECTED;
/**
* 启动链路数据上报
*/
@DefaultImplementor
public class TraceSegmentServiceClient implements BootService, IConsumer<TraceSegment>, TracingContextListener, GRPCChannelListener {
private static final ILog LOGGER = LogManager.getLogger(TraceSegmentServiceClient.class);
@ -179,6 +182,7 @@ public class TraceSegmentServiceClient implements BootService, IConsumer<TraceSe
@Override
public void statusChanged(GRPCChannelStatus status) {
// 当grpc连接成功后,创建上报数据的grpc客户端
if (CONNECTED.equals(status)) {
Channel channel = ServiceManager.INSTANCE.findService(GRPCChannelManager.class).getChannel();
serviceStub = TraceSegmentReportServiceGrpc.newStub(channel);

BIN
skywalking源码.xmind Normal file
View File

Binary file not shown.