消息队列消息积压问题详解

// RabbitMQ - 增加消费者 channel.basicQos(10); // 每次预取10条 channel.basicConsume(queue, false, new DefaultConsumer(channel) { public void handleDelivery(...) { // 处理消息 process(msg); channel.basicAck(deliveryTag, false); } }); // Kafka - 增加分区和消费者 // 消费者组自动 rebalance properties.setProperty("group.id", "fast-consumer-group");

// 消息优先级实现 public class PriorityMessage { int priority; // 1=紧急 2=普通 3=批量 long timestamp; byte[] body; } // 紧急消息走快车道 if (msg.priority == 1) { executor.submit(() -> processUrgent(msg)); } else { batchQueue.add(msg); }

// RabbitMQ - 设置消息 TTL AMQP.BasicProperties props = new AMQP.BasicProperties.Builder() .expiration("60000") // 60秒过期 .build(); // Kafka - 消息过期清理 // broker 端配置 retention.ms=604800000 // 7天 log.cleanup.policy=delete // 死信队列收集无法处理的消息 x-dead-letter-exchange="dlx.exchange" x-dead-letter-routing-key="dead.messages"

public class CircuitBreaker { private int failureThreshold = 5; private long recoveryTimeout = 30000; public Result execute(Message msg) { if (isOpen()) { // 熔断开启，消息入队等待 fallbackQueue.add(msg); return Result.CIRCUIT_OPEN; } try { return process(msg); } catch (Exception e) { recordFailure(); throw e; } } }