模拟消息队列的消费逻辑－Java

分享一个大牛的人工智能教程。零基础通俗易懂风趣幽默希望你也加入到人工智能的队伍中来请轻击人工智能教程​​​​​https://www.captainai.net/troubleshooter这是一个生产级消息队列消费逻辑模拟重点突出线程模型、可靠性保证、背压控制、顺序消费、以及如何测试验证。一、需求澄清第一步你需要先问清楚场景消费语义最多一次At-most-once、至少一次At-least-once、精确一次Exactly-once顺序要求全局有序还是分区有序消息类型普通消息、延迟消息、事务消息异常处理消费失败后重试、死信、还是跳过性能指标预期QPS、消息大小例如我假设需要实现至少一次消费语义 分区有序支持重试死信用多线程模型提升吞吐。这是最经典的中间态设计。二、核心架构设计先画图再编码[生产者] → [Queue/Broker] → [Consumer Group] ↓ [Worker Pool] (多线程拉取) ↓ [Message Processor] [Ack机制] ↓ (失败时) [Retry Queue] → [Dead Letter Queue]三、完整Java实现1. 消息模型// Message.java import java.time.Instant; import java.util.UUID; public class Message { private final String id; private final String topic; private final String key; // 用于分区 private final byte[] body; private final Instant timestamp; private int retryCount; public Message(String topic, String key, byte[] body) { this.id UUID.randomUUID().toString(); this.topic topic; this.key key; this.body body; this.timestamp Instant.now(); this.retryCount 0; } // Getters and setters public String getId() { return id; } public String getTopic() { return topic; } public String getKey() { return key; } public byte[] getBody() { return body; } public Instant getTimestamp() { return timestamp; } public int getRetryCount() { return retryCount; } public void incrementRetry() { retryCount; } }2. 消费接口定义// MessageConsumer.java FunctionalInterface public interface MessageConsumer { /** * 消费消息 * return true表示消费成功false表示失败需重试 */ boolean consume(Message message); }3. 核心消费引擎// MessageQueueConsumer.java import java.util.*; import java.util.concurrent.*; import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicLong; import java.util.concurrent.locks.ReentrantLock; import org.slf4j.Logger; import org.slf4j.LoggerFactory; /** * 模拟消息队列消费逻辑 * 特性 * - 线程池并发消费 * - 分区有序同一key的消息顺序处理 * - 失败重试 死信队列 * - 优雅关闭 */ public class MessageQueueConsumer { private static final Logger log LoggerFactory.getLogger(MessageQueueConsumer.class); private final String consumerGroup; private final MessageConsumer consumer; private final int concurrency; // 并发消费线程数 private final int maxRetries; // 最大重试次数 private final long retryDelayMs; // 重试延迟 private final int queueCapacity; // 内部队列容量 // 核心组件 private final BlockingQueueMessage messageQueue; private final ExecutorService workerPool; private final ScheduledExecutorService retryScheduler; private final AtomicBoolean running; private final ConcurrentHashMapString, ReentrantLock partitionLocks; // 死信队列 private final BlockingQueueMessage deadLetterQueue; // 监控指标 private final AtomicLong processedCount; private final AtomicLong successCount; private final AtomicLong failureCount; private final AtomicLong retryCount; public MessageQueueConsumer(String consumerGroup, MessageConsumer consumer, int concurrency, int maxRetries, long retryDelayMs) { this(consumerGroup, consumer, concurrency, maxRetries, retryDelayMs, 10000); } public MessageQueueConsumer(String consumerGroup, MessageConsumer consumer, int concurrency, int maxRetries, long retryDelayMs, int queueCapacity) { this.consumerGroup consumerGroup; this.consumer consumer; this.concurrency concurrency; this.maxRetries maxRetries; this.retryDelayMs retryDelayMs; this.queueCapacity queueCapacity; this.messageQueue new LinkedBlockingQueue(queueCapacity); this.workerPool Executors.newFixedThreadPool(concurrency); this.retryScheduler Executors.newScheduledThreadPool(concurrency / 2); this.running new AtomicBoolean(true); this.partitionLocks new ConcurrentHashMap(); this.deadLetterQueue new LinkedBlockingQueue(); this.processedCount new AtomicLong(0); this.successCount new AtomicLong(0); this.failureCount new AtomicLong(0); this.retryCount new AtomicLong(0); startWorkers(); } /** * 提交消息到消费队列 */ public void submit(Message message) { try { messageQueue.put(message); log.debug(Message submitted to queue: {}, message.getId()); } catch (InterruptedException e) { Thread.currentThread().interrupt(); log.error(Failed to submit message: {}, message.getId(), e); } } /** * 批量提交消息 */ public void submitAll(ListMessage messages) { for (Message msg : messages) { submit(msg); } } /** * 启动工作线程 */ private void startWorkers() { for (int i 0; i concurrency; i) { workerPool.submit(this::consumeLoop); } log.info(Consumer started with {} workers, concurrency); } /** * 主消费循环 */ private void consumeLoop() { while (running.get()) { try { Message message messageQueue.poll(100, TimeUnit.MILLISECONDS); if (message null) { continue; } // 保证分区有序 processWithPartitionOrder(message); } catch (InterruptedException e) { Thread.currentThread().interrupt(); break; } } } /** * 保证同一key的消息顺序处理 */ private void processWithPartitionOrder(Message message) { String lockKey message.getKey() ! null ? message.getKey() : message.getTopic(); ReentrantLock lock partitionLocks.computeIfAbsent(lockKey, k - new ReentrantLock()); lock.lock(); try { processMessage(message); } finally { lock.unlock(); } } /** * 实际消费逻辑带重试 */ private void processMessage(Message message) { processedCount.incrementAndGet(); try { boolean success consumer.consume(message); if (success) { successCount.incrementAndGet(); log.debug(Message consumed successfully: {}, message.getId()); } else { handleFailure(message); } } catch (Exception e) { log.error(Consumer threw exception for message: {}, message.getId(), e); handleFailure(message); } } /** * 处理消费失败 */ private void handleFailure(Message message) { failureCount.incrementAndGet(); if (message.getRetryCount() maxRetries) { // 需要重试 message.incrementRetry(); retryCount.incrementAndGet(); long delay calculateBackoffDelay(message.getRetryCount()); log.warn(Message {} failed (retry {}/{}), retrying in {}ms, message.getId(), message.getRetryCount(), maxRetries, delay); retryScheduler.schedule(() - { if (running.get()) { submit(message); } }, delay, TimeUnit.MILLISECONDS); } else { // 超过最大重试次数进死信队列 log.error(Message {} exceeded max retries, moving to DLQ, message.getId()); deadLetterQueue.offer(message); } } /** * 指数退避 抖动 */ private long calculateBackoffDelay(int retryCount) { long baseDelay retryDelayMs * (long) Math.pow(2, retryCount - 1); long jitter (long) (baseDelay * 0.1 * Math.random()); return Math.min(baseDelay jitter, 30000); // 最大30秒 } /** * 优雅关闭 */ public void shutdown() { log.info(Shutting down consumer...); running.set(false); workerPool.shutdown(); retryScheduler.shutdown(); try { if (!workerPool.awaitTermination(30, TimeUnit.SECONDS)) { workerPool.shutdownNow(); } if (!retryScheduler.awaitTermination(30, TimeUnit.SECONDS)) { retryScheduler.shutdownNow(); } } catch (InterruptedException e) { workerPool.shutdownNow(); retryScheduler.shutdownNow(); Thread.currentThread().interrupt(); } log.info(Consumer shutdown complete. Stats: processed{}, success{}, failure{}, retry{}, dlq{}, processedCount.get(), successCount.get(), failureCount.get(), retryCount.get(), deadLetterQueue.size()); } /** * 获取死信队列中的消息用于人工处理 */ public ListMessage getDeadLetters() { ListMessage deadLetters new ArrayList(); deadLetterQueue.drainTo(deadLetters); return deadLetters; } /** * 获取消费统计信息 */ public ConsumerStats getStats() { return new ConsumerStats( processedCount.get(), successCount.get(), failureCount.get(), retryCount.get(), deadLetterQueue.size(), messageQueue.size() ); } public static class ConsumerStats { private final long processed; private final long success; private final long failure; private final long retry; private final int deadLetterSize; private final int pendingSize; public ConsumerStats(long processed, long success, long failure, long retry, int deadLetterSize, int pendingSize) { this.processed processed; this.success success; this.failure failure; this.retry retry; this.deadLetterSize deadLetterSize; this.pendingSize pendingSize; } Override public String toString() { return String.format(Stats{processed%d, success%d, failure%d, retry%d, dlq%d, pending%d}, processed, success, failure, retry, deadLetterSize, pendingSize); } } }4. 背压控制版本高级特性// BackpressuredConsumer.java - 简化版 /** * 支持背压的消费者当下游处理不过来时自动限流 */ public class BackpressuredConsumer extends MessageQueueConsumer { private final Semaphore backpressureSemaphore; private final int maxInflight; public BackpressuredConsumer(String consumerGroup, MessageConsumer consumer, int concurrency, int maxRetries, long retryDelayMs, int maxInflight) { super(consumerGroup, consumer, concurrency, maxRetries, retryDelayMs); this.maxInflight maxInflight; this.backpressureSemaphore new Semaphore(maxInflight); } Override private void processMessage(Message message) { try { backpressureSemaphore.acquire(); try { super.processMessage(message); } finally { backpressureSemaphore.release(); } } catch (InterruptedException e) { Thread.currentThread().interrupt(); } } }四、测试策略重点1. 基础功能测试Test void testBasicConsumption() throws InterruptedException { CountDownLatch latch new CountDownLatch(10); MessageConsumer mockConsumer msg - { latch.countDown(); return true; }; MessageQueueConsumer consumer new MessageQueueConsumer( test-group, mockConsumer, 2, 3, 100 ); for (int i 0; i 10; i) { consumer.submit(new Message(topic, key, (msg i).getBytes())); } assertTrue(latch.await(5, TimeUnit.SECONDS)); ConsumerStats stats consumer.getStats(); assertEquals(10, stats.getProcessed()); assertEquals(10, stats.getSuccess()); consumer.shutdown(); }2. 重试与死信测试Test void testRetryAndDeadLetter() throws InterruptedException { AtomicInteger attemptCount new AtomicInteger(0); MessageConsumer failingConsumer msg - { int attempt attemptCount.incrementAndGet(); return attempt 2; // 第3次成功 }; MessageQueueConsumer consumer new MessageQueueConsumer( test-group, failingConsumer, 1, 2, 10 // 最多重试2次 ); consumer.submit(new Message(topic, key, test.getBytes())); Thread.sleep(500); // 等待重试完成 ConsumerStats stats consumer.getStats(); assertEquals(1, stats.getProcessed()); // 只有1条消息 assertEquals(0, stats.getSuccess()); // 最终失败 assertEquals(1, stats.getDeadLetterSize()); // 进死信队列 assertEquals(2, stats.getRetry()); // 重试了2次 consumer.shutdown(); }3. 分区有序性测试关键Test void testPartitionOrdering() throws InterruptedException { ListInteger consumedOrder Collections.synchronizedList(new ArrayList()); MessageConsumer consumer msg - { int seq Integer.parseInt(new String(msg.getBody())); consumedOrder.add(seq); return true; }; MessageQueueConsumer queueConsumer new MessageQueueConsumer( test-group, consumer, 4, 0, 0 ); // 向两个不同key发送乱序消息 // key1: 1,2,3,4 // key2: 100,200,300 for (int i 1; i 4; i) { queueConsumer.submit(new Message(topic, key1, String.valueOf(i).getBytes())); } for (int i 100; i 300; i 100) { queueConsumer.submit(new Message(topic, key2, String.valueOf(i).getBytes())); } Thread.sleep(1000); // 验证每个分区内部的顺序 ListInteger key1Order consumedOrder.stream() .filter(n - n 4) .collect(Collectors.toList()); ListInteger key2Order consumedOrder.stream() .filter(n - n 100) .collect(Collectors.toList()); assertEquals(Arrays.asList(1,2,3,4), key1Order); assertEquals(Arrays.asList(100,200,300), key2Order); queueConsumer.shutdown(); }4. 背压与限流测试Test void testBackpressure() throws InterruptedException { AtomicInteger processingFlag new AtomicInteger(0); CountDownLatch processingStarted new CountDownLatch(1); MessageConsumer slowConsumer msg - { processingStarted.countDown(); processingFlag.incrementAndGet(); Thread.sleep(2000); // 模拟慢消费 return true; }; BackpressuredConsumer consumer new BackpressuredConsumer( test-group, slowConsumer, 1, 0, 0, 2 // 最多2个inflight ); // 快速提交100条消息 for (int i 0; i 100; i) { consumer.submit(new Message(topic, key, (msg i).getBytes())); } processingStarted.await(); // 验证内部队列已满背压生效 ConsumerStats stats consumer.getStats(); assertTrue(stats.getPendingSize() 100); // 队列应该有积压 consumer.shutdown(); }5. 并发安全性测试Test void testConcurrentSubmit() throws InterruptedException { MessageQueueConsumer consumer new MessageQueueConsumer( test-group, msg - true, 10, 0, 0 ); int threadCount 50; int messagesPerThread 1000; ExecutorService executor Executors.newFixedThreadPool(threadCount); CountDownLatch latch new CountDownLatch(threadCount); for (int t 0; t threadCount; t) { executor.submit(() - { for (int i 0; i messagesPerThread; i) { consumer.submit(new Message(topic, key i % 100, (msg i).getBytes())); } latch.countDown(); }); } latch.await(30, TimeUnit.SECONDS); Thread.sleep(2000); // 等待消费完成 ConsumerStats stats consumer.getStats(); assertEquals(threadCount * messagesPerThread, stats.getProcessed()); executor.shutdown(); consumer.shutdown(); }6. 性能基准测试Test void benchmarkThroughput() { MessageQueueConsumer consumer new MessageQueueConsumer( benchmark, msg - true, 8, 0, 0, 100000 ); int messageCount 500_000; long start System.nanoTime(); for (int i 0; i messageCount; i) { consumer.submit(new Message(topic, key (i % 100), (msg i).getBytes())); } // 等待消费完成 while (consumer.getStats().getProcessed() messageCount) { Thread.sleep(100); } long duration System.nanoTime() - start; double throughput messageCount * 1000.0 / (duration / 1_000_000); System.out.printf(Throughput: %.2f msg/sec%n, throughput); consumer.shutdown(); }五、进阶问题Q1如何保证Exactly-Once语义回答At-least-once 幂等消费。可以在Message中加入全局唯一ID下游系统用Redis或DB去重表记录已处理的ID。实现支持扩展示例如public class ExactlyOnceConsumer extends MessageQueueConsumer { private final SetString processedIds ConcurrentHashMap.newKeySet(); Override protected boolean tryConsume(Message message) { if (processedIds.contains(message.getId())) { return true; // 已处理过直接返回成功 } boolean success super.tryConsume(message); if (success) { processedIds.add(message.getId()); } return success; } }Q2消息积压时如何处理回答动态扩缩容监控队列长度超过阈值时自动增加worker线程数优先级队列高优消息优先处理流量控制发送端限流避免消费端被压垮分片扩展增加更多的分区通过水平扩展消费实例Q3如何测试消息不会丢失回答我会设计故障注入测试在消费过程中随机kill consumer线程模拟网络中断、DB连接超时验证重试机制是否正确触发消息最终是否被处理使用Chaos Monkey框架Q4如何评估这个消费系统的容量回答采用容量规划三步法单线程压测找出一条消息的平均处理时间线性推算需要多少并发达到目标QPS实际峰值模拟用生产流量回放找到拐点同时设置水位告警当队列深度 阈值或消费延迟 5秒时报警六、总结这个消费模型具备生产级可靠性分区有序保证业务正确性、重试死信处理异常、背压机制防止下游被打垮。同时提供了一套完整的测试策略覆盖功能、并发、性能和混沌场景。更重要的是这个设计是可观测的——暴露了关键的metrics方便监控和容量规划。