高并发分布式消息系统架构设计与实战

public class BusinessPartitioner implements Partitioner {
    
    @Override
    public int partition(String topic, Object key, byte[] keyBytes,
                        Object value, byte[] valueBytes, Cluster cluster) {
        List<PartitionInfo> partitions = cluster.partitionsForTopic(topic);
        int numPartitions = partitions.size();
        
        // 1. 顺序消息：按业务Key哈希到固定分区
        if (keyBytes != null && isSequentialTopic(topic)) {
            return Math.abs(Utils.murmur2(keyBytes)) % numPartitions;
        }
        
        // 2. 广播消息：轮询所有分区
        if (isBroadcastTopic(topic)) {
            return roundRobinPartition(topic, numPartitions);
        }
        
        // 3. 优先级消息：VIP用户消息路由到高性能分区
        if (isPriorityMessage(value)) {
            return getPriorityPartition(numPartitions);
        }
        
        // 4. 默认：基于粘性分区，提高批量发送效率
        return stickyPartition(cluster, topic);
    }
    
    private boolean isSequentialTopic(String topic) {
        return topic.endsWith("-seq") || topic.contains("order");
    }
}

Properties props = new Properties();
// 消费者组ID
props.put(ConsumerConfig.GROUP_ID_CONFIG, "order-payment-group");

// 消费位点重置策略：earliest/latest/none
props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");

// 手动提交偏移量，确保消息处理完成后再提交
props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");

// 批量消费提升吞吐量
props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 500);
props.put(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG, 300000);

// 心跳机制防止Rebalance风暴
props.put(ConsumerConfig.HEARTBEAT_INTERVAL_MS_CONFIG, 3000);
props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, 10000);

// 消费线程数配置
props.put(ConsumerConfig.MAX_PARTITION_FETCH_BYTES_CONFIG, 5242880); // 5MB

@Component
public class IdempotentConsumer {
    
    @Autowired
    private RedisTemplate<String, String> redisTemplate;
    
    @Autowired
    private IdempotentRecordRepository recordRepository;
    
    @KafkaListener(topics = "payment-success")
    public void consume(ConsumerRecord<String, String> record) {
        String idempotentKey = extractIdempotentKey(record);
        
        // 第一层：Redis布隆过滤器快速判断
        if (!bloomFilter.mightContain(idempotentKey)) {
            // 肯定未处理过，直接处理
            processMessage(record);
            bloomFilter.put(idempotentKey);
            return;
        }
        
        // 第二层：Redis SETNX分布式锁，5分钟过期
        String lockKey = "idempotent:lock:" + idempotentKey;
        Boolean locked = redisTemplate.opsForValue()
            .setIfAbsent(lockKey, "1", 5, TimeUnit.MINUTES);
        
        if (!locked) {
            log.warn("消息正在处理或已处理: {}", idempotentKey);
            return;
        }
        
        try {
            // 第三层：数据库唯一索引兜底
            IdempotentRecord idRecord = new IdempotentRecord();
            idRecord.setIdempotentKey(idempotentKey);
            idRecord.setCreateTime(LocalDateTime.now());
            
            recordRepository.insert(idRecord);
            processMessage(record);
        } catch (DuplicateKeyException e) {
            log.warn("重复消息已丢弃: {}", idempotentKey);
        } finally {
            redisTemplate.delete(lockKey);
        }
    }
}

public class OrderedMessageProcessor {
    
    // 订单ID -> 消息队列（按版本号排序）
    private final ConcurrentHashMap<String, PriorityBlockingQueue<OrderMessage>> 
        orderQueues = new ConcurrentHashMap<>();
    
    // 订单当前处理版本
    private final ConcurrentHashMap<String, Long> currentVersions = 
        new ConcurrentHashMap<>();
    
    public void processMessage(OrderMessage message) {
        String orderId = message.getOrderId();
        long version = message.getVersion();
        long currentVersion = currentVersions.getOrDefault(orderId, 0L);
        
        // 版本号检查
        if (version <= currentVersion) {
            log.warn("消息版本过期，orderId={}, version={}, current={}", 
                orderId, version, currentVersion);
            return;
        }
        
        // 放入内存队列等待处理
        orderQueues.computeIfAbsent(orderId, k -> 
            new PriorityBlockingQueue<>(Comparator.comparingLong(OrderMessage::getVersion))
        ).offer(message);
        
        // 触发顺序处理
        processQueue(orderId);
    }
    
    private void processQueue(String orderId) {
        PriorityBlockingQueue<OrderMessage> queue = orderQueues.get(orderId);
        Long currentVersion = currentVersions.getOrDefault(orderId, 0L);
        
        OrderMessage message;
        while ((message = queue.peek()) != null && message.getVersion() == currentVersion + 1) {
            queue.poll();
            doProcess(message);
            currentVersions.put(orderId, message.getVersion());
            currentVersion = message.getVersion();
        }
    }
}

@Component
public class LagMonitor {
    
    @Autowired
    private KafkaConsumer<String, String> adminConsumer;
    
    @Autowired
    private KubernetesClient k8sClient;
    
    @Scheduled(fixedRate = 30000)
    public void checkLag() {
        Map<String, ConsumerGroupDescription> groups = 
            adminConsumer.describeConsumerGroups(GROUPS).all().get();
        
        groups.forEach((groupId, description) -> {
            long totalLag = calculateTotalLag(groupId);
            
            // 分级处理
            if (totalLag > 1_000_000) {
                // 紧急：立即扩容至最大副本数
                scaleConsumers(groupId, MAX_REPLICAS);
                sendEmergencyAlert(groupId, totalLag);
            } else if (totalLag > 100_000) {
                // 严重：按比例扩容
                int targetReplicas = calculateTargetReplicas(totalLag);
                scaleConsumers(groupId, targetReplicas);
                sendWarningAlert(groupId, totalLag);
            } else if (totalLag > 10_000) {
                // 警告：开启批量消费模式
                enableBatchMode(groupId);
            }
        });
    }
    
    private void scaleConsumers(String groupId, int targetReplicas) {
        k8sClient.apps().deployments()
            .inNamespace("message-platform")
            .withName(groupId + "-consumer")
            .scale(targetReplicas);
    }
}

# MM2集群配置
clusters:
  - alias: "beijing"
    bootstrap.servers: "kafka-beijing1:9092,kafka-beijing2:9092"
  - alias: "shanghai"
    bootstrap.servers: "kafka-shanghai1:9092,kafka-shanghai2:9092"
  - alias: "shenzhen"
    bootstrap.servers: "kafka-shenzhen1:9092,kafka-shenzhen2:9092"

# 复制流程配置
source.cluster.alias: "beijing"
target.cluster.alias: "shanghai"

# 复制Topic白名单
topics: "order-events|payment-events|logistics-events"

# 复制参数优化
replication.factor: 3
num.streams: 8
refresh.topics.enabled: true
sync.topic.acls.enabled: false

# 消息转换：添加机房来源标识
replication.policy.class: "org.apache.kafka.connect.mirror.CustomReplicationPolicy"
replication.policy.separator: "."

# 高性能核心配置
num.network.threads=8
num.io.threads=16
socket.send.buffer.bytes=102400
socket.receive.buffer.bytes=102400
socket.request.max.bytes=104857600

# 高吞吐配置
batch.size=32768
linger.ms=5
compression.type=lz4
buffer.memory=67108864

# 数据可靠性配置
acks=all
retries=3
retry.backoff.ms=100
max.in.flight.requests.per.connection=5
enable.idempotence=true

# 日志刷盘策略
log.flush.interval.messages=10000
log.flush.interval.ms=1000
log.retention.hours=168
log.segment.bytes=1073741824

# 副本配置
default.replication.factor=3
min.insync.replicas=2
num.partitions=12
auto.create.topics.enable=false

@Component
public class DeadLetterHandler {
    
    private static final int MAX_RETRY_TIMES = 3;
    private static final long[] RETRY_DELAYS = {1000L, 5000L, 30000L};
    
    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;
    
    @Autowired
    private DeadLetterRepository dlqRepository;
    
    public void handleError(ConsumerRecord<String, String> record, 
                           Exception exception, int retryTimes) {
        String topic = record.topic();
        String dlqTopic = topic + "-dlq";
        
        // 包装死信消息
        DeadLetterMessage dlqMessage = DeadLetterMessage.builder()
            .originalTopic(topic)
            .partition(record.partition())
            .offset(record.offset())
            .key(record.key())
            .value(record.value())
            .exceptionMessage(exception.getMessage())
            .stackTrace(ExceptionUtils.getStackTrace(exception))
            .retryTimes(retryTimes)
            .failedTime(LocalDateTime.now())
            .build();
        
        if (retryTimes < MAX_RETRY_TIMES) {
            // 本地重试：延时后重新投递
            scheduleRetry(dlqMessage, RETRY_DELAYS[retryTimes]);
        } else {
            // 最终进入死信队列
            kafkaTemplate.send(dlqTopic, dlqMessage.toJson());
            dlqRepository.save(dlqMessage);
            
            // 发送告警通知
            alertService.sendAlert("消息消费失败", dlqMessage);
        }
    }
    
    // 死信队列消费：人工介入处理
    @KafkaListener(topics = "*-dlq", containerFactory = "dlqContainerFactory")
    public void processDLQ(ConsumerRecord<String, String> record) {
        DeadLetterMessage dlqMessage = DeadLetterMessage.fromJson(record.value());
        
        // 记录到管理后台供人工处理
        dlqManagementService.createTicket(dlqMessage);
        
        // 可选择自动修复：如数据补全、格式转换等
        if (autoFixService.canAutoFix(dlqMessage)) {
            autoFixService.fixAndRetry(dlqMessage);
        }
    }
}

public class CapacityBasedAssignor extends AbstractPartitionAssignor {
    
    @Override
    public String name() {
        return "capacity-based";
    }
    
    @Override
    public Map<String, List<TopicPartition>> assign(
            Map<String, Integer> partitionsPerTopic,
            Map<String, Subscription> subscriptions) {
        
        // 获取消费者的处理能力评分
        Map<String, Integer> consumerCapacities = subscriptions.entrySet().stream()
            .collect(Collectors.toMap(
                Map.Entry::getKey,
                e -> extractCapacity(e.getValue().userData())
            ));
        
        // 按处理能力加权分配分区
        Map<String, List<TopicPartition>> assignment = new HashMap<>();
        int totalCapacity = consumerCapacities.values().stream().mapToInt(Integer::intValue).sum();
        
        subscriptions.keySet().forEach(memberId -> assignment.put(memberId, new ArrayList<>()));
        
        partitionsPerTopic.forEach((topic, numPartitions) -> {
            List<TopicPartition> partitions = partitions(topic, numPartitions);
            
            // 按处理能力比例分配
            int currentPartition = 0;
            for (Map.Entry<String, Integer> entry : consumerCapacities.entrySet()) {
                String consumer = entry.getKey();
                int capacity = entry.getValue();
                int partitionCount = (int) Math.round(
                    (double) numPartitions * capacity / totalCapacity
                );
                
                for (int i = 0; i < partitionCount && currentPartition < numPartitions; i++) {
                    assignment.get(consumer).add(partitions.get(currentPartition++));
                }
            }
        });
        
        return assignment;
    }
}

# Kafka关键指标告警
groups:
  - name: kafka-alerts
    rules:
      # 消息积压告警
      - alert: KafkaHighConsumerLag
        expr: kafka_consumer_group_lag_sum > 100000
        for: 5m
        labels:
          severity: critical
        annotations:
          summary: "Kafka消费者积压过高"
          description: "ConsumerGroup  积压超过10万条"

      # Broker离线告警
      - alert: KafkaBrokerOffline
        expr: kafka_brokers < 9
        for: 1m
        labels:
          severity: critical
        annotations:
          summary: "Kafka Broker离线"
          description: "当前在线Broker数量: "

      # ISR不足告警
      - alert: KafkaInsufficientReplicas
        expr: kafka_topic_partition_in_sync_replicas < 2
        for: 5m
        labels:
          severity: warning
        annotations:
          summary: "Topic分区ISR不足"
          description: "Topic  分区  ISR数量不足"

      # 消费延迟告警
      - alert: KafkaConsumerProcessingDelay
        expr: kafka_consumer_records_consumed_rate < 100
        for: 10m
        labels:
          severity: warning
        annotations:
          summary: "Kafka消费速率过低"
          description: "ConsumerGroup  消费速率低于100条/秒"