消息队列中的"吵闹邻居问题"：识别、影响与解决方案

// 解决方案的五大支柱
public class NoisyNeighborSolution {
    
    // 1. 流量控制 - Rate Limiting
    private RateLimiter rateLimiter;
    
    // 2. 公平调度 - Fair Dispatch
    private FairScheduler scheduler;
    
    // 3. 资源隔离 - Resource Isolation
    private ResourcePool resourcePool;
    
    // 4. 优先级管理 - Priority Management
    private PriorityQueue priorityQueue;
    
    // 5. 监控预警 - Monitoring & Alerting
    private MetricsCollector metrics;
}

import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * 令牌桶速率限制器 - 通用实现
 */
public class TokenBucketRateLimiter {
    
    private final int capacity;           // 桶容量
    private final int refillRate;         // 每秒补充令牌数
    private final AtomicInteger tokens;   // 当前令牌数
    private final ScheduledExecutorService scheduler;
    
    public TokenBucketRateLimiter(int capacity, int refillRatePerSecond) {
        this.capacity = capacity;
        this.refillRate = refillRatePerSecond;
        this.tokens = new AtomicInteger(capacity);
        this.scheduler = Executors.newScheduledThreadPool(1);
        
        // 定时补充令牌
        scheduler.scheduleAtFixedRate(this::refill, 0, 1, TimeUnit.SECONDS);
    }
    
    /**
     * 尝试获取令牌
     */
    public boolean tryAcquire() {
        return tryAcquire(1);
    }
    
    /**
     * 尝试获取多个令牌
     */
    public boolean tryAcquire(int permits) {
        int current = tokens.get();
        if (current < permits) {
            return false;
        }
        return tokens.compareAndSet(current, current - permits);
    }
    
    /**
     * 阻塞式获取令牌
     */
    public void acquire() throws InterruptedException {
        while (!tryAcquire()) {
            Thread.sleep(100);  // 等待100ms后重试
        }
    }
    
    /**
     * 补充令牌
     */
    private void refill() {
        tokens.updateAndGet(current -> 
            Math.min(capacity, current + refillRate));
    }
}

import java.util.concurrent.*;
import java.util.Map;

/**
 * 公平消息调度器
 */
public class FairMessageScheduler {
    
    // 每个队列/租户的配额
    private final Map<String, Integer> quotas = new ConcurrentHashMap<>();
    
    // 当前周期内的使用量
    private final Map<String, AtomicInteger> usage = new ConcurrentHashMap<>();
    
    // 预取配置
    private final int defaultPrefetch = 10;
    private final int minPrefetch = 1;
    private final int maxPrefetch = 100;
    
    /**
     * 计算动态预取数量
     */
    public int calculatePrefetchCount(String queueName) {
        // 获取配额
        int quota = quotas.getOrDefault(queueName, defaultPrefetch);
        
        // 获取当前使用量
        int currentUsage = usage.computeIfAbsent(queueName, 
            k -> new AtomicInteger(0)).get();
        
        // 动态调整：使用量越大，预取越少（实现公平）
        double usageRatio = (double) currentUsage / quota;
        int prefetch = (int) (defaultPrefetch * (1 - usageRatio * 0.8));
        
        // 确保在合理范围内
        return Math.max(minPrefetch, Math.min(maxPrefetch, prefetch));
    }
    
    /**
     * 记录消息消费
     */
    public void recordConsumption(String queueName, int count) {
        usage.computeIfAbsent(queueName, k -> new AtomicInteger(0))
             .addAndGet(count);
    }
    
    /**
     * 重置使用统计（每个周期调用）
     */
    public void resetUsage() {
        usage.clear();
    }
    
    /**
     * 设置队列配额
     */
    public void setQuota(String queueName, int quota) {
        quotas.put(queueName, quota);
    }
}

import java.util.concurrent.*;
import java.util.Comparator;

/**
 * 优先级消息处理器
 */
public class PriorityMessageProcessor {
    
    // 优先级队列
    private final PriorityBlockingQueue<PriorityMessage> queue;
    
    // 工作线程池
    private final ExecutorService executorService;
    
    // 处理超时配置
    private final long timeoutMs;
    
    public PriorityMessageProcessor(int poolSize, long timeoutMs) {
        this.queue = new PriorityBlockingQueue<>(1000, 
            Comparator.comparing(PriorityMessage::getPriority).reversed()
                      .thenComparing(PriorityMessage::getTimestamp));
        
        this.executorService = new ThreadPoolExecutor(
            poolSize / 2,  // 核心线程数
            poolSize,       // 最大线程数
            60L, TimeUnit.SECONDS,
            new LinkedBlockingQueue<>(1000),
            new ThreadPoolExecutor.CallerRunsPolicy()
        );
        
        this.timeoutMs = timeoutMs;
        
        // 启动消费线程
        startConsumers(poolSize);
    }
    
    /**
     * 提交消息
     */
    public boolean submit(String messageId, Object payload, int priority) {
        PriorityMessage message = new PriorityMessage(
            messageId, payload, priority, System.currentTimeMillis()
        );
        
        // 检查消息是否过期
        if (isExpired(message)) {
            handleExpiredMessage(message);
            return false;
        }
        
        return queue.offer(message);
    }
    
    /**
     * 启动消费者
     */
    private void startConsumers(int count) {
        for (int i = 0; i < count; i++) {
            executorService.submit(this::consume);
        }
    }
    
    /**
     * 消费逻辑
     */
    private void consume() {
        while (!Thread.currentThread().isInterrupted()) {
            try {
                PriorityMessage message = queue.poll(1, TimeUnit.SECONDS);
                if (message != null) {
                    processWithTimeout(message);
                }
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                break;
            }
        }
    }
    
    /**
     * 带超时的处理
     */
    private void processWithTimeout(PriorityMessage message) {
        Future<?> future = executorService.submit(() -> 
            processMessage(message)
        );
        
        try {
            future.get(timeoutMs, TimeUnit.MILLISECONDS);
        } catch (TimeoutException e) {
            future.cancel(true);
            handleTimeout(message);
        } catch (Exception e) {
            handleError(message, e);
        }
    }
    
    // 消息类定义
    static class PriorityMessage {
        private final String id;
        private final Object payload;
        private final int priority;
        private final long timestamp;
        
        // 构造函数和getter省略...
    }
    
    // 抽象方法，子类实现
    protected abstract void processMessage(PriorityMessage message);
    protected abstract void handleTimeout(PriorityMessage message);
    protected abstract void handleError(PriorityMessage message, Exception e);
    protected abstract void handleExpiredMessage(PriorityMessage message);
    protected abstract boolean isExpired(PriorityMessage message);
}

import java.util.concurrent.*;
import java.util.Map;

/**
 * 资源隔离池 - 为不同租户/队列分配独立资源
 */
public class IsolatedResourcePool {
    
    // 每个租户的独立线程池
    private final Map<String, ExecutorService> tenantPools;
    
    // 共享池（用于低优先级任务）
    private final ExecutorService sharedPool;
    
    // 资源配置
    private final Map<String, ResourceConfig> configs;
    
    public IsolatedResourcePool() {
        this.tenantPools = new ConcurrentHashMap<>();
        this.configs = new ConcurrentHashMap<>();
        
        // 创建共享池
        this.sharedPool = new ForkJoinPool(
            Runtime.getRuntime().availableProcessors(),
            ForkJoinPool.defaultForkJoinWorkerThreadFactory,
            null, true
        );
    }
    
    /**
     * 为租户分配资源
     */
    public void allocateResources(String tenant, ResourceConfig config) {
        configs.put(tenant, config);
        
        // 创建独立线程池
        ExecutorService pool = new ThreadPoolExecutor(
            config.minThreads,
            config.maxThreads,
            60L, TimeUnit.SECONDS,
            new ArrayBlockingQueue<>(config.queueSize),
            new ThreadFactory() {
                private final AtomicInteger counter = new AtomicInteger(0);
                @Override
                public Thread newThread(Runnable r) {
                    Thread t = new Thread(r);
                    t.setName("Pool-" + tenant + "-" + counter.getAndIncrement());
                    t.setPriority(config.threadPriority);
                    return t;
                }
            },
            config.rejectionPolicy
        );
        
        tenantPools.put(tenant, pool);
    }
    
    /**
     * 提交任务到对应的资源池
     */
    public Future<?> submit(String tenant, Runnable task) {
        ExecutorService pool = tenantPools.get(tenant);
        
        if (pool == null) {
            // 没有专属池，使用共享池
            return sharedPool.submit(task);
        }
        
        return pool.submit(task);
    }
    
    /**
     * 资源配置
     */
    static class ResourceConfig {
        int minThreads = 1;
        int maxThreads = 10;
        int queueSize = 100;
        int threadPriority = Thread.NORM_PRIORITY;
        RejectedExecutionHandler rejectionPolicy = 
            new ThreadPoolExecutor.CallerRunsPolicy();
        
        // Builder模式构造，省略...
    }
}

import java.util.concurrent.*;
import java.util.concurrent.atomic.*;

/**
 * 自适应监控器 - 根据运行指标自动调整参数
 */
public class AdaptiveMonitor {
    
    // 性能指标
    private final Map<String, QueueMetrics> metrics;
    
    // 调整策略
    private final AdaptiveStrategy strategy;
    
    // 监控周期
    private final ScheduledExecutorService scheduler;
    
    public AdaptiveMonitor(AdaptiveStrategy strategy) {
        this.metrics = new ConcurrentHashMap<>();
        this.strategy = strategy;
        this.scheduler = Executors.newScheduledThreadPool(1);
        
        // 每30秒执行一次自适应调整
        scheduler.scheduleAtFixedRate(
            this::performAdaptation, 30, 30, TimeUnit.SECONDS
        );
    }
    
    /**
     * 记录处理指标
     */
    public void recordProcessing(String queue, long duration, boolean success) {
        QueueMetrics m = metrics.computeIfAbsent(queue, 
            k -> new QueueMetrics(queue));
        
        m.recordProcessing(duration, success);
    }
    
    /**
     * 执行自适应调整
     */
    private void performAdaptation() {
        metrics.forEach((queue, metric) -> {
            AdaptiveAction action = strategy.analyze(metric);
            applyAction(queue, action);
            metric.reset();  // 重置统计
        });
    }
    
    /**
     * 应用调整动作
     */
    private void applyAction(String queue, AdaptiveAction action) {
        switch (action.type) {
            case INCREASE_CONCURRENCY:
                increaseConcurrency(queue, action.value);
                break;
            case DECREASE_CONCURRENCY:
                decreaseConcurrency(queue, action.value);
                break;
            case ADJUST_RATE_LIMIT:
                adjustRateLimit(queue, action.value);
                break;
            case ENABLE_CIRCUIT_BREAKER:
                enableCircuitBreaker(queue);
                break;
            default:
                // No action needed
        }
    }
    
    /**
     * 队列指标
     */
    static class QueueMetrics {
        private final String queue;
        private final AtomicLong totalProcessed = new AtomicLong(0);
        private final AtomicLong totalFailed = new AtomicLong(0);
        private final AtomicLong totalDuration = new AtomicLong(0);
        private final AtomicLong maxDuration = new AtomicLong(0);
        
        public QueueMetrics(String queue) {
            this.queue = queue;
        }
        
        public void recordProcessing(long duration, boolean success) {
            if (success) {
                totalProcessed.incrementAndGet();
            } else {
                totalFailed.incrementAndGet();
            }
            totalDuration.addAndGet(duration);
            
            // 更新最大处理时间
            long currentMax = maxDuration.get();
            while (duration > currentMax) {
                if (maxDuration.compareAndSet(currentMax, duration)) {
                    break;
                }
                currentMax = maxDuration.get();
            }
        }
        
        public double getSuccessRate() {
            long total = totalProcessed.get() + totalFailed.get();
            return total > 0 ? (double) totalProcessed.get() / total : 1.0;
        }
        
        public double getAverageDuration() {
            long count = totalProcessed.get();
            return count > 0 ? (double) totalDuration.get() / count : 0;
        }
        
        public void reset() {
            totalProcessed.set(0);
            totalFailed.set(0);
            totalDuration.set(0);
            maxDuration.set(0);
        }
    }
    
    /**
     * 自适应策略接口
     */
    interface AdaptiveStrategy {
        AdaptiveAction analyze(QueueMetrics metrics);
    }
    
    /**
     * 自适应动作
     */
    static class AdaptiveAction {
        enum Type {
            NO_ACTION,
            INCREASE_CONCURRENCY,
            DECREASE_CONCURRENCY,
            ADJUST_RATE_LIMIT,
            ENABLE_CIRCUIT_BREAKER
        }
        
        final Type type;
        final int value;
        
        AdaptiveAction(Type type, int value) {
            this.type = type;
            this.value = value;
        }
    }
    
    // 抽象方法，子类实现
    protected abstract void increaseConcurrency(String queue, int delta);
    protected abstract void decreaseConcurrency(String queue, int delta);
    protected abstract void adjustRateLimit(String queue, int newLimit);
    protected abstract void enableCircuitBreaker(String queue);
}

@Configuration
@EnableRabbit
public class RabbitMQNoisyNeighborConfig {
    
    @Value("${mq.noisy-neighbor.enabled:false}")
    private boolean enableProtection;
    
    @Value("${mq.rate-limit:100}")
    private int rateLimit;
    
    @Value("${mq.prefetch:10}")
    private int prefetchCount;
    
    @Bean
    public SimpleRabbitListenerContainerFactory rabbitListenerContainerFactory(
            ConnectionFactory connectionFactory) {
        
        SimpleRabbitListenerContainerFactory factory = 
            new SimpleRabbitListenerContainerFactory();
        factory.setConnectionFactory(connectionFactory);
        
        if (enableProtection) {
            // 启用防噪保护
            factory.setPrefetchCount(prefetchCount);
            factory.setConcurrentConsumers(3);
            factory.setMaxConcurrentConsumers(10);
            factory.setConsumerBatchEnabled(false);
            
            // 设置消息拦截器
            factory.setAdviceChain(
                rateLimitAdvice(),
                retryAdvice(),
                circuitBreakerAdvice()
            );
        }
        
        return factory;
    }
    
    @Bean
    public MethodInterceptor rateLimitAdvice() {
        return invocation -> {
            // 速率限制逻辑
            if (!rateLimiter.tryAcquire()) {
                throw new RateLimitException("Rate limit exceeded");
            }
            return invocation.proceed();
        };
    }
}

@Configuration
@EnableKafka
public class KafkaNoisyNeighborConfig {
    
    @Bean
    public ConcurrentKafkaListenerContainerFactory<String, Object> 
            kafkaListenerContainerFactory() {
        
        ConcurrentKafkaListenerContainerFactory<String, Object> factory = 
            new ConcurrentKafkaListenerContainerFactory<>();
        
        factory.setConsumerFactory(consumerFactory());
        
        // 配置防噪参数
        factory.setConcurrency(3);  // 并发消费者数
        factory.getContainerProperties().setPollTimeout(3000);
        factory.getContainerProperties().setIdleBetweenPolls(1000);
        
        // 设置批量消费
        factory.setBatchListener(true);
        factory.getContainerProperties().setMaxPollRecordsConfig(100);
        
        // 设置错误处理器
        factory.setCommonErrorHandler(new DefaultErrorHandler(
            new FixedBackOff(1000L, 3L)
        ));
        
        return factory;
    }
}

public class PerformanceBenchmark {
    
    public static void main(String[] args) {
        // 测试场景配置
        BenchmarkConfig config = new BenchmarkConfig()
            .setProducerCount(10)      // 生产者数量
            .setConsumerCount(5)       // 消费者数量
            .setMessageCount(100000)   // 消息总量
            .setMessageSize(1024)      // 消息大小(bytes)
            .setNoisyRatio(0.2);       // 噪声比例(20%的大消息)
        
        // 运行测试
        BenchmarkResult result = runBenchmark(config);
        
        // 输出结果
        System.out.println("=== Performance Benchmark Results ===");
        System.out.println("Throughput: " + result.throughput + " msg/s");
        System.out.println("P50 Latency: " + result.p50Latency + " ms");
        System.out.println("P99 Latency: " + result.p99Latency + " ms");
        System.out.println("Success Rate: " + result.successRate + "%");
        System.out.println("Resource Usage: CPU=" + result.cpuUsage + 
                         "%, Memory=" + result.memoryUsage + "MB");
    }
}