kris/datai

Fork 0

Kris 3d290a8dd1 feat：归档当前所有文件

2026-01-28 18:23:25 +08:00

24 KiB

Raw Blame History

Prompt - 性能优化和限流处理

输入引用

引用相关的 docs 文档链接：

REQ-010-17.md - 性能优化和限流处理需求文档
0026-performance-optimization.md - 性能优化和限流处理架构决策
REQ-010.md - Salesforce元数据拉取和部署主需求文档
REQ-010-1.md - 数据库表结构设计和创建
REQ-010-2.md - 基础实体类和Mapper创建
REQ-010-5.md - Metadata API客户端封装
REQ-010-6.md - 元数据拉取核心功能
REQ-010-8.md - 元数据部署核心功能
metadata-module.md - Salesforce Metadata API 模块说明（唯一真源）
index.md - Salesforce SOAP API Java 客户端参考文档（唯一真源）

Context Maps

强制列出本次 Prompt 依赖的 Canvas 文件：

Authentication.canvas - 项目架构视觉化展示
相关节点: 集成核心 - 提供与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 创建限流配置类

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 创建限流注解

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 创建限流切面

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 创建限流异常类

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 创建限流服务

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 创建线程池配置类

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 创建优先级任务类

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 创建并发控制服务

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 创建缓存配置类

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 创建缓存服务

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 创建批量操作服务

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 创建查询优化服务

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 创建流式文件处理服务

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 创建并行文件处理服务

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 创建性能监控配置类

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 创建性能监控服务

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文档	代码示例和实现指南	无	无

24 KiB Raw Blame History Unescape Escape