790 lines
24 KiB
Markdown
790 lines
24 KiB
Markdown
# Prompt - 性能优化和限流处理
|
||
|
||
## 输入引用
|
||
|
||
引用相关的 docs 文档链接：
|
||
|
||
- [REQ-010-17.md](../requirements/REQ-010-17.md) - 性能优化和限流处理需求文档
|
||
- [0026-performance-optimization.md](../decisions/adr/0026-performance-optimization.md) - 性能优化和限流处理架构决策
|
||
- [REQ-010.md](../requirements/REQ-010.md) - Salesforce元数据拉取和部署主需求文档
|
||
- [REQ-010-1.md](../requirements/REQ-010-1.md) - 数据库表结构设计和创建
|
||
- [REQ-010-2.md](../requirements/REQ-010-2.md) - 基础实体类和Mapper创建
|
||
- [REQ-010-5.md](../requirements/REQ-010-5.md) - Metadata API客户端封装
|
||
- [REQ-010-6.md](../requirements/REQ-010-6.md) - 元数据拉取核心功能
|
||
- [REQ-010-8.md](../requirements/REQ-010-8.md) - 元数据部署核心功能
|
||
- [metadata-module.md](../reference-code/com/docs/metadata-module.md) - Salesforce Metadata API 模块说明（唯一真源）
|
||
- [index.md](../reference-code/com/docs/index.md) - Salesforce SOAP API Java 客户端参考文档（唯一真源）
|
||
|
||
## Context Maps
|
||
|
||
强制列出本次 Prompt 依赖的 Canvas 文件：
|
||
|
||
- [Authentication.canvas](../Authentication.canvas) - 项目架构视觉化展示
|
||
- **相关节点**: [集成核心](node_integration_core) - 提供与Salesforce的各种连接方式
|
||
- **快照时间**: 2026-01-19 00:00:00
|
||
|
||
## 目标
|
||
|
||
实现性能优化和限流处理功能，包括API限流处理、并发控制、缓存优化、数据库优化、文件处理优化、性能监控。优化系统性能，避免API限流，提高系统稳定性。
|
||
|
||
## 输出格式
|
||
|
||
代码示例（语言：Java）
|
||
|
||
## 约束
|
||
|
||
- **技术栈限制**: 必须基于现有的Spring Boot 3 + Vue 3技术栈
|
||
- **架构约束**: 必须遵循Authentication.canvas中定义的架构和调用关系
|
||
- **模块约束**: 必须在datai-salesforce-metadata模块下实现
|
||
- **数据库约束**: 必须使用MyBatis Plus作为持久层框架
|
||
- **性能约束**: 系统性能必须满足要求，不能影响用户体验
|
||
- **依赖约束**: 必须依赖于所有其他子需求
|
||
|
||
## Rule Set
|
||
|
||
"请严格参考 @Authentication.canvas 中的状态机转移逻辑，不要自行发挥。"
|
||
|
||
**具体规则**：
|
||
- 必须使用Canvas中定义的类名和方法名
|
||
- 必须遵循Canvas中定义的调用关系
|
||
- 必须参考Canvas中的流程图逻辑
|
||
|
||
## 验收标准
|
||
|
||
- **功能完整性**: 所有性能优化和限流处理功能能够正常工作
|
||
- **性能指标**: 系统性能满足要求，API响应时间、数据库查询时间、文件处理时间都在合理范围内
|
||
- **稳定性**: 系统稳定性高，不会因为性能问题导致系统崩溃
|
||
- **代码规范性**: 代码符合项目编码规范，有清晰的注释
|
||
- **可维护性**: 代码结构清晰，易于扩展和维护
|
||
- **可测试性**: 代码易于单元测试和集成测试
|
||
|
||
## 风险
|
||
|
||
- **限流处理风险**: 限流处理不当可能导致API调用失败
|
||
- **并发控制风险**: 并发控制不当可能导致系统资源耗尽
|
||
- **缓存风险**: 缓存不一致可能导致数据错误
|
||
- **数据库优化风险**: 数据库优化不当可能导致查询性能下降
|
||
- **文件处理风险**: 文件处理优化不当可能导致内存溢出
|
||
- **性能监控风险**: 性能监控不准确可能导致误判
|
||
|
||
## 实现指南
|
||
|
||
### 1. API限流处理
|
||
|
||
#### 1.1 创建限流配置类
|
||
|
||
```java
|
||
package com.datai.salesforce.metadata.config;
|
||
|
||
import com.google.common.util.concurrent.RateLimiter;
|
||
import lombok.Data;
|
||
import org.springframework.boot.context.properties.ConfigurationProperties;
|
||
import org.springframework.context.annotation.Bean;
|
||
import org.springframework.context.annotation.Configuration;
|
||
|
||
@Data
|
||
@Configuration
|
||
@ConfigurationProperties(prefix = "rate-limit")
|
||
public class RateLimitConfig {
|
||
|
||
private int permitsPerSecond = 10;
|
||
private int maxRetries = 3;
|
||
private long initialRetryInterval = 1000;
|
||
private double retryMultiplier = 2.0;
|
||
|
||
@Bean
|
||
public RateLimiter rateLimiter() {
|
||
return RateLimiter.create(permitsPerSecond);
|
||
}
|
||
}
|
||
```
|
||
|
||
#### 1.2 创建限流注解
|
||
|
||
```java
|
||
package com.datai.salesforce.metadata.annotation;
|
||
|
||
import java.lang.annotation.*;
|
||
|
||
@Target(ElementType.METHOD)
|
||
@Retention(RetentionPolicy.RUNTIME)
|
||
@Documented
|
||
public @interface RateLimit {
|
||
|
||
int permitsPerSecond() default 10;
|
||
|
||
int maxRetries() default 3;
|
||
|
||
long initialRetryInterval() default 1000;
|
||
|
||
double retryMultiplier() default 2.0;
|
||
}
|
||
```
|
||
|
||
#### 1.3 创建限流切面
|
||
|
||
```java
|
||
package com.datai.salesforce.metadata.aspect;
|
||
|
||
import com.google.common.util.concurrent.RateLimiter;
|
||
import lombok.RequiredArgsConstructor;
|
||
import lombok.extern.slf4j.Slf4j;
|
||
import org.aspectj.lang.ProceedingJoinPoint;
|
||
import org.aspectj.lang.annotation.Around;
|
||
import org.aspectj.lang.annotation.Aspect;
|
||
import org.springframework.stereotype.Component;
|
||
|
||
@Slf4j
|
||
@Aspect
|
||
@Component
|
||
@RequiredArgsConstructor
|
||
public class RateLimitAspect {
|
||
|
||
private final RateLimiter rateLimiter;
|
||
|
||
@Around("@annotation(rateLimit)")
|
||
public Object around(ProceedingJoinPoint joinPoint, RateLimit rateLimit) throws Throwable {
|
||
if (!rateLimiter.tryAcquire()) {
|
||
log.warn("Rate limit exceeded, method: {}", joinPoint.getSignature().getName());
|
||
throw new RuntimeException("Rate limit exceeded");
|
||
}
|
||
return joinPoint.proceed();
|
||
}
|
||
}
|
||
```
|
||
|
||
#### 1.4 创建限流异常类
|
||
|
||
```java
|
||
package com.datai.salesforce.metadata.exception;
|
||
|
||
import lombok.Getter;
|
||
|
||
@Getter
|
||
public class RateLimitException extends RuntimeException {
|
||
|
||
private final int statusCode;
|
||
|
||
public RateLimitException(int statusCode, String message) {
|
||
super(message);
|
||
this.statusCode = statusCode;
|
||
}
|
||
|
||
public RateLimitException(int statusCode, String message, Throwable cause) {
|
||
super(message, cause);
|
||
this.statusCode = statusCode;
|
||
}
|
||
}
|
||
```
|
||
|
||
#### 1.5 创建限流服务
|
||
|
||
```java
|
||
package com.datai.salesforce.metadata.service;
|
||
|
||
import com.google.common.util.concurrent.RateLimiter;
|
||
import lombok.RequiredArgsConstructor;
|
||
import lombok.extern.slf4j.Slf4j;
|
||
import org.springframework.retry.annotation.Backoff;
|
||
import org.springframework.retry.annotation.Retryable;
|
||
import org.springframework.stereotype.Service;
|
||
|
||
@Slf4j
|
||
@Service
|
||
@RequiredArgsConstructor
|
||
public class RateLimitService {
|
||
|
||
private final RateLimiter rateLimiter;
|
||
|
||
@Retryable(
|
||
value = {RateLimitException.class},
|
||
maxAttempts = 3,
|
||
backoff = @Backoff(delay = 1000, multiplier = 2.0)
|
||
)
|
||
public <T> T executeWithRateLimit(RunnableWithResult<T> runnable) {
|
||
if (!rateLimiter.tryAcquire()) {
|
||
log.warn("Rate limit exceeded");
|
||
throw new RateLimitException(429, "Rate limit exceeded");
|
||
}
|
||
return runnable.run();
|
||
}
|
||
|
||
@FunctionalInterface
|
||
public interface RunnableWithResult<T> {
|
||
T run();
|
||
}
|
||
}
|
||
```
|
||
|
||
### 2. 并发控制
|
||
|
||
#### 2.1 创建线程池配置类
|
||
|
||
```java
|
||
package com.datai.salesforce.metadata.config;
|
||
|
||
import lombok.Data;
|
||
import org.springframework.boot.context.properties.ConfigurationProperties;
|
||
import org.springframework.context.annotation.Bean;
|
||
import org.springframework.context.annotation.Configuration;
|
||
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
|
||
|
||
import java.util.concurrent.Executor;
|
||
import java.util.concurrent.PriorityBlockingQueue;
|
||
import java.util.concurrent.ThreadPoolExecutor;
|
||
|
||
@Data
|
||
@Configuration
|
||
@ConfigurationProperties(prefix = "thread-pool")
|
||
public class ThreadPoolConfig {
|
||
|
||
private int corePoolSize = 5;
|
||
private int maxPoolSize = 10;
|
||
private int queueCapacity = 100;
|
||
private long keepAliveTime = 60;
|
||
|
||
@Bean(name = "taskExecutor")
|
||
public Executor taskExecutor() {
|
||
ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
|
||
executor.setCorePoolSize(corePoolSize);
|
||
executor.setMaxPoolSize(maxPoolSize);
|
||
executor.setQueueCapacity(queueCapacity);
|
||
executor.setKeepAliveSeconds((int) keepAliveTime);
|
||
executor.setThreadNamePrefix("task-executor-");
|
||
executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
|
||
executor.initialize();
|
||
return executor;
|
||
}
|
||
|
||
@Bean(name = "priorityTaskExecutor")
|
||
public Executor priorityTaskExecutor() {
|
||
ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
|
||
executor.setCorePoolSize(corePoolSize);
|
||
executor.setMaxPoolSize(maxPoolSize);
|
||
executor.setQueueCapacity(queueCapacity);
|
||
executor.setKeepAliveSeconds((int) keepAliveTime);
|
||
executor.setThreadNamePrefix("priority-task-executor-");
|
||
executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
|
||
executor.setTaskDecorator(runnable -> {
|
||
if (runnable instanceof PriorityTask) {
|
||
return (PriorityTask) runnable;
|
||
}
|
||
return new PriorityTask(runnable, 5);
|
||
});
|
||
executor.initialize();
|
||
return executor;
|
||
}
|
||
}
|
||
```
|
||
|
||
#### 2.2 创建优先级任务类
|
||
|
||
```java
|
||
package com.datai.salesforce.metadata.task;
|
||
|
||
import lombok.Getter;
|
||
|
||
@Getter
|
||
public class PriorityTask implements Runnable, Comparable<PriorityTask> {
|
||
|
||
private final Runnable runnable;
|
||
private final int priority;
|
||
|
||
public PriorityTask(Runnable runnable, int priority) {
|
||
this.runnable = runnable;
|
||
this.priority = priority;
|
||
}
|
||
|
||
@Override
|
||
public void run() {
|
||
runnable.run();
|
||
}
|
||
|
||
@Override
|
||
public int compareTo(PriorityTask other) {
|
||
return Integer.compare(other.priority, this.priority);
|
||
}
|
||
}
|
||
```
|
||
|
||
#### 2.3 创建并发控制服务
|
||
|
||
```java
|
||
package com.datai.salesforce.metadata.service;
|
||
|
||
import lombok.RequiredArgsConstructor;
|
||
import lombok.extern.slf4j.Slf4j;
|
||
import org.springframework.scheduling.annotation.Async;
|
||
import org.springframework.stereotype.Service;
|
||
|
||
import java.util.concurrent.CompletableFuture;
|
||
import java.util.concurrent.Executor;
|
||
|
||
@Slf4j
|
||
@Service
|
||
@RequiredArgsConstructor
|
||
public class ConcurrencyControlService {
|
||
|
||
private final Executor taskExecutor;
|
||
private final Executor priorityTaskExecutor;
|
||
|
||
@Async("taskExecutor")
|
||
public CompletableFuture<Void> executeAsync(Runnable runnable) {
|
||
try {
|
||
runnable.run();
|
||
return CompletableFuture.completedFuture(null);
|
||
} catch (Exception e) {
|
||
log.error("Execute async task failed", e);
|
||
return CompletableFuture.failedFuture(e);
|
||
}
|
||
}
|
||
|
||
@Async("priorityTaskExecutor")
|
||
public CompletableFuture<Void> executeAsyncWithPriority(Runnable runnable, int priority) {
|
||
try {
|
||
runnable.run();
|
||
return CompletableFuture.completedFuture(null);
|
||
} catch (Exception e) {
|
||
log.error("Execute async task with priority failed", e);
|
||
return CompletableFuture.failedFuture(e);
|
||
}
|
||
}
|
||
}
|
||
```
|
||
|
||
### 3. 缓存优化
|
||
|
||
#### 3.1 创建缓存配置类
|
||
|
||
```java
|
||
package com.datai.salesforce.metadata.config;
|
||
|
||
import org.springframework.cache.annotation.EnableCaching;
|
||
import org.springframework.context.annotation.Bean;
|
||
import org.springframework.context.annotation.Configuration;
|
||
import org.springframework.data.redis.cache.RedisCacheConfiguration;
|
||
import org.springframework.data.redis.cache.RedisCacheManager;
|
||
import org.springframework.data.redis.connection.RedisConnectionFactory;
|
||
import org.springframework.data.redis.serializer.GenericJackson2JsonRedisSerializer;
|
||
import org.springframework.data.redis.serializer.RedisSerializationContext;
|
||
import org.springframework.data.redis.serializer.StringRedisSerializer;
|
||
|
||
import java.time.Duration;
|
||
|
||
@Configuration
|
||
@EnableCaching
|
||
public class CacheConfig {
|
||
|
||
@Bean
|
||
public RedisCacheManager cacheManager(RedisConnectionFactory connectionFactory) {
|
||
RedisCacheConfiguration config = RedisCacheConfiguration.defaultCacheConfig()
|
||
.entryTtl(Duration.ofHours(1))
|
||
.serializeKeysWith(RedisSerializationContext.SerializationPair.fromSerializer(new StringRedisSerializer()))
|
||
.serializeValuesWith(RedisSerializationContext.SerializationPair.fromSerializer(new GenericJackson2JsonRedisSerializer()))
|
||
.disableCachingNullValues();
|
||
|
||
return RedisCacheManager.builder(connectionFactory)
|
||
.cacheDefaults(config)
|
||
.transactionAware()
|
||
.build();
|
||
}
|
||
}
|
||
```
|
||
|
||
#### 3.2 创建缓存服务
|
||
|
||
```java
|
||
package com.datai.salesforce.metadata.service;
|
||
|
||
import com.datai.salesforce.metadata.entity.OrgConfig;
|
||
import com.datai.salesforce.metadata.entity.TaskDefinition;
|
||
import com.datai.salesforce.metadata.entity.MetadataComponent;
|
||
import com.datai.salesforce.metadata.mapper.OrgConfigMapper;
|
||
import com.datai.salesforce.metadata.mapper.TaskDefinitionMapper;
|
||
import com.datai.salesforce.metadata.mapper.MetadataComponentMapper;
|
||
import lombok.RequiredArgsConstructor;
|
||
import lombok.extern.slf4j.Slf4j;
|
||
import org.springframework.cache.annotation.CacheEvict;
|
||
import org.springframework.cache.annotation.CachePut;
|
||
import org.springframework.cache.annotation.Cacheable;
|
||
import org.springframework.stereotype.Service;
|
||
|
||
import java.util.List;
|
||
|
||
@Slf4j
|
||
@Service
|
||
@RequiredArgsConstructor
|
||
public class CacheService {
|
||
|
||
private final OrgConfigMapper orgConfigMapper;
|
||
private final TaskDefinitionMapper taskDefinitionMapper;
|
||
private final MetadataComponentMapper metadataComponentMapper;
|
||
|
||
@Cacheable(value = "orgConfig", key = "#id", unless = "#result == null")
|
||
public OrgConfig getOrgConfigById(Long id) {
|
||
log.info("Cache miss for org config: {}", id);
|
||
return orgConfigMapper.selectById(id);
|
||
}
|
||
|
||
@CachePut(value = "orgConfig", key = "#orgConfig.id")
|
||
public OrgConfig updateOrgConfig(OrgConfig orgConfig) {
|
||
orgConfigMapper.updateById(orgConfig);
|
||
return orgConfig;
|
||
}
|
||
|
||
@CacheEvict(value = "orgConfig", key = "#id")
|
||
public void deleteOrgConfig(Long id) {
|
||
orgConfigMapper.deleteById(id);
|
||
}
|
||
|
||
@Cacheable(value = "taskDefinition", key = "#id", unless = "#result == null")
|
||
public TaskDefinition getTaskDefinitionById(Long id) {
|
||
log.info("Cache miss for task definition: {}", id);
|
||
return taskDefinitionMapper.selectById(id);
|
||
}
|
||
|
||
@CachePut(value = "taskDefinition", key = "#taskDefinition.id")
|
||
public TaskDefinition updateTaskDefinition(TaskDefinition taskDefinition) {
|
||
taskDefinitionMapper.updateById(taskDefinition);
|
||
return taskDefinition;
|
||
}
|
||
|
||
@CacheEvict(value = "taskDefinition", key = "#id")
|
||
public void deleteTaskDefinition(Long id) {
|
||
taskDefinitionMapper.deleteById(id);
|
||
}
|
||
|
||
@Cacheable(value = "metadataComponent", key = "#id", unless = "#result == null")
|
||
public MetadataComponent getMetadataComponentById(Long id) {
|
||
log.info("Cache miss for metadata component: {}", id);
|
||
return metadataComponentMapper.selectById(id);
|
||
}
|
||
|
||
@CachePut(value = "metadataComponent", key = "#metadataComponent.id")
|
||
public MetadataComponent updateMetadataComponent(MetadataComponent metadataComponent) {
|
||
metadataComponentMapper.updateById(metadataComponent);
|
||
return metadataComponent;
|
||
}
|
||
|
||
@CacheEvict(value = "metadataComponent", key = "#id")
|
||
public void deleteMetadataComponent(Long id) {
|
||
metadataComponentMapper.deleteById(id);
|
||
}
|
||
}
|
||
```
|
||
|
||
### 4. 数据库优化
|
||
|
||
#### 4.1 创建批量操作服务
|
||
|
||
```java
|
||
package com.datai.salesforce.metadata.service;
|
||
|
||
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
|
||
import com.datai.salesforce.metadata.entity.MetadataComponent;
|
||
import com.datai.salesforce.metadata.mapper.MetadataComponentMapper;
|
||
import lombok.RequiredArgsConstructor;
|
||
import lombok.extern.slf4j.Slf4j;
|
||
import org.springframework.stereotype.Service;
|
||
|
||
import java.util.List;
|
||
|
||
@Slf4j
|
||
@Service
|
||
@RequiredArgsConstructor
|
||
public class BatchOperationService extends ServiceImpl<MetadataComponentMapper, MetadataComponent> {
|
||
|
||
private final MetadataComponentMapper metadataComponentMapper;
|
||
|
||
public boolean saveBatch(List<MetadataComponent> list) {
|
||
return saveBatch(list, 100);
|
||
}
|
||
|
||
public boolean updateBatchById(List<MetadataComponent> list) {
|
||
return updateBatchById(list, 100);
|
||
}
|
||
|
||
public boolean removeBatchByIds(List<Long> ids) {
|
||
return removeByIds(ids);
|
||
}
|
||
}
|
||
```
|
||
|
||
#### 4.2 创建查询优化服务
|
||
|
||
```java
|
||
package com.datai.salesforce.metadata.service;
|
||
|
||
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
|
||
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
|
||
import com.datai.salesforce.metadata.entity.MetadataComponent;
|
||
import com.datai.salesforce.metadata.mapper.MetadataComponentMapper;
|
||
import lombok.RequiredArgsConstructor;
|
||
import lombok.extern.slf4j.Slf4j;
|
||
import org.springframework.stereotype.Service;
|
||
|
||
@Slf4j
|
||
@Service
|
||
@RequiredArgsConstructor
|
||
public class QueryOptimizationService {
|
||
|
||
private final MetadataComponentMapper metadataComponentMapper;
|
||
|
||
public Page<MetadataComponent> pageQuery(int pageNum, int pageSize, String componentType, String apiName) {
|
||
QueryWrapper<MetadataComponent> queryWrapper = new QueryWrapper<>();
|
||
if (componentType != null && !componentType.isEmpty()) {
|
||
queryWrapper.eq("component_type", componentType);
|
||
}
|
||
if (apiName != null && !apiName.isEmpty()) {
|
||
queryWrapper.like("api_name", apiName);
|
||
}
|
||
queryWrapper.orderByDesc("create_time");
|
||
return metadataComponentMapper.selectPage(new Page<>(pageNum, pageSize), queryWrapper);
|
||
}
|
||
|
||
public List<MetadataComponent> batchQuery(List<Long> ids) {
|
||
return metadataComponentMapper.selectBatchIds(ids);
|
||
}
|
||
}
|
||
```
|
||
|
||
### 5. 文件处理优化
|
||
|
||
#### 5.1 创建流式文件处理服务
|
||
|
||
```java
|
||
package com.datai.salesforce.metadata.service;
|
||
|
||
import lombok.RequiredArgsConstructor;
|
||
import lombok.extern.slf4j.Slf4j;
|
||
import org.springframework.stereotype.Service;
|
||
|
||
import java.io.*;
|
||
import java.nio.file.Files;
|
||
import java.nio.file.Path;
|
||
import java.util.zip.GZIPInputStream;
|
||
import java.util.zip.GZIPOutputStream;
|
||
|
||
@Slf4j
|
||
@Service
|
||
@RequiredArgsConstructor
|
||
public class FileProcessingService {
|
||
|
||
public void processLargeFile(Path filePath, FileProcessor processor) throws IOException {
|
||
try (BufferedReader reader = Files.newBufferedReader(filePath)) {
|
||
String line;
|
||
while ((line = reader.readLine()) != null) {
|
||
processor.process(line);
|
||
}
|
||
}
|
||
}
|
||
|
||
public void compressFile(Path sourcePath, Path targetPath) throws IOException {
|
||
try (InputStream in = Files.newInputStream(sourcePath);
|
||
OutputStream out = new GZIPOutputStream(Files.newOutputStream(targetPath))) {
|
||
byte[] buffer = new byte[8192];
|
||
int len;
|
||
while ((len = in.read(buffer)) > 0) {
|
||
out.write(buffer, 0, len);
|
||
}
|
||
}
|
||
}
|
||
|
||
public void decompressFile(Path sourcePath, Path targetPath) throws IOException {
|
||
try (InputStream in = new GZIPInputStream(Files.newInputStream(sourcePath));
|
||
OutputStream out = Files.newOutputStream(targetPath)) {
|
||
byte[] buffer = new byte[8192];
|
||
int len;
|
||
while ((len = in.read(buffer)) > 0) {
|
||
out.write(buffer, 0, len);
|
||
}
|
||
}
|
||
}
|
||
|
||
@FunctionalInterface
|
||
public interface FileProcessor {
|
||
void process(String line);
|
||
}
|
||
}
|
||
```
|
||
|
||
#### 5.2 创建并行文件处理服务
|
||
|
||
```java
|
||
package com.datai.salesforce.metadata.service;
|
||
|
||
import lombok.RequiredArgsConstructor;
|
||
import lombok.extern.slf4j.Slf4j;
|
||
import org.springframework.scheduling.annotation.Async;
|
||
import org.springframework.stereotype.Service;
|
||
|
||
import java.nio.file.Path;
|
||
import java.util.List;
|
||
import java.util.concurrent.CompletableFuture;
|
||
import java.util.stream.Collectors;
|
||
|
||
@Slf4j
|
||
@Service
|
||
@RequiredArgsConstructor
|
||
public class ParallelFileProcessingService {
|
||
|
||
@Async
|
||
public CompletableFuture<Void> processFilesInParallel(List<Path> filePaths, FileProcessingService.FileProcessor processor) {
|
||
List<CompletableFuture<Void>> futures = filePaths.stream()
|
||
.map(filePath -> CompletableFuture.runAsync(() -> {
|
||
try {
|
||
processFile(filePath, processor);
|
||
} catch (Exception e) {
|
||
log.error("Process file failed: {}", filePath, e);
|
||
}
|
||
}))
|
||
.collect(Collectors.toList());
|
||
|
||
return CompletableFuture.allOf(futures.toArray(new CompletableFuture[0]));
|
||
}
|
||
|
||
private void processFile(Path filePath, FileProcessingService.FileProcessor processor) throws Exception {
|
||
try (BufferedReader reader = java.nio.file.Files.newBufferedReader(filePath)) {
|
||
String line;
|
||
while ((line = reader.readLine()) != null) {
|
||
processor.process(line);
|
||
}
|
||
}
|
||
}
|
||
}
|
||
```
|
||
|
||
### 6. 性能监控
|
||
|
||
#### 6.1 创建性能监控配置类
|
||
|
||
```java
|
||
package com.datai.salesforce.metadata.config;
|
||
|
||
import io.micrometer.core.instrument.MeterRegistry;
|
||
import io.micrometer.core.instrument.Timer;
|
||
import org.springframework.context.annotation.Bean;
|
||
import org.springframework.context.annotation.Configuration;
|
||
|
||
@Configuration
|
||
public class PerformanceMonitoringConfig {
|
||
|
||
@Bean
|
||
public Timer apiCallTimer(MeterRegistry registry) {
|
||
return Timer.builder("api.call.timer")
|
||
.description("API call timer")
|
||
.register(registry);
|
||
}
|
||
|
||
@Bean
|
||
public Timer databaseQueryTimer(MeterRegistry registry) {
|
||
return Timer.builder("database.query.timer")
|
||
.description("Database query timer")
|
||
.register(registry);
|
||
}
|
||
|
||
@Bean
|
||
public Timer fileProcessingTimer(MeterRegistry registry) {
|
||
return Timer.builder("file.processing.timer")
|
||
.description("File processing timer")
|
||
.register(registry);
|
||
}
|
||
}
|
||
```
|
||
|
||
#### 6.2 创建性能监控服务
|
||
|
||
```java
|
||
package com.datai.salesforce.metadata.service;
|
||
|
||
import io.micrometer.core.instrument.Counter;
|
||
import io.micrometer.core.instrument.MeterRegistry;
|
||
import io.micrometer.core.instrument.Timer;
|
||
import lombok.RequiredArgsConstructor;
|
||
import lombok.extern.slf4j.Slf4j;
|
||
import org.springframework.stereotype.Service;
|
||
|
||
import java.util.concurrent.TimeUnit;
|
||
|
||
@Slf4j
|
||
@Service
|
||
@RequiredArgsConstructor
|
||
public class PerformanceMonitoringService {
|
||
|
||
private final MeterRegistry meterRegistry;
|
||
private final Timer apiCallTimer;
|
||
private final Timer databaseQueryTimer;
|
||
private final Timer fileProcessingTimer;
|
||
|
||
public void recordApiCall() {
|
||
Counter.builder("api.call.count")
|
||
.description("API call count")
|
||
.register(meterRegistry)
|
||
.increment();
|
||
}
|
||
|
||
public void recordApiCallTime(long duration, TimeUnit unit) {
|
||
apiCallTimer.record(duration, unit);
|
||
}
|
||
|
||
public void recordDatabaseQueryTime(long duration, TimeUnit unit) {
|
||
databaseQueryTimer.record(duration, unit);
|
||
}
|
||
|
||
public void recordFileProcessingTime(long duration, TimeUnit unit) {
|
||
fileProcessingTimer.record(duration, unit);
|
||
}
|
||
|
||
public <T> T recordApiCall(RunnableWithResult<T> runnable) {
|
||
long startTime = System.currentTimeMillis();
|
||
try {
|
||
recordApiCall();
|
||
T result = runnable.run();
|
||
long duration = System.currentTimeMillis() - startTime;
|
||
recordApiCallTime(duration, TimeUnit.MILLISECONDS);
|
||
return result;
|
||
} catch (Exception e) {
|
||
log.error("API call failed", e);
|
||
throw new RuntimeException(e);
|
||
}
|
||
}
|
||
|
||
public <T> T recordDatabaseQuery(RunnableWithResult<T> runnable) {
|
||
long startTime = System.currentTimeMillis();
|
||
try {
|
||
T result = runnable.run();
|
||
long duration = System.currentTimeMillis() - startTime;
|
||
recordDatabaseQueryTime(duration, TimeUnit.MILLISECONDS);
|
||
return result;
|
||
} catch (Exception e) {
|
||
log.error("Database query failed", e);
|
||
throw new RuntimeException(e);
|
||
}
|
||
}
|
||
|
||
public <T> T recordFileProcessing(RunnableWithResult<T> runnable) {
|
||
long startTime = System.currentTimeMillis();
|
||
try {
|
||
T result = runnable.run();
|
||
long duration = System.currentTimeMillis() - startTime;
|
||
recordFileProcessingTime(duration, TimeUnit.MILLISECONDS);
|
||
return result;
|
||
} catch (Exception e) {
|
||
log.error("File processing failed", e);
|
||
throw new RuntimeException(e);
|
||
}
|
||
}
|
||
|
||
@FunctionalInterface
|
||
public interface RunnableWithResult<T> {
|
||
T run() throws Exception;
|
||
}
|
||
}
|
||
```
|
||
|
||
## 使用记录
|
||
|
||
| 日期 | 使用场景 | 输入参数 | 输出结果 | 反馈 | 改进措施 |
|
||
|------|---------|---------|---------|------|----------|
|
||
| 2026-01-19 | 实现性能优化和限流处理功能 | REQ-010-17需求文档，0026-performance-optimization ADR文档 | 代码示例和实现指南 | 无 | 无 |
|