1
# Kafka：查看 Consumer Lag
2
kafka-consumer-groups --bootstrap-server localhost:9092 \
3
    --describe --group order-consumer-group
4

5
# 输出：
6
# TOPIC   PARTITION  CURRENT-OFFSET  LOG-END-OFFSET  LAG
7
# orders  0          12345           56789           44444
8
# orders  1          23456           67890           44434
9

10
# RabbitMQ：查看队列深度
11
rabbitmqctl list_queues name messages consumers message_rates
12

13
# RocketMQ：查看消费进度
14
./mqadmin consumerProgress -g order-consumer-group

1
# Kafka：增加分区（注意：会影响 Key 路由）
2
kafka-topics --bootstrap-server localhost:9092 \
3
    --alter --topic orders --partitions 16
4

5
# RocketMQ：增加队列
6
./mqadmin updateTopic -n localhost:9876 -t TopicOrder -c DefaultCluster -r 16

1
// Kafka：多线程消费优化
2
KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
3
consumer.subscribe(List.of("orders"));
4

5
ExecutorService executor = Executors.newFixedThreadPool(8);
6

7
while (true) {
8
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
9
    // 按 Partition 分组，每个 Partition 内保持有序
10
    Map<TopicPartition, List<ConsumerRecord<String, String>>> partitioned =
11
        records.partitions().stream()
12
            .collect(Collectors.toMap(
13
                tp -> tp,
14
                tp -> records.records(tp)
15
            ));
16

17
    List<Future<?>> futures = new ArrayList<>();
18
    for (var entry : partitioned.entrySet()) {
19
        futures.add(executor.submit(() -> {
20
            for (var record : entry.getValue()) {
21
                processRecord(record);
22
            }
23
        }));
24
    }
25

26
    // 等待所有线程完成
27
    for (Future<?> f : futures) f.get();
28
    consumer.commitSync();
29
}

1
// 策略一：只消费最新消息（跳过积压）
2
props.put("auto.offset.reset", "latest");  // 从最新开始消费
3

4
// 策略二：重置 Offset 到最近位置
5
kafka-consumer-groups --bootstrap-server localhost:9092 \
6
    --group order-consumer-group \
7
    --reset-offsets --to-latest --execute --topic orders
8

9
// 策略三：按时间重置（跳过超过保留期的消息）
10
kafka-consumer-groups --bootstrap-server localhost:9092 \
11
    --group order-consumer-group \
12
    --reset-offsets --to-datetime 2026-10-12T00:00:00 --execute --topic orders

1
// Kafka：Consumer 端流控
2
// 通过控制 poll 频率和 max.poll.records 实现反压
3
props.put("max.poll.records", 100);        // 每次 poll 最多 100 条
4
props.put("max.poll.interval.ms", 300000); // 5 分钟内必须处理完
5

6
// 如果处理慢，自动降低 poll 频率
7
while (true) {
8
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
9
    if (records.count() > 0) {
10
        processRecords(records);
11
        // 处理慢 → 下次 poll 延迟
12
    }
13
}

1
// RabbitMQ：Consumer 端流控
2
channel.basicQos(100);  // 预取 100 条
3

4
channel.basicConsume("order-queue", false, new DefaultConsumer(channel) {
5
    @Override
6
    public void handleDelivery(String tag, Envelope envelope,
7
                               AMQP.BasicProperties props, byte[] body) {
8
        processMessage(body);
9
        // 手动确认后，RabbitMQ 才会投递新消息
10
        channel.basicAck(envelope.getDeliveryTag(), false);
11
    }
12
});
13

14
// 如果处理慢，不 ACK → RabbitMQ 不再投递新消息 → 自动反压

1
// Pulsar：Consumer 端流控
2
Consumer<byte[]> consumer = client.newConsumer()
3
    .topic("orders")
4
    .subscriptionName("order-processor")
5
    .receiverQueueSize(100)  // 接收队列大小
6
    .subscribe();
7

8
// Pulsar 使用 permit 机制：Consumer 消费一条消息后发放一个 permit
9
// Broker 只在有 permit 时才投递消息

1
// Kafka Producer：缓冲区满时阻塞
2
props.put("buffer.memory", 67108864);     // 64MB 缓冲区
3
props.put("max.block.ms", 60000);         // 缓冲区满时最多阻塞 60 秒
4

5
try {
6
    producer.send(record, (metadata, exception) -> {
7
        if (exception != null) {
8
            log.error("Send failed", exception);
9
        }
10
    });
11
} catch (BufferExhaustedException e) {
12
    // 缓冲区满，需要降速或降级
13
    log.warn("Producer buffer full, applying backpressure");
14
}

1
// Kafka：手动实现死信队列
2
while (true) {
3
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
4
    for (var record : records) {
5
        int retryCount = getRetryCount(record);
6
        try {
7
            processRecord(record);
8
        } catch (Exception e) {
9
            if (retryCount >= MAX_RETRIES) {
10
                // 发送到死信 Topic
11
                ProducerRecord<String, String> dlqRecord = new ProducerRecord<>(
12
                    "orders.DLQ",
13
                    record.key(),
14
                    record.value()
15
                );
16
                dlqProducer.send(dlqRecord);
17
                log.error("Message sent to DLQ after {} retries: {}", retryCount, record.value());
18
            } else {
19
                // 发送到重试 Topic
20
                sendToRetryTopic(record, retryCount + 1);
21
            }
22
        }
23
    }
24
    consumer.commitSync();
25
}

1
# RabbitMQ：查看死信队列
2
rabbitmqctl list_queues name messages arguments | grep dlx
3

4
# RocketMQ：查看死信队列
5
./mqadmin examineTopicStats -t %DLQ%order-consumer-group
6

7
# Kafka：消费死信 Topic
8
kafka-console-consumer --bootstrap-server localhost:9092 --topic orders.DLQ

1
// 临时消费者：快速消费积压消息，不做业务处理，只转储
2
public class EmergencyConsumer {
3
    private final KafkaConsumer<String, String> consumer;
4
    private final Producer<File> fileProducer;  // 写入本地文件
5

6
    public void startEmergencyConsume() {
7
        while (true) {
8
            ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
9
            for (var record : records) {
10
                // 转储到文件，后续回填
11
                fileProducer.write(record.key(), record.value(), record.timestamp());
12
            }
13
            consumer.commitSync();
14

15
            if (getLag() < THRESHOLD) {
16
                log.info("Lag recovered, stopping emergency consumer");
17
                break;
18
            }
19
        }
20
    }
21
}

1
// 按消息优先级消费
2
public class PriorityConsumer {
3
    public void consumeWithPriority() {
4
        // 先消费高优先级 Topic
5
        consumeTopic("orders-high-priority");
6
        // 再消费普通 Topic
7
        consumeTopic("orders-normal");
8
        // 最后消费低优先级 Topic
9
        consumeTopic("orders-low-priority");
10
    }
11

12
    private void consumeTopic(String topic) {
13
        // 消费到 Lag < 阈值后切换
14
        while (getLag(topic) > LAG_THRESHOLD) {
15
            var records = consumer.poll(Duration.ofMillis(100));
16
            processRecords(records);
17
        }
18
    }
19
}

1
# Kubernetes HPA：基于 Consumer Lag 自动伸缩
2
apiVersion: autoscaling/v2
3
kind: HorizontalPodAutoscaler
4
metadata:
5
  name: order-consumer-hpa
6
spec:
7
  scaleTargetRef:
8
    apiVersion: apps/v1
9
    kind: Deployment
10
    name: order-consumer
11
  minReplicas: 2
12
  maxReplicas: 20
13
  metrics:
14
    - type: External
15
      external:
16
        metric:
17
          name: kafka_consumer_lag
18
        target:
19
          type: AverageValue
20
          averageValue: "10000"

1
// Producer 端限流
2
public class RateLimitedProducer {
3
    private final RateLimiter rateLimiter = RateLimiter.create(5000); // 5000 msg/s
4

5
    public void send(String topic, String key, byte[] value) {
6
        if (!rateLimiter.tryAcquire(100, TimeUnit.MILLISECONDS)) {
7
            // 限流：丢弃或缓冲
8
            log.warn("Rate limited, message buffered");
9
            bufferMessage(topic, key, value);
10
            return;
11
        }
12
        producer.send(new ProducerRecord<>(topic, key, value));
13
    }
14
}