410 lines
12 KiB
Markdown
410 lines
12 KiB
Markdown
# 碰撞避免系统架构优化设计
|
||
|
||
本文档提供了对碰撞避免系统架构的全面分析和优化建议，重点关注系统的线程管理、资源利用、生命周期控制和测试实践等方面。
|
||
|
||
## 1. 问题分析
|
||
|
||
通过代码审查和测试分析，我们发现系统存在以下架构设计问题：
|
||
|
||
### 1.1 线程管理问题
|
||
|
||
- **无限阻塞线程设计**：
|
||
- 数据处理器使用 `BlockingQueue.take()` 方法无限期阻塞线程
|
||
- 许多后台线程没有超时机制，导致资源无法及时释放
|
||
- 应用关闭时，阻塞线程不会收到通知，导致进程无法正常退出
|
||
|
||
- **生命周期管理不当**：
|
||
- 使用 `@PostConstruct` 启动线程但没有相应的 `@PreDestroy` 清理机制
|
||
- 缺少应用程序关闭钩子，导致资源无法正确释放
|
||
- 服务间依赖关系复杂，缺乏明确的启动和关闭顺序
|
||
|
||
- **线程池配置不合理**：
|
||
- 线程池配置过大（核心10线程，最大100线程）
|
||
- 队列容量设置较大（100），可能导致内存压力
|
||
- 没有设置优雅关闭参数，导致应用关闭时线程不会终止
|
||
|
||
### 1.2 资源管理问题
|
||
|
||
- **外部连接管理不当**：
|
||
- 数据采集器与外部API的连接没有合理的重试和超时策略
|
||
- HTTP客户端（RestTemplate）配置不完善，缺少超时设置
|
||
- 数据库连接（MongoDB、Redis）缺少连接池管理
|
||
|
||
- **内存管理不佳**：
|
||
- 历史数据保留机制不完善，可能导致内存溢出
|
||
- 使用大量的ConcurrentHashMap但缺少大小限制
|
||
- 没有有效的垃圾回收策略
|
||
|
||
### 1.3 测试设计问题
|
||
|
||
- **测试隔离不充分**：
|
||
- 测试运行时仍尝试连接外部资源（数据库、Kafka、数据采集API）
|
||
- 测试配置未完全禁用不必要的服务
|
||
- 缺少专门的测试数据生成机制
|
||
|
||
- **单元测试与集成测试混合**：
|
||
- 测试边界不清晰，不符合单一职责原则
|
||
- 集成测试依赖外部资源，导致测试不稳定
|
||
- 测试代码重复，缺少通用测试基类
|
||
|
||
## 2. 优化方案
|
||
|
||
### 2.1 线程管理优化
|
||
|
||
#### 2.1.1 线程池配置优化
|
||
|
||
```java
|
||
@Configuration
|
||
public class ThreadPoolConfig {
|
||
@Bean(name = "processingExecutor")
|
||
public Executor processingExecutor() {
|
||
ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
|
||
// 减小核心线程数和最大线程数
|
||
executor.setCorePoolSize(5);
|
||
executor.setMaxPoolSize(10);
|
||
// 减小队列容量，避免过多任务堆积
|
||
executor.setQueueCapacity(25);
|
||
executor.setThreadNamePrefix("data-process-");
|
||
// 添加优雅关闭配置
|
||
executor.setWaitForTasksToCompleteOnShutdown(true);
|
||
executor.setAwaitTerminationSeconds(5);
|
||
executor.initialize();
|
||
return executor;
|
||
}
|
||
|
||
// 添加专用于数据采集的线程池
|
||
@Bean(name = "collectorExecutor")
|
||
public Executor collectorExecutor() {
|
||
ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
|
||
executor.setCorePoolSize(3);
|
||
executor.setMaxPoolSize(5);
|
||
executor.setQueueCapacity(10);
|
||
executor.setThreadNamePrefix("data-collect-");
|
||
executor.setWaitForTasksToCompleteOnShutdown(true);
|
||
executor.setAwaitTerminationSeconds(5);
|
||
executor.initialize();
|
||
return executor;
|
||
}
|
||
}
|
||
```
|
||
|
||
#### 2.1.2 应用生命周期管理
|
||
|
||
```java
|
||
@Slf4j
|
||
@SpringBootApplication
|
||
@EnableScheduling
|
||
@EnableMongoRepositories
|
||
@EnableConfigurationProperties
|
||
public class CollisionAvoidanceApplication {
|
||
public static void main(String[] args) {
|
||
ConfigurableApplicationContext context = SpringApplication.run(CollisionAvoidanceApplication.class, args);
|
||
|
||
// 添加关闭钩子，确保资源正确释放
|
||
Runtime.getRuntime().addShutdownHook(new Thread(() -> {
|
||
log.info("应用程序关闭中，正在清理资源...");
|
||
try {
|
||
if (context != null && context.isActive()) {
|
||
context.close();
|
||
}
|
||
} catch (Exception e) {
|
||
log.error("关闭应用程序时出错", e);
|
||
}
|
||
log.info("应用程序已安全关闭");
|
||
}));
|
||
}
|
||
}
|
||
```
|
||
|
||
#### 2.1.3 服务组件生命周期管理
|
||
|
||
所有后台服务都应实现 `@PreDestroy` 方法：
|
||
|
||
```java
|
||
@Component
|
||
public class SomeBackgroundService {
|
||
|
||
private final AtomicBoolean running = new AtomicBoolean(false);
|
||
|
||
@PostConstruct
|
||
public void init() {
|
||
running.set(true);
|
||
// 启动逻辑
|
||
}
|
||
|
||
@PreDestroy
|
||
public void shutdown() {
|
||
log.info("关闭服务...");
|
||
running.set(false);
|
||
// 等待线程结束、释放资源等
|
||
log.info("服务已关闭");
|
||
}
|
||
}
|
||
```
|
||
|
||
### 2.2 阻塞操作优化
|
||
|
||
#### 2.2.1 使用非阻塞方法替代无限阻塞
|
||
|
||
```java
|
||
// 修改前：无限期阻塞
|
||
Map<MovingObjectType, Set<String>> delta = movingObjectRepository.takeUpdate();
|
||
|
||
// 修改后：有超时的阻塞
|
||
Map<MovingObjectType, Set<String>> delta = movingObjectRepository.pollUpdate(500, TimeUnit.MILLISECONDS);
|
||
if (delta == null || delta.isEmpty()) {
|
||
// 处理超时情况
|
||
continue;
|
||
}
|
||
```
|
||
|
||
#### 2.2.2 添加中断和状态检查
|
||
|
||
```java
|
||
private void processLoop() {
|
||
try {
|
||
while (running.get() && !Thread.currentThread().isInterrupted()) {
|
||
try {
|
||
// 处理逻辑
|
||
} catch (InterruptedException e) {
|
||
Thread.currentThread().interrupt();
|
||
log.warn("线程被中断");
|
||
break;
|
||
} catch (Exception e) {
|
||
log.error("处理异常", e);
|
||
// 添加延迟，避免CPU占用过高
|
||
Thread.sleep(1000);
|
||
}
|
||
}
|
||
} catch (Exception e) {
|
||
log.error("处理循环异常", e);
|
||
}
|
||
log.info("退出处理循环");
|
||
}
|
||
```
|
||
|
||
### 2.3 资源管理优化
|
||
|
||
#### 2.3.1 外部连接优化
|
||
|
||
对于RestTemplate：
|
||
|
||
```java
|
||
@Configuration
|
||
public class RestTemplateConfig {
|
||
@Bean
|
||
public RestTemplate restTemplate() {
|
||
RestTemplate template = new RestTemplate();
|
||
|
||
// 设置连接超时和读取超时
|
||
HttpComponentsClientHttpRequestFactory factory = new HttpComponentsClientHttpRequestFactory();
|
||
factory.setConnectTimeout(5000);
|
||
factory.setReadTimeout(5000);
|
||
factory.setConnectionRequestTimeout(5000);
|
||
|
||
template.setRequestFactory(factory);
|
||
|
||
// 添加重试机制
|
||
template.setErrorHandler(new CustomResponseErrorHandler());
|
||
|
||
return template;
|
||
}
|
||
}
|
||
```
|
||
|
||
#### 2.3.2 内存管理优化
|
||
|
||
```java
|
||
// 添加缓存容量限制
|
||
@Bean
|
||
public CacheManager cacheManager() {
|
||
SimpleCacheManager cacheManager = new SimpleCacheManager();
|
||
|
||
// 配置缓存
|
||
ConcurrentMapCache movingObjectsCache = new ConcurrentMapCache(
|
||
"movingObjects",
|
||
CacheBuilder.newBuilder()
|
||
.maximumSize(1000) // 最大条目数
|
||
.expireAfterWrite(5, TimeUnit.MINUTES) // 写入后过期时间
|
||
.build().asMap(),
|
||
false);
|
||
|
||
cacheManager.setCaches(Arrays.asList(movingObjectsCache));
|
||
return cacheManager;
|
||
}
|
||
```
|
||
|
||
### 2.4 测试设计优化
|
||
|
||
#### 2.4.1 测试配置优化
|
||
|
||
```yaml
|
||
# application-test.yml
|
||
spring:
|
||
# 禁用各种自动配置
|
||
autoconfigure:
|
||
exclude:
|
||
- org.springframework.boot.autoconfigure.mongo.MongoAutoConfiguration
|
||
- org.springframework.boot.autoconfigure.data.mongo.MongoDataAutoConfiguration
|
||
|
||
# 禁用调度任务
|
||
task:
|
||
scheduling:
|
||
enabled: false
|
||
execution:
|
||
enabled: false
|
||
|
||
data:
|
||
collector:
|
||
disabled: true # 禁用数据采集器
|
||
processor:
|
||
enabled: false # 禁用数据处理器
|
||
```
|
||
|
||
#### 2.4.2 测试基类设计
|
||
|
||
```java
|
||
@SpringBootTest
|
||
@ActiveProfiles("test")
|
||
@Import(TestConfig.class)
|
||
@DirtiesContext(classMode = DirtiesContext.ClassMode.AFTER_CLASS)
|
||
public abstract class BaseIntegrationTest {
|
||
// 通用测试设置和工具方法
|
||
|
||
@BeforeEach
|
||
public void setup() {
|
||
// 测试初始化逻辑
|
||
}
|
||
|
||
@AfterEach
|
||
public void cleanup() {
|
||
// 测试清理逻辑
|
||
}
|
||
}
|
||
```
|
||
|
||
## 3. 新的架构设计原则
|
||
|
||
### 3.1 关注点分离
|
||
|
||
- 数据采集、数据处理、碰撞检测等功能应严格分离
|
||
- 每个组件专注于单一职责，通过明确的接口进行通信
|
||
- 使用事件驱动模型代替直接依赖，降低组件间耦合
|
||
|
||
### 3.2 弹性设计
|
||
|
||
- 所有外部服务调用都应有超时、重试和熔断机制
|
||
- 系统应能在任何组件故障时保持核心功能运行
|
||
- 提供降级策略和应急模式，确保系统可用性
|
||
|
||
### 3.3 可观测性
|
||
|
||
- 添加全面的日志、指标和追踪功能
|
||
- 实现自定义健康检查，监控系统关键组件
|
||
- 设计故障注入机制，测试系统弹性
|
||
|
||
### 3.4 资源效率
|
||
|
||
- 合理配置线程池和连接池，避免资源浪费
|
||
- 实现资源限制和保护策略，防止过载
|
||
- 采用分级缓存策略，优化内存使用
|
||
|
||
## 4. 建议采用的架构模式
|
||
|
||
### 4.1 事件驱动架构
|
||
|
||
```java
|
||
// 事件定义
|
||
public class MovingObjectUpdatedEvent {
|
||
private final MovingObjectType type;
|
||
private final Set<String> objectIds;
|
||
|
||
// 构造器、getter等
|
||
}
|
||
|
||
// 发布事件
|
||
@Service
|
||
public class DataCollectorService {
|
||
private final ApplicationEventPublisher eventPublisher;
|
||
|
||
@Autowired
|
||
public DataCollectorService(ApplicationEventPublisher eventPublisher) {
|
||
this.eventPublisher = eventPublisher;
|
||
}
|
||
|
||
@Scheduled(fixedRate = 5000)
|
||
public void collectData() {
|
||
// 采集数据
|
||
|
||
// 发布事件
|
||
eventPublisher.publishEvent(new MovingObjectUpdatedEvent(type, updatedIds));
|
||
}
|
||
}
|
||
|
||
// 监听事件
|
||
@Service
|
||
public class DataProcessor {
|
||
@EventListener
|
||
public void handleMovingObjectUpdates(MovingObjectUpdatedEvent event) {
|
||
// 处理更新通知
|
||
}
|
||
}
|
||
```
|
||
|
||
### 4.2 响应式编程模型
|
||
|
||
使用 Spring WebFlux 和 Project Reactor：
|
||
|
||
```java
|
||
@Service
|
||
public class ReactiveDataService {
|
||
private final MovingObjectRepository repository;
|
||
|
||
// 提供响应式API
|
||
public Flux<MovingObject> getMovingObjects(MovingObjectType type) {
|
||
return Flux.fromIterable(repository.getByType(type).values());
|
||
}
|
||
|
||
// 响应式处理逻辑
|
||
public Mono<ProcessingResult> processData(Flux<MovingObject> objects) {
|
||
return objects
|
||
.filter(this::isValid)
|
||
.flatMap(this::transform)
|
||
.reduce(new ProcessingResult(), this::accumulate);
|
||
}
|
||
}
|
||
```
|
||
|
||
### 4.3 健康检查和监控机制
|
||
|
||
```java
|
||
@Component
|
||
public class DataCollectorHealthIndicator implements HealthIndicator {
|
||
private final DataCollectorService dataCollectorService;
|
||
|
||
@Autowired
|
||
public DataCollectorHealthIndicator(DataCollectorService dataCollectorService) {
|
||
this.dataCollectorService = dataCollectorService;
|
||
}
|
||
|
||
@Override
|
||
public Health health() {
|
||
if (!dataCollectorService.isConnected()) {
|
||
return Health.down()
|
||
.withDetail("reason", "无法连接到数据源")
|
||
.build();
|
||
}
|
||
|
||
return Health.up()
|
||
.withDetail("lastUpdateTime", dataCollectorService.getLastUpdateTime())
|
||
.withDetail("collectedItems", dataCollectorService.getCollectedItemsCount())
|
||
.build();
|
||
}
|
||
}
|
||
```
|
||
|
||
## 5. 结论
|
||
|
||
碰撞避免系统需要从根本上改进线程管理、资源使用和测试实践。我们推荐采用更现代的架构设计原则和模式，特别是事件驱动架构和响应式编程模型，以提高系统的可靠性、可伸缩性和可维护性。
|
||
|
||
通过实施这些建议，系统将能够更高效地处理并发任务，更可靠地管理资源，并提供更好的测试覆盖，从而确保在各种条件下的稳定运行。 |