public class ConcurrencyController {
    private Semaphore semaphore;
    private int maxConcurrency;
    
    public ConcurrencyController(int maxConcurrency) {
        this.maxConcurrency = maxConcurrency;
        this.semaphore = new Semaphore(maxConcurrency);
    }
    
    public void acquire() throws InterruptedException {
        semaphore.acquire();
    }
    
    public void release() {
        semaphore.release();
    }
    
    public int getAvailablePermits() {
        return semaphore.availablePermits();
    }
}

线程池控制

通过线程池控制并发执行的任务数量：

public class TaskExecutor {
    private ExecutorService executorService;
    private ConcurrencyController concurrencyController;
    
    public TaskExecutor(int maxConcurrency) {
        this.executorService = Executors.newFixedThreadPool(maxConcurrency);
        this.concurrencyController = new ConcurrencyController(maxConcurrency);
    }
    
    public Future<TestResult> submitTask(TestTask task) {
        return executorService.submit(() -> {
            try {
                concurrencyController.acquire();
                return task.execute();
            } finally {
                concurrencyController.release();
            }
        });
    }
}

分布式锁

在分布式环境中，需要使用分布式锁来控制并发：

public class DistributedConcurrencyController {
    private RedisTemplate<String, String> redisTemplate;
    private String lockKey;
    private int maxConcurrency;
    
    public boolean tryAcquire() {
        String script = "local current = redis.call('GET', KEYS[1]) " +
                       "if current == false or tonumber(current) < tonumber(ARGV[1]) then " +
                       "  redis.call('INCR', KEYS[1]) " +
                       "  redis.call('EXPIRE', KEYS[1], ARGV[2]) " +
                       "  return 1 " +
                       "else " +
                       "  return 0 " +
                       "end";
        
        Long result = redisTemplate.execute(
            new DefaultRedisScript<>(script, Long.class),
            Collections.singletonList(lockKey),
            String.valueOf(maxConcurrency),
            "300" // 5分钟过期
        );
        
        return result != null && result == 1;
    }
    
    public void release() {
        redisTemplate.boundValueOps(lockKey).decrement();
    }
}

队列机制设计

队列机制是任务调度的重要组成部分，用于管理等待执行的任务。

队列类型选择

根据不同场景选择合适的队列类型：

FIFO队列：先进先出，适用于大多数场景
优先级队列：根据优先级排序，适用于需要优先执行的任务
延迟队列：支持延迟执行，适用于定时任务
阻塞队列：支持阻塞操作，适用于生产者-消费者模式

队列实现

基于Redis实现分布式任务队列：

public class TaskQueue {
    private RedisTemplate<String, Object> redisTemplate;
    private String queueName;
    
    // 入队
    public void enqueue(TestTask task) {
        redisTemplate.opsForList().leftPush(queueName, task);
    }
    
    // 出队
    public TestTask dequeue() {
        return (TestTask) redisTemplate.opsForList().rightPop(queueName);
    }
    
    // 阻塞出队
    public TestTask dequeue(long timeout, TimeUnit unit) {
        return (TestTask) redisTemplate.opsForList().rightPop(queueName, timeout, unit);
    }
    
    // 获取队列长度
    public long size() {
        return redisTemplate.opsForList().size(queueName);
    }
    
    // 优先级队列入队
    public void enqueueWithPriority(TestTask task, int priority) {
        redisTemplate.opsForZSet().add(queueName + ":priority", task, priority);
    }
    
    // 优先级队列出队
    public TestTask dequeueWithPriority() {
        Set<Object> tasks = redisTemplate.opsForZSet().range(queueName + ":priority", 0, 0);
        if (tasks != null && !tasks.isEmpty()) {
            TestTask task = (TestTask) tasks.iterator().next();
            redisTemplate.opsForZSet().remove(queueName + ":priority", task);
            return task;
        }
        return null;
    }
}

队列监控

建立队列监控机制，实时了解队列状态：

队列长度监控：监控队列中等待的任务数量
处理速度监控：监控任务的处理速度
等待时间监控：监控任务在队列中的等待时间
队列健康检查：定期检查队列的健康状态

优先级调度策略

优先级调度能够确保重要任务得到及时执行。

优先级定义

合理定义任务优先级：

业务优先级：根据业务重要性定义优先级
时间敏感性：根据任务的时间敏感性定义优先级
资源需求：根据任务的资源需求定义优先级
用户等级：根据用户等级定义优先级

优先级实现

public enum TaskPriority {
    HIGHEST(1, "最高优先级"),
    HIGH(2, "高优先级"),
    NORMAL(3, "普通优先级"),
    LOW(4, "低优先级"),
    LOWEST(5, "最低优先级");
    
    private int level;
    private String description;
    
    TaskPriority(int level, String description) {
        this.level = level;
        this.description = description;
    }
    
    public int getLevel() {
        return level;
    }
    
    public String getDescription() {
        return description;
    }
}

动态优先级调整

根据系统状态动态调整任务优先级：

public class PriorityAdjuster {
    private Map<String, Integer> priorityAdjustments = new ConcurrentHashMap<>();
    
    // 根据等待时间调整优先级
    public int adjustPriorityByWaitTime(TestTask task, long waitTime) {
        int basePriority = task.getPriority().getLevel();
        int adjustment = 0;
        
        if (waitTime > TimeUnit.HOURS.toMillis(1)) {
            adjustment = -2; // 提升两个优先级级别
        } else if (waitTime > TimeUnit.MINUTES.toMillis(30)) {
            adjustment = -1; // 提升一个优先级级别
        }
        
        return Math.max(1, basePriority + adjustment);
    }
    
    // 根据系统负载调整优先级
    public int adjustPriorityByLoad(TestTask task, double systemLoad) {
        int basePriority = task.getPriority().getLevel();
        int adjustment = 0;
        
        if (systemLoad < 0.3) {
            adjustment = 1; // 降低优先级
        } else if (systemLoad > 0.8) {
            adjustment = -1; // 提升优先级
        }
        
        return Math.max(1, Math.min(5, basePriority + adjustment));
    }
}

调度算法优化

通过优化调度算法提高任务执行效率。

公平调度算法

确保所有任务都能得到公平的执行机会：

public class FairScheduler {
    private Map<String, Long> taskExecutionCount = new ConcurrentHashMap<>();
    private Map<String, Long> taskWaitTime = new ConcurrentHashMap<>();
    
    public TestTask selectNextTask(List<TestTask> pendingTasks) {
        return pendingTasks.stream()
            .min(Comparator.comparing(task -> {
                long executionCount = taskExecutionCount.getOrDefault(task.getTaskId(), 0L);
                long waitTime = taskWaitTime.getOrDefault(task.getTaskId(), 0L);
                int priority = task.getPriority().getLevel();
                
                // 综合考虑优先级、执行次数和等待时间
                return priority * 1000 + executionCount * 10 - waitTime;
            }))
            .orElse(null);
    }
}

负载感知调度

根据系统负载情况调整调度策略：

public class LoadAwareScheduler {
    private SystemLoadMonitor loadMonitor;
    
    public TestTask selectNextTask(List<TestTask> pendingTasks) {
        double currentLoad = loadMonitor.getCurrentLoad();
        
        if (currentLoad > 0.8) {
            // 高负载时优先执行轻量级任务
            return pendingTasks.stream()
                .filter(task -> task.getEstimatedResourceUsage() < 0.3)
                .min(Comparator.comparing(TestTask::getPriority))
                .orElse(getHighestPriorityTask(pendingTasks));
        } else {
            // 正常负载时按优先级调度
            return getHighestPriorityTask(pendingTasks);
        }
    }
    
    private TestTask getHighestPriorityTask(List<TestTask> tasks) {
        return tasks.stream()
            .min(Comparator.comparing(task -> task.getPriority().getLevel()))
            .orElse(null);
    }
}

资源配额管理

通过资源配额管理控制不同用户或团队的资源使用。

配额定义

public class ResourceQuota {
    private String quotaId;
    private String userId;
    private int maxConcurrency;
    private int maxQueueSize;
    private long maxExecutionTime; // 毫秒
    private int maxTaskCountPerHour;
    
    // 配额检查
    public boolean checkQuota(TaskExecutionMetrics metrics) {
        return metrics.getCurrentConcurrency() < maxConcurrency &&
               metrics.getQueueSize() < maxQueueSize &&
               metrics.getHourlyTaskCount() < maxTaskCountPerHour;
    }
}

配额 enforcement

public class QuotaEnforcer {
    private Map<String, ResourceQuota> userQuotas = new ConcurrentHashMap<>();
    private Map<String, TaskExecutionMetrics> userMetrics = new ConcurrentHashMap<>();
    
    public boolean checkAndEnforceQuota(TestTask task) {
        String userId = task.getCreatedBy();
        ResourceQuota quota = userQuotas.get(userId);
        TaskExecutionMetrics metrics = userMetrics.computeIfAbsent(userId, 
            k -> new TaskExecutionMetrics());
        
        if (quota != null && !quota.checkQuota(metrics)) {
            // 配额超限，拒绝任务
            return false;
        }
        
        // 更新配额使用情况
        metrics.incrementTaskCount();
        return true;
    }
}

监控与告警

建立完善的监控和告警机制。

指标监控

监控关键调度指标：

并发数量：当前并发执行的任务数量
队列长度：等待执行的任务数量
执行延迟：任务从入队到开始执行的时间
资源利用率：系统资源的使用情况

告警机制

设置合理的告警阈值：

# 调度监控告警配置
alerts:
  - name: "high_concurrency"
    metric: "current_concurrency"
    threshold: 80
    operator: ">"
    severity: "warning"
    description: "当前并发数过高"
  
  - name: "long_queue"
    metric: "queue_length"
    threshold: 100
    operator: ">"
    severity: "warning"
    description: "任务队列过长"
  
  - name: "high_execution_delay"
    metric: "average_execution_delay"
    threshold: 300000 # 5分钟
    operator: ">"
    severity: "critical"
    description: "任务执行延迟过高"

故障处理与恢复

建立完善的故障处理和恢复机制。

任务超时处理

public class TaskTimeoutHandler {
    private ScheduledExecutorService timeoutScheduler = Executors.newScheduledThreadPool(5);
    
    public void scheduleTimeoutCheck(TestTask task, long timeout) {
        timeoutScheduler.schedule(() -> {
            if (task.getStatus() == TaskStatus.RUNNING) {
                // 任务超时，强制终止
                task.cancel();
                // 记录超时日志
                logTaskTimeout(task);
                // 发送告警
                sendTimeoutAlert(task);
            }
        }, timeout, TimeUnit.MILLISECONDS);
    }
}

死锁检测与恢复

public class DeadlockDetector {
    private Map<String, Long> taskStartTime = new ConcurrentHashMap<>();
    private ScheduledExecutorService detectorScheduler = Executors.newScheduledThreadPool(1);
    
    public void startDeadlockDetection() {
        detectorScheduler.scheduleAtFixedRate(() -> {
            long currentTime = System.currentTimeMillis();
            taskStartTime.entrySet().stream()
                .filter(entry -> currentTime - entry.getValue() > TimeUnit.HOURS.toMillis(1))
                .forEach(entry -> {
                    // 检测到可能的死锁任务
                    handlePotentialDeadlock(entry.getKey());
                });
        }, 0, 10, TimeUnit.MINUTES);
    }
    
    private void handlePotentialDeadlock(String taskId) {
        // 发送告警
        sendDeadlockAlert(taskId);
        // 尝试强制终止任务
        forceTerminateTask(taskId);
    }
}

总结

并发控制、队列机制与优先级调度是测试平台任务调度的核心组成部分。通过合理的并发控制保护系统资源，通过高效的队列机制管理任务执行顺序，通过智能的优先级调度确保重要任务及时执行，我们能够构建一个稳定、高效的测试平台调度系统。在实际应用中，我们需要根据具体的业务场景和系统架构，不断优化调度策略和实现方案，确保调度系统能够满足不断变化的业务需求。