深入探讨Java超时自动取消的实现方案

public class TimeoutHandler {
    private final ExecutorService executorService = Executors.newCachedThreadPool();
    
    public <T> CompletableFuture<T> withTimeout(CompletableFuture<T> future, long timeout, TimeUnit unit) {
        CompletableFuture<T> timeoutFuture = new CompletableFuture<>();
        
        // 设置超时调度
        ScheduledFuture<?> scheduledFuture = executorService.schedule(() -> {
            timeoutFuture.completeExceptionally(
                new TimeoutException("Operation timed out after " + timeout + " " + unit)
            );
        }, timeout, unit);
        
        // 注册原始任务完成的回调
        future.whenComplete((result, error) -> {
            scheduledFuture.cancel(false); // 取消超时调度
            if (error != null) {
                timeoutFuture.completeExceptionally(error);
            } else {
                timeoutFuture.complete(result);
            }
        });
        
        return timeoutFuture;
    }
    
    // 实际使用示例
    public CompletableFuture<String> executeWithTimeout(String taskId) {
        CompletableFuture<String> task = CompletableFuture.supplyAsync(() -> {
            // 实际业务逻辑
            return processTask(taskId);
        });
        
        return withTimeout(task, 5, TimeUnit.SECONDS)
            .exceptionally(throwable -> {
                // 超时或异常处理逻辑
                handleTimeout(taskId);
                throw new RuntimeException("Task execution failed", throwable);
            });
    }
}

@Configuration
public class ThreadPoolConfig {
    @Bean
    public ThreadPoolExecutor businessThreadPool() {
        return new ThreadPoolExecutor(
            10,                       // 核心线程数
            20,                       // 最大线程数
            60L,                      // 空闲线程存活时间
            TimeUnit.SECONDS,
            new LinkedBlockingQueue<>(500),  // 工作队列
            new ThreadFactoryBuilder()
                .setNameFormat("business-pool-%d")
                .setUncaughtExceptionHandler((t, e) -> log.error("Thread {} threw exception", t.getName(), e))
                .build(),
            new ThreadPoolExecutor.CallerRunsPolicy()  // 拒绝策略
        );
    }
}

@Service
public class DistributedTimeoutHandler {
    @Autowired
    private StringRedisTemplate redisTemplate;
    
    public void startTask(String taskId, long timeout) {
        // 设置任务状态和超时时间
        String taskKey = "task:" + taskId;
        redisTemplate.opsForValue().set(taskKey, "RUNNING", timeout, TimeUnit.SECONDS);
        
        // 注册超时监听器
        redisTemplate.execute(new RedisCallback<Object>() {
            @Override
            public Object doInRedis(RedisConnection connection) throws DataAccessException {
                connection.subscribe((message, pattern) -> {
                    String expiredKey = new String(message.getBody());
                    if (expiredKey.equals(taskKey)) {
                        handleTaskTimeout(taskId);
                    }
                }, "__keyevent@*__:expired".getBytes());
                return null;
            }
        });
    }
    
    private void handleTaskTimeout(String taskId) {
        // 发送取消信号
        String cancelSignalKey = "cancel:" + taskId;
        redisTemplate.opsForValue().set(cancelSignalKey, "TIMEOUT", 60, TimeUnit.SECONDS);
        
        // 通知相关服务
        notifyServices(taskId);
    }
}

@Service
public class MQTimeoutHandler {
    @Autowired
    private RocketMQTemplate rocketMQTemplate;
    
    public void scheduleTimeout(String taskId, long timeout) {
        Message<?> message = MessageBuilder.withPayload(
            new TimeoutMessage(taskId, System.currentTimeMillis())
        ).build();
        
        // 发送延迟消息
        rocketMQTemplate.syncSend(
            "TIMEOUT_TOPIC",
            message,
            timeout * 1000,  // 超时时间转换为毫秒
            delayLevel(timeout)  // 获取对应的延迟级别
        );
    }
    
    @RocketMQMessageListener(
        topic = "TIMEOUT_TOPIC",
        consumerGroup = "timeout-consumer-group"
    )
    public class TimeoutMessageListener implements RocketMQListener<TimeoutMessage> {
        @Override
        public void onMessage(TimeoutMessage message) {
            String taskId = message.getTaskId();
            // 检查任务是否仍在执行
            if (isTaskStillRunning(taskId)) {
                cancelTask(taskId);
            }
        }
    }
}

spring:
  cloud:
    gateway:
      routes:
        - id: timeout_route
          uri: lb://service-name
          predicates:
            - Path=/api/**
          filters:
            - name: CircuitBreaker
              args:
                name: myCircuitBreaker
                fallbackUri: forward:/fallback
          metadata:
            response-timeout: 5000
            connect-timeout: 1000

@Configuration
public class SentinelConfig {
    @PostConstruct
    public void init() {
        FlowRule rule = new FlowRule();
        rule.setResource("serviceA");
        rule.setGrade(RuleConstant.FLOW_GRADE_QPS);
        rule.setCount(10);
        
        DegradeRule degradeRule = new DegradeRule();
        degradeRule.setResource("serviceA");
        degradeRule.setGrade(RuleConstant.DEGRADE_GRADE_RT);
        degradeRule.setCount(200);
        degradeRule.setTimeWindow(10);
        
        FlowRuleManager.loadRules(Collections.singletonList(rule));
        DegradeRuleManager.loadRules(Collections.singletonList(degradeRule));
    }
}

@Aspect
@Component
public class TimeoutMonitorAspect {
    private final MeterRegistry registry;
    
    public TimeoutMonitorAspect(MeterRegistry registry) {
        this.registry = registry;
    }
    
    @Around("@annotation(timeout)")
    public Object monitorTimeout(ProceedingJoinPoint joinPoint, Timeout timeout) {
        Timer.Sample sample = Timer.start(registry);
        try {
            return joinPoint.proceed();
        } catch (TimeoutException e) {
            registry.counter("timeout.errors", 
                "class", joinPoint.getSignature().getDeclaringTypeName(),
                "method", joinPoint.getSignature().getName()
            ).increment();
            throw e;
        } finally {
            sample.stop(registry.timer("method.execution.time", 
                "class", joinPoint.getSignature().getDeclaringTypeName(),
                "method", joinPoint.getSignature().getName()
            ));
        }
    }
}