SkyWalking 源码分析 —— Collector Server Component 服务器组件

1. 引言

1.1 SkyWalking 概述

SkyWalking 是一个开源的 APM（应用性能监控）系统，专门为微服务、云原生和容器化架构设计。它提供了分布式追踪、服务网格遥测分析、度量聚合和可视化一体化解决方案。Collector 作为 SkyWalking 的核心服务器组件，承担着数据处理、聚合和存储的重要职责。

1.2 Collector 组件的重要性

Collector 是 SkyWalking 架构中的中枢神经系统，负责接收来自各种探针（Agent）的数据，进行实时处理和分析，并将结果存储到后端存储中。其设计直接影响整个系统的性能、可靠性和扩展性。

2. Collector 整体架构设计

2.1 模块化架构

Collector 采用高度模块化的设计，每个模块都有明确的职责边界：

java

// 模块定义接口
public interface Module {
String name();
Class<? extends ModuleProvider> provider();
}

// 模块提供者接口
public interface ModuleProvider {
String name();
ModuleConfig createConfigBeanIfAbsent();
void prepare() throws ServiceNotProvidedException;
void start() throws ServiceNotProvidedException;
void notifyAfterCompleted() throws ServiceNotProvidedException;
String[] requiredModules();
}

2.2 核心模块组成

Collector 由以下几个核心模块构成：

2.3 数据流架构

Collector 的数据处理遵循清晰的流水线模式：

text

Agent数据 → 接收模块 → 数据解析 → 数据聚合 → 存储 → 查询服务

3. 启动流程深度分析

3.1 入口类分析

Collector 的启动入口位于 ApplicationStartUp 类：

java

public class ApplicationStartUp {
public static void main(String[] args) {
// 1. 初始化配置管理器
ConfigInitializer.initialize();

// 2. 创建并配置模块管理器
ModuleManager moduleManager = new ModuleManager();

// 3. 初始化核心模块
BootstrapFlow bootstrapFlow = new BootstrapFlow(moduleManager);
bootstrapFlow.start();

// 4. 启动完成后的回调
bootstrapFlow.notifyAfterCompleted();
}
}

3.2 模块初始化过程

模块初始化采用分层启动策略：

java

public class BootstrapFlow {
private void start() {
// 第一阶段：准备阶段
for (ModuleProvider provider : providers) {
provider.prepare();
}

// 第二阶段：启动阶段
for (ModuleProvider provider : providers) {
provider.start();
}

// 第三阶段：完成阶段
for (ModuleProvider provider : providers) {
provider.notifyAfterCompleted();
}
}
}

3.3 依赖注入机制

SkyWalking 使用自定义的轻量级依赖注入机制：

java

public class ModuleManager {
private Map<String, ModuleProvider> loadedProviders = new HashMap<>();

public void init(ApplicationConfiguration applicationConfiguration) {
// 解析模块依赖关系
List<String> startupModuleList = applicationConfiguration.moduleList();

// 拓扑排序确保依赖顺序
List<String> sortedModules = sortModules(startupModuleList);

// 按顺序初始化模块
for (String moduleName : sortedModules) {
ModuleProvider provider = createProvider(moduleName);
loadedProviders.put(moduleName, provider);
provider.prepare();
}
}
}

4. 网络通信层详解

4.1 gRPC 服务器实现

Collector 使用 gRPC 作为主要的通信协议：

java

public class GRPCServer {
private Server server;
private final int port;

public void start() throws IOException {
server = ServerBuilder.forPort(port)
.addService(new TraceSegmentService())
.addService(new JVMMetricService())
.addService(new ManagementService())
.build()
.start();

// 注册关闭钩子
Runtime.getRuntime().addShutdownHook(new Thread(this::stop));
}

// 跟踪段服务实现
private class TraceSegmentService extends TraceSegmentReportServiceGrpc.TraceSegmentReportServiceImplBase {
@Override
public void collect(UpstreamSegment request,
StreamObserver<Commands> responseObserver) {
// 处理跟踪数据
segmentProcessor.process(request);
responseObserver.onNext(Commands.newBuilder().build());
responseObserver.onCompleted();
}
}
}

4.2 HTTP 服务器设计

HTTP 服务器主要用于管理接口和查询接口：

java

public class JettyServer {
private Server server;

public void start() {
server = new Server(port);

ServletContextHandler context = new ServletContextHandler();
context.setContextPath("/");

// 注册 GraphQL 查询服务
context.addServlet(new ServletHolder(new GraphQLQueryHandler()), "/graphql");

// 注册健康检查端点
context.addServlet(new ServletHolder(new HealthCheckHandler()), "/health");

server.setHandler(context);
server.start();
}
}

4.3 连接管理和负载均衡

java

public class ConnectionManager {
private final Map<String, List<Connection>> connections = new ConcurrentHashMap<>();

public void register(String serviceName, Connection connection) {
connections.computeIfAbsent(serviceName, k -> new CopyOnWriteArrayList<>())
.add(connection);
}

public Connection select(String serviceName) {
List<Connection> connList = connections.get(serviceName);
if (connList == null || connList.isEmpty()) {
return null;
}
// 使用轮询负载均衡策略
return roundRobinSelect(connList);
}
}

5. 数据处理引擎

5.1 数据接收器（Receiver）

Receiver 负责接收不同类型的数据：

java

public abstract class Receiver {
private final DataProcessor processor;

public void receive(RemoteData data) {
// 数据验证
if (!validate(data)) {
return;
}

// 数据预处理
PreprocessedData preprocessed = preprocess(data);

// 提交到处理队列
processor.process(preprocessed);
}

protected abstract boolean validate(RemoteData data);
protected abstract PreprocessedData preprocess(RemoteData data);
}

5.2 数据处理器（Processor）

Processor 实现具体的数据处理逻辑：

java

public class SegmentProcessor implements DataProcessor<SegmentObject> {
private final Buffer buffer;
private final AnalysisWorker analysisWorker;

@Override
public void process(SegmentObject segment) {
// 1. 数据缓冲
buffer.write(segment);

// 2. 触发分析
if (buffer.isReadyForAnalysis()) {
analysisWorker.analyze(buffer.drain());
}
}
}

5.3 数据聚合器（Aggregator）

Aggregator 负责数据的聚合计算：

java

public class MetricsAggregator {
private final TimeWindow window;
private final ConcurrentHashMap<MetricKey, MetricValue> metrics = new ConcurrentHashMap<>();

public void aggregate(MetricData data) {
MetricKey key = buildKey(data);
metrics.compute(key, (k, existing) -> {
if (existing == null) {
return createNewValue(data);
}
return mergeValue(existing, data);
});

// 检查时间窗口
if (window.shouldFlush()) {
flushMetrics();
}
}

private void flushMetrics() {
Map<MetricKey, MetricValue> snapshot = new HashMap<>(metrics);
metrics.clear();

// 异步存储
storageService.save(snapshot);
}
}

6. 存储抽象层

6.1 存储接口设计

java

public interface StorageDAO {
// 度量数据存储
void saveMetrics(List<Metric> metrics);

// 跟踪数据存储
void saveTraces(List<Trace> traces);

// 查询接口
List<Trace> queryTraces(TraceQuery query);
List<Metric> queryMetrics(MetricQuery query);
}

6.2 Elasticsearch 实现

java

public class ElasticSearchDAO implements StorageDAO {
private final RestHighLevelClient client;
private final IndexManager indexManager;

@Override
public void saveMetrics(List<Metric> metrics) {
BulkRequest bulkRequest = new BulkRequest();

for (Metric metric : metrics) {
IndexRequest request = new IndexRequest(indexManager.getMetricIndex(metric))
.source(convertToMap(metric));
bulkRequest.add(request);
}

// 批量写入
client.bulk(bulkRequest, RequestOptions.DEFAULT);
}

@Override
public List<Trace> queryTraces(TraceQuery query) {
SearchRequest searchRequest = new SearchRequest(indexManager.getTraceIndex());

// 构建查询条件
BoolQueryBuilder boolQuery = QueryBuilders.boolQuery();
boolQuery.must(QueryBuilders.termQuery("serviceId", query.getServiceId()));

if (query.getStartTime() != null && query.getEndTime() != null) {
boolQuery.must(QueryBuilders.rangeQuery("startTime")
.gte(query.getStartTime())
.lte(query.getEndTime()));
}

searchRequest.source(new SearchSourceBuilder().query(boolQuery));

SearchResponse response = client.search(searchRequest, RequestOptions.DEFAULT);
return convertToTraces(response);
}
}

6.3 存储优化策略

java

public class StorageOptimizer {
// 批量写入优化
public void optimizedBulkWrite(List<DataPoint> dataPoints) {
if (dataPoints.size() < BATCH_SIZE) {
// 等待更多数据
buffer.addAll(dataPoints);
return;
}

// 分批写入
List<List<DataPoint>> batches = partition(dataPoints, BATCH_SIZE);
for (List<DataPoint> batch : batches) {
asyncWrite(batch);
}
}

// 索引管理优化
public void manageIndices() {
// 自动创建新索引
createNewIndexIfNeeded();

// 清理过期索引
deleteExpiredIndices();

// 索引优化
optimizeIndices();
}
}

7. 集群管理机制

7.1 集群协调器

java

public class ClusterCoordinator {
private final ServiceDiscovery discovery;
private final ServiceRegistry registry;
private final LoadBalancer loadBalancer;

public void init() {
// 注册当前节点
registry.register(buildNodeInfo());

// 发现其他节点
discovery.watchNodes(this::onNodesChanged);
}

private void onNodesChanged(List<Node> nodes) {
// 更新负载均衡器
loadBalancer.updateNodes(nodes);

// 重新分配任务
taskScheduler.redistributeTasks();
}
}

7.2 分布式锁实现

java

public class DistributedLock {
private final String lockKey;
private final String lockValue;
private final long ttl;

public boolean tryLock() {
// 使用 Redis 或 ZooKeeper 实现分布式锁
return redisTemplate.opsForValue()
.setIfAbsent(lockKey, lockValue, Duration.ofSeconds(ttl));
}

public void unlock() {
// 只能解锁自己持有的锁
if (lockValue.equals(redisTemplate.opsForValue().get(lockKey))) {
redisTemplate.delete(lockKey);
}
}
}

7.3 数据分片策略

java

public class ShardingStrategy {
private final int totalShards;

public int calculateShard(String entityId) {
// 基于一致性哈希的分片策略
int hashCode = Objects.hashCode(entityId);
return Math.abs(hashCode) % totalShards;
}

public boolean isResponsible(String entityId) {
int shard = calculateShard(entityId);
return shard == currentShard;
}
}

8. 配置管理系统

8.1 配置加载机制

java

public class ConfigurationManager {
private final Map<String, Object> configurations = new ConcurrentHashMap<>();

public void loadConfigurations() {
// 1. 加载默认配置
loadFromClasspath("application.yml");

// 2. 加载外部配置
loadFromFileSystem("config/application.yml");

// 3. 加载环境变量
loadFromEnvironment();

// 4. 加载运行时配置
loadFromRuntime();
}

public <T> T getConfig(String key, Class<T> type) {
Object value = configurations.get(key);
return type.cast(value);
}
}

8.2 热更新配置

java

public class HotUpdateConfiguration {
private final WatchService watchService;

public void watchConfigChanges() {
Path configPath = Paths.get("config");
configPath.register(watchService, ENTRY_MODIFY);

while (true) {
WatchKey key = watchService.take();
for (WatchEvent<?> event : key.pollEvents()) {
if (event.context().toString().endsWith(".yml")) {
// 重新加载配置
reloadConfiguration();
// 通知配置变更
notifyConfigurationChange();
}
}
key.reset();
}
}
}

9. 监控和度量

9.1 性能度量收集

java

public class PerformanceMetrics {
private final Meter requestMeter;
private final Timer responseTimer;
private final Counter errorCounter;

public void recordRequest() {
requestMeter.mark();
}

public void recordResponseTime(long duration) {
responseTimer.record(duration, TimeUnit.MILLISECONDS);
}

public void recordError() {
errorCounter.increment();
}

public PerformanceSnapshot getSnapshot() {
return new PerformanceSnapshot(
requestMeter.getCount(),
responseTimer.getMeanRate(),
errorCounter.getCount()
);
}
}

9.2 健康检查机制

java

public class HealthCheckManager {
private final List<HealthCheck> healthChecks = new CopyOnWriteArrayList<>();

public HealthStatus checkHealth() {
List<HealthIssue> issues = new ArrayList<>();

for (HealthCheck check : healthChecks) {
HealthResult result = check.check();
if (!result.isHealthy()) {
issues.add(result.getIssue());
}
}

return new HealthStatus(issues.isEmpty(), issues);
}

// 存储健康检查
public class StorageHealthCheck implements HealthCheck {
@Override
public HealthResult check() {
try {
storageDAO.healthCheck();
return HealthResult.healthy();
} catch (Exception e) {
return HealthResult.unhealthy("Storage unavailable: " + e.getMessage());
}
}
}
}

10. 容错和高可用

10.1 故障转移机制

java

public class FailoverManager {
private final List<ServiceEndpoint> endpoints;
private int currentEndpointIndex = 0;

public <T> T executeWithFailover(Callable<T> operation) {
int retries = 0;
while (retries < maxRetries) {
try {
ServiceEndpoint endpoint = getNextEndpoint();
return operation.call(endpoint);
} catch (Exception e) {
retries++;
markEndpointAsDown(currentEndpointIndex);
if (retries == maxRetries) {
throw new RuntimeException("All endpoints failed", e);
}
}
}
return null;
}

private synchronized ServiceEndpoint getNextEndpoint() {
currentEndpointIndex = (currentEndpointIndex + 1) % endpoints.size();
return endpoints.get(currentEndpointIndex);
}
}

10.2 数据备份和恢复

java

public class DataBackupManager {
private final StorageDAO primaryStorage;
private final StorageDAO backupStorage;

public void backupData(BackupData data) {
// 异步备份到次要存储
CompletableFuture.runAsync(() -> {
try {
backupStorage.save(data);
} catch (Exception e) {
logger.error("Backup failed", e);
}
});
}

public void restoreFromBackup() {
// 从备份恢复数据
List<BackupData> backupData = backupStorage.getAll();
primaryStorage.batchSave(backupData);
}
}

11. 性能优化策略

11.1 内存管理优化

java

public class MemoryPool {
private final Queue<byte[]> bufferPool = new ConcurrentLinkedQueue<>();
private final int bufferSize;

public byte[] borrowBuffer() {
byte[] buffer = bufferPool.poll();
if (buffer == null) {
buffer = new byte[bufferSize];
}
return buffer;
}

public void returnBuffer(byte[] buffer) {
if (buffer != null && buffer.length == bufferSize) {
// 清空缓冲区以便重用
Arrays.fill(buffer, (byte) 0);
bufferPool.offer(buffer);
}
}
}

11.2 缓存策略

java

public class QueryCache {
private final Cache<CacheKey, CacheValue> cache;
private final LoadingCache<CacheKey, CacheValue> loadingCache;

public QueryCache() {
this.cache = Caffeine.newBuilder()
.maximumSize(10000)
.expireAfterWrite(10, TimeUnit.MINUTES)
.build();

this.loadingCache = Caffeine.newBuilder()
.maximumSize(1000)
.build(this::loadFromStorage);
}

public CacheValue get(CacheKey key) {
return loadingCache.get(key);
}
}

12. 扩展性设计

12.1 插件机制

java

public class PluginManager {
private final Map<String, Plugin> plugins = new ConcurrentHashMap<>();

public void loadPlugins() {
ServiceLoader<Plugin> loader = ServiceLoader.load(Plugin.class);
for (Plugin plugin : loader) {
plugins.put(plugin.name(), plugin);
plugin.initialize();
}
}

public void installPlugin(Plugin plugin) {
plugins.put(plugin.name(), plugin);
plugin.install();
}
}

12.2 自定义处理器

java

public class ProcessorChain {
private final List<DataProcessor> processors = new CopyOnWriteArrayList<>();

public void process(DataContext context) {
for (DataProcessor processor : processors) {
if (!processor.process(context)) {
// 处理器返回 false 时中断链
break;
}
}
}

public void addProcessor(DataProcessor processor) {
processors.add(processor);
// 按优先级排序
processors.sort(Comparator.comparingInt(DataProcessor::getPriority));
}
}

13. 测试策略

13.1 单元测试

java

public class SegmentProcessorTest {
@Test
public void testSegmentProcessing() {
// 给定
SegmentProcessor processor = new SegmentProcessor();
SegmentObject segment = createTestSegment();

// 当
processor.process(segment);

// 则
verify(storageDAO).save(any(Segment.class));
}
}

13.2 集成测试

java

public class CollectorIntegrationTest {
@Test
public void testEndToEndProcessing() {
// 启动嵌入式 Collector
CollectorServer server = new CollectorServer();
server.start();

// 发送测试数据
sendTestData();

// 验证结果
assertQueryResults();

server.stop();
}
}

14. 部署和运维

14.1 容器化部署

dockerfile

FROM openjdk:11-jre-slim
COPY skywalking-collector.tar.gz /app/
WORKDIR /app
RUN tar -xzf skywalking-collector.tar.gz
EXPOSE 11800 12800
CMD ["bin/collector-startup.sh"]

14.2 配置管理

yaml

cluster:
standalone:
selector: ${SW_CLUSTER:standalone}

storage:
elasticsearch:
nameSpace: ${SW_NAMESPACE:""}
clusterNodes: ${SW_STORAGE_ES_CLUSTER_NODES:localhost:9200}

receiver-sharing-server:
default:

receiver-trace:
default:
bufferPath: ${SW_RECEIVER_BUFFER_PATH:../trace-buffer/}
bufferOffsetMaxFileSize: ${SW_RECEIVER_BUFFER_OFFSET_MAX_FILE_SIZE:100}

15. 总结

15.1 设计亮点

15.2 性能考量

15.3 可扩展性