一、顺序消息核心挑战与模型

1. 顺序消息的三种语义

图表

代码

复制

下载

全屏

graph TB
    A[顺序消息保证级别] --> B[全局顺序]
    A --> C[分区顺序]
    A --> D[会话顺序]
    
    B --> E[单分区/主题]
    C --> F[同一Key的消息]
    D --> G[同一生产者的消息]
    
    subgraph "适用场景"
        E --> H[证券交易行情]
        F --> I[电商订单状态流]
        G --> J[用户操作日志]
    end

2. 消息顺序性破坏的常见原因

java

复制

下载

// 顺序性破坏场景模拟
public class OrderViolationScenarios {
    
    /**
     * 场景1: 生产者并发发送
     */
    public void producerConcurrencyIssue() {
        // 多个线程同时发送顺序相关的消息
        ExecutorService executor = Executors.newFixedThreadPool(10);
        
        for (int i = 0; i < 10; i++) {
            final int messageId = i;
            executor.submit(() -> {
                // 线程执行顺序不确定，可能导致消息乱序
                producer.send(new Message()
                    .setKey("order-123")
                    .setBody("状态更新-" + messageId)
                    .setSequence(messageId));
            });
        }
    }
    
    /**
     * 场景2: 消息重试乱序
     */
    public void retryDisorderIssue() {
        // 消息1发送失败，重试期间消息2先到达
        producer.sendAsync(message1, new SendCallback() {
            @Override
            public void onException(Throwable e) {
                // 重试消息1
                producer.send(message1); // 可能晚于消息2到达
            }
        });
        
        // 消息2正常发送
        producer.send(message2);
    }
    
    /**
     * 场景3: 消费者并行处理
     */
    public void consumerParallelismIssue() {
        // 同一分区的消息被多个消费者线程并行处理
        consumer.subscribe("order-topic");
        consumer.registerMessageListener((messages, context) -> {
            // 默认并发消费，顺序无法保证
            messages.parallelStream().forEach(this::processMessage);
        });
    }
}

二、生产者顺序保证机制

1. 严格顺序生产者实现

java

复制

下载

// 基于内存队列的严格顺序生产者
public class StrictOrderProducer {
    
    private final Map<String, BlockingQueue<Message>> keyQueues = new ConcurrentHashMap<>();
    private final Map<String, Thread> senderThreads = new ConcurrentHashMap<>();
    private final Producer<Message> underlyingProducer;
    private final ExecutorService dispatcher = Executors.newCachedThreadPool();
    
    /**
     * 顺序发送消息（同一Key的消息严格有序）
     */
    public CompletableFuture<SendResult> sendInOrder(String key, Message message) {
        CompletableFuture<SendResult> future = new CompletableFuture<>();
        
        // 获取或创建该Key的发送队列
        BlockingQueue<SendTask> queue = keyQueues.computeIfAbsent(key, 
            k -> new LinkedBlockingQueue<>());
        
        // 创建发送任务
        SendTask task = new SendTask(message, future);
        queue.offer(task);
        
        // 确保发送线程运行
        ensureSenderThread(key, queue);
        
        return future;
    }
    
    /**
     * 为每个Key创建专用的发送线程
     */
    private void ensureSenderThread(String key, BlockingQueue<SendTask> queue) {
        senderThreads.computeIfAbsent(key, k -> {
            Thread thread = new Thread(() -> {
                while (!Thread.currentThread().isInterrupted()) {
                    try {
                        // 从队列取出任务，顺序执行
                        SendTask task = queue.take();
                        
                        // 同步发送，确保顺序
                        SendResult result = underlyingProducer.send(task.message);
                        task.future.complete(result);
                        
                    } catch (InterruptedException e) {
                        Thread.currentThread().interrupt();
                        break;
                    } catch (Exception e) {
                        // 发送失败，重试逻辑（保持顺序）
                        handleSendFailure(task, e);
                    }
                }
            }, "OrderedSender-" + key);
            
            thread.setDaemon(true);
            thread.start();
            return thread;
        });
    }
    
    /**
     * 带顺序保证的重试机制
     */
    private void handleSendFailure(SendTask failedTask, Exception e) {
        // 暂停当前Key的所有发送
        pauseKeySending(failedTask.message.getKey());
        
        // 指数退避重试
        int retryCount = 0;
        while (retryCount < MAX_RETRIES) {
            try {
                Thread.sleep(calculateBackoff(retryCount));
                SendResult result = underlyingProducer.send(failedTask.message);
                failedTask.future.complete(result);
                break;
            } catch (Exception retryException) {
                retryCount++;
                if (retryCount == MAX_RETRIES) {
                    failedTask.future.completeExceptionally(retryException);
                }
            }
        }
        
        // 恢复该Key的发送
        resumeKeySending(failedTask.message.getKey());
    }
    
    private static class SendTask {
        final Message message;
        final CompletableFuture<SendResult> future;
        
        SendTask(Message message, CompletableFuture<SendResult> future) {
            this.message = message;
            this.future = future;
        }
    }
}

2. 消息序列号与时间戳机制

java

复制

下载

// 基于序列号的顺序验证
public class SequenceNumberManager {
    
    private final Map<String, AtomicLong> lastSequenceNumbers = new ConcurrentHashMap<>();
    private final Map<String, Object> keyLocks = new ConcurrentHashMap<>();
    
    /**
     * 为消息生成严格递增的序列号
     */
    public long generateSequenceNumber(String key) {
        // 对每个Key加锁，确保序列号严格递增
        Object lock = keyLocks.computeIfAbsent(key, k -> new Object());
        
        synchronized (lock) {
            AtomicLong sequence = lastSequenceNumbers.computeIfAbsent(key, 
                k -> new AtomicLong(0));
            return sequence.incrementAndGet();
        }
    }
    
    /**
     * 验证消息顺序性
     */
    public boolean validateSequence(String key, long currentSequence) {
        AtomicLong lastSequence = lastSequenceNumbers.get(key);
        
        if (lastSequence == null) {
            // 第一条消息
            return true;
        }
        
        long last = lastSequence.get();
        
        if (currentSequence == last + 1) {
            // 顺序正确
            lastSequence.set(currentSequence);
            return true;
        } else if (currentSequence <= last) {
            // 重复消息或乱序消息
            log.warn("Out-of-order message detected. Key: {}, Expected: {}, Actual: {}", 
                    key, last + 1, currentSequence);
            return false;
        } else {
            // 消息跳跃（可能有消息丢失）
            log.error("Missing messages detected. Key: {}, Last: {}, Current: {}", 
                     key, last, currentSequence);
            return false;
        }
    }
    
    /**
     * 分布式序列号生成（基于数据库）
     */
    public long generateDistributedSequence(String key) {
        // 使用数据库事务确保序列号全局唯一且递增
        return jdbcTemplate.execute((Connection conn) -> {
            // 1. 获取排他锁
            conn.createStatement().execute(
                "SELECT * FROM message_sequences WHERE message_key = '" + key + "' FOR UPDATE");
            
            // 2. 读取当前序列号
            ResultSet rs = conn.createStatement().executeQuery(
                "SELECT sequence FROM message_sequences WHERE message_key = '" + key + "'");
            
            long currentSequence = 1;
            if (rs.next()) {
                currentSequence = rs.getLong("sequence") + 1;
                
                // 3. 更新序列号
                conn.createStatement().executeUpdate(
                    "UPDATE message_sequences SET sequence = " + currentSequence + 
                    " WHERE message_key = '" + key + "'");
            } else {
                // 4. 插入新记录
                conn.createStatement().executeUpdate(
                    "INSERT INTO message_sequences(message_key, sequence) VALUES('" + 
                    key + "', " + currentSequence + ")");
            }
            
            return currentSequence;
        });
    }
}

三、消息队列服务端顺序保证

1. RocketMQ顺序消息实现

java

复制

下载

// RocketMQ顺序消息核心实现
public class RocketMQOrderedMessage {
    
    /**
     * 顺序消息生产者
     */
    public class OrderedProducer {
        private final DefaultMQProducer producer;
        private final Map<String, MessageQueue> queueSelectorCache = new ConcurrentHashMap<>();
        
        public void sendOrdered(String topic, String keys, Message message) {
            try {
                // 根据消息Key选择队列（相同Key到相同队列）
                MessageQueue queue = selectMessageQueue(topic, keys);
                
                // 发送消息
                SendResult sendResult = producer.send(message, 
                    new MessageQueueSelector() {
                        @Override
                        public MessageQueue select(List<MessageQueue> mqs, 
                                                  Message msg, Object arg) {
                            String key = (String) arg;
                            int index = Math.abs(key.hashCode()) % mqs.size();
                            return mqs.get(index);
                        }
                    }, keys);
                
                // 验证发送结果
                if (sendResult.getSendStatus() != SendStatus.SEND_OK) {
                    throw new MQClientException("Send failed: " + sendResult.getSendStatus());
                }
                
            } catch (Exception e) {
                // 顺序消息发送失败需要特殊处理
                handleOrderedSendFailure(topic, keys, message, e);
            }
        }
        
        /**
         * 顺序消息失败处理
         */
        private void handleOrderedSendFailure(String topic, String keys, 
                                             Message message, Exception e) {
            // 1. 记录失败消息到本地存储
            persistFailedMessage(topic, keys, message);
            
            // 2. 暂停该队列的发送
            pauseQueueSending(topic, keys);
            
            // 3. 定时重试（保持顺序）
            scheduleOrderedRetry(topic, keys);
        }
    }
    
    /**
     * 顺序消息消费者
     */
    public class OrderedConsumer {
        private final DefaultMQPushConsumer consumer;
        private final Map<MessageQueue, ProcessQueue> processQueueTable = new ConcurrentHashMap<>();
        
        public void startOrderedConsuming() {
            // 注册顺序消息监听器
            consumer.registerMessageListener(new MessageListenerOrderly() {
                @Override
                public ConsumeOrderlyStatus consumeMessage(List<MessageExt> msgs,
                                                          ConsumeOrderlyContext context) {
                    MessageQueue queue = context.getMessageQueue();
                    
                    // 获取该队列的锁（确保同一队列串行消费）
                    synchronized (getQueueLock(queue)) {
                        for (MessageExt msg : msgs) {
                            try {
                                // 处理消息
                                processMessage(msg);
                                
                                // 更新消费进度
                                updateConsumeOffset(queue, msg);
                                
                            } catch (Exception e) {
                                // 顺序消息消费失败需要特殊处理
                                if (needSuspend(e)) {
                                    // 挂起当前队列的消费
                                    return ConsumeOrderlyStatus.SUSPEND_CURRENT_QUEUE_A_MOMENT;
                                } else {
                                    // 跳过当前消息继续消费（需确保业务允许）
                                    log.error("Skip message due to error", e);
                                }
                            }
                        }
                    }
                    
                    return ConsumeOrderlyStatus.SUCCESS;
                }
            });
            
            consumer.start();
        }
        
        /**
         * 队列级别的锁管理
         */
        private Object getQueueLock(MessageQueue queue) {
            return processQueueTable.computeIfAbsent(queue, 
                q -> new ProcessQueue()).getLock();
        }
        
        /**
         * 消费进度管理
         */
        private void updateConsumeOffset(MessageQueue queue, MessageExt msg) {
            // 确保严格按顺序提交offset
            ProcessQueue pq = processQueueTable.get(queue);
            long currentOffset = pq.getLastConsumeOffset();
            
            if (msg.getQueueOffset() == currentOffset + 1) {
                // 顺序正确，更新offset
                pq.setLastConsumeOffset(msg.getQueueOffset());
                consumer.updateConsumeOffset(queue, msg.getQueueOffset());
            } else {
                // 乱序消息，等待前面的消息
                log.warn("Out-of-order consumption detected. Queue: {}, Expected: {}, Actual: {}",
                        queue, currentOffset + 1, msg.getQueueOffset());
            }
        }
    }
}

 篇幅限制下面就只能给大家展示小册部分内容了。整理了一份核心面试笔记包括了：Java面试、Spring、JVM、MyBatis、Redis、MySQL、并发编程、微服务、Linux、Springboot、SpringCloud、MQ、Kafc

需要全套面试笔记及答案
【点击此处即可/免费获取】​​​

2. Kafka顺序保证实现

java

复制

下载

// Kafka顺序消息保证策略
public class KafkaOrderGuarantee {
    
    /**
     * Kafka生产者顺序保证
     */
    public class OrderedKafkaProducer {
        private final KafkaProducer<String, String> producer;
        private final Map<String, Semaphore> keySemaphores = new ConcurrentHashMap<>();
        
        /**
         * 保证同一Key的消息顺序
         */
        public Future<RecordMetadata> sendOrdered(String topic, String key, String value) {
            // 获取该Key的信号量（确保同一Key串行发送）
            Semaphore semaphore = keySemaphores.computeIfAbsent(key, 
                k -> new Semaphore(1));
            
            try {
                semaphore.acquire();
                
                // 发送消息
                ProducerRecord<String, String> record = new ProducerRecord<>(topic, key, value);
                
                return producer.send(record, (metadata, exception) -> {
                    // 发送完成后释放信号量
                    semaphore.release();
                    
                    if (exception != null) {
                        handleSendFailure(topic, key, value, exception);
                    }
                });
                
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                throw new RuntimeException("Interrupted while waiting for send lock", e);
            }
        }
        
        /**
         * 配置保证顺序的关键参数
         */
        private Properties getOrderedProducerConfig() {
            Properties props = new Properties();
            
            // 关键配置：确保消息不重排
            props.put("max.in.flight.requests.per.connection", "1"); // 最重要！
            props.put("acks", "all");                                // 所有副本确认
            props.put("retries", Integer.MAX_VALUE);                 // 无限重试
            props.put("max.block.ms", Long.MAX_VALUE);               // 无限阻塞
            
            // 启用幂等和事务（如果需要更强保证）
            props.put("enable.idempotence", "true");
            props.put("transactional.id", "ordered-producer-" + UUID.randomUUID());
            
            return props;
        }
    }
    
    /**
     * Kafka消费者顺序保证
     */
    public class OrderedKafkaConsumer {
        private final KafkaConsumer<String, String> consumer;
        private final ExecutorService threadPool;
        private final Map<TopicPartition, ConsumerWorker> partitionWorkers = new ConcurrentHashMap<>();
        
        /**
         * 分区级别的顺序消费
         */
        public void startPartitionOrderedConsumption() {
            // 订阅主题
            consumer.subscribe(Collections.singletonList("order-topic"));
            
            // 为每个分区创建独立的消费线程
            while (true) {
                ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
                
                // 按分区分组处理
                for (TopicPartition partition : records.partitions()) {
                    List<ConsumerRecord<String, String>> partitionRecords = 
                        records.records(partition);
                    
                    // 获取或创建该分区的消费线程
                    ConsumerWorker worker = partitionWorkers.computeIfAbsent(partition,
                        p -> new ConsumerWorker(p));
                    
                    // 提交到对应线程处理
                    worker.submitRecords(partitionRecords);
                }
            }
        }
        
        /**
         * 分区消费工作线程
         */
        private class ConsumerWorker {
            private final TopicPartition partition;
            private final LinkedBlockingQueue<ConsumerRecord<String, String>> queue;
            private final Thread workerThread;
            private volatile boolean running = true;
            
            public ConsumerWorker(TopicPartition partition) {
                this.partition = partition;
                this.queue = new LinkedBlockingQueue<>();
                
                this.workerThread = new Thread(() -> {
                    while (running) {
                        try {
                            // 从队列中顺序消费消息
                            ConsumerRecord<String, String> record = queue.take();
                            processRecord(record);
                            
                            // 手动提交offset（单条提交确保顺序）
                            Map<TopicPartition, OffsetAndMetadata> offsets = new HashMap<>();
                            offsets.put(partition, 
                                new OffsetAndMetadata(record.offset() + 1));
                            consumer.commitSync(offsets);
                            
                        } catch (InterruptedException e) {
                            Thread.currentThread().interrupt();
                            break;
                        } catch (Exception e) {
                            // 处理失败，暂停该分区的消费
                            handleProcessingFailure(e);
                        }
                    }
                }, "ConsumerWorker-" + partition);
                
                workerThread.start();
            }
            
            public void submitRecords(List<ConsumerRecord<String, String>> records) {
                // 将消息按offset排序后加入队列
                records.sort(Comparator.comparing(ConsumerRecord::offset));
                queue.addAll(records);
            }
        }
    }
}

四、消费者顺序处理保证

1. 消费者并发控制策略

java

复制

下载

// 基于锁的消费者顺序控制
public class ConsumerOrderController {
    
    /**
     * Key级别的锁控制
     */
    public class KeyLockConsumer {
        private final Map<String, ReentrantLock> keyLocks = new ConcurrentHashMap<>();
        private final LoadingCache<String, ReentrantLock> lockCache;
        
        public KeyLockConsumer() {
            // 使用Guava Cache自动清理不用的锁
            lockCache = CacheBuilder.newBuilder()
                .maximumSize(10000)
                .expireAfterAccess(10, TimeUnit.MINUTES)
                .build(new CacheLoader<String, ReentrantLock>() {
                    @Override
                    public ReentrantLock load(String key) {
                        return new ReentrantLock();
                    }
                });
        }
        
        public void processOrderedByKey(Message message) {
            String key = extractKey(message);
            ReentrantLock lock = lockCache.getUnchecked(key);
            
            try {
                // 尝试获取锁，如果获取不到则跳过（稍后重试）
                if (lock.tryLock(100, TimeUnit.MILLISECONDS)) {
                    try {
                        // 处理消息
                        processMessage(message);
                    } finally {
                        lock.unlock();
                    }
                } else {
                    // 锁被占用，稍后重试
                    log.debug("Lock busy for key: {}, will retry later", key);
                    requeueMessage(message);
                }
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                throw new RuntimeException(e);
            }
        }
    }
    
    /**
     * 基于数据库行锁的顺序处理
     */
    public class DatabaseLockConsumer {
        private final DataSource dataSource;
        
        public void processWithDatabaseLock(Message message) {
            String businessKey = extractBusinessKey(message);
            
            Connection conn = null;
            try {
                conn = dataSource.getConnection();
                conn.setAutoCommit(false);
                
                // 1. 获取行级锁
                PreparedStatement lockStmt = conn.prepareStatement(
                    "SELECT * FROM processing_locks WHERE business_key = ? FOR UPDATE");
                lockStmt.setString(1, businessKey);
                lockStmt.executeQuery();
                
                // 2. 检查是否已有处理中的消息
                PreparedStatement checkStmt = conn.prepareStatement(
                    "SELECT status FROM message_status WHERE business_key = ? " +
                    "AND expected_sequence = ?");
                checkStmt.setString(1, businessKey);
                checkStmt.setLong(2, message.getSequence());
                
                ResultSet rs = checkStmt.executeQuery();
                if (rs.next() && "PROCESSING".equals(rs.getString("status"))) {
                    // 已在处理中，跳过
                    conn.rollback();
                    return;
                }
                
                // 3. 标记为处理中
                PreparedStatement updateStmt = conn.prepareStatement(
                    "INSERT INTO message_status (business_key, expected_sequence, status) " +
                    "VALUES (?, ?, 'PROCESSING') ON DUPLICATE KEY UPDATE status = 'PROCESSING'");
                updateStmt.setString(1, businessKey);
                updateStmt.setLong(2, message.getSequence());
                updateStmt.executeUpdate();
                
                // 4. 处理消息
                processMessage(message);
                
                // 5. 更新状态为完成
                PreparedStatement completeStmt = conn.prepareStatement(
                    "UPDATE message_status SET status = 'COMPLETED' " +
                    "WHERE business_key = ? AND expected_sequence = ?");
                completeStmt.setString(1, businessKey);
                completeStmt.setLong(2, message.getSequence());
                completeStmt.executeUpdate();
                
                conn.commit();
                
            } catch (SQLException e) {
                if (conn != null) {
                    try { conn.rollback(); } catch (SQLException ex) {}
                }
                throw new RuntimeException("Database lock failed", e);
            } finally {
                if (conn != null) {
                    try { conn.close(); } catch (SQLException e) {}
                }
            }
        }
    }
}

2. 消费者状态机管理

java

复制

下载

// 基于状态机的顺序消费
public class StatefulOrderedConsumer {
    
    private final Map<String, ConsumerStateMachine> stateMachines = new ConcurrentHashMap<>();
    
    /**
     * 消费者状态机
     */
    private class ConsumerStateMachine {
        private final String key;
        private volatile long expectedSequence = 1;
        private final BlockingQueue<BufferedMessage> buffer = new LinkedBlockingQueue<>();
        private final Thread processorThread;
        
        public ConsumerStateMachine(String key) {
            this.key = key;
            
            this.processorThread = new Thread(() -> {
                while (!Thread.currentThread().isInterrupted()) {
                    try {
                        // 从缓冲队列取出消息
                        BufferedMessage buffered = buffer.take();
                        
                        // 检查是否是期望的序列号
                        if (buffered.sequence == expectedSequence) {
                            // 处理消息
                            processMessage(buffered.message);
                            expectedSequence++;
                            
                            // 检查缓冲区中是否有连续的消息
                            processBufferedMessages();
                            
                        } else if (buffered.sequence > expectedSequence) {
                            // 未来的消息，先缓冲起来
                            buffer.put(buffered);
                        } else {
                            // 重复或过时的消息，丢弃
                            log.warn("Duplicate or stale message. Key: {}, Expected: {}, Actual: {}",
                                    key, expectedSequence, buffered.sequence);
                        }
                        
                    } catch (InterruptedException e) {
                        Thread.currentThread().interrupt();
                        break;
                    }
                }
            }, "StateMachine-" + key);
            
            processorThread.start();
        }
        
        public void submitMessage(Message message, long sequence) {
            buffer.offer(new BufferedMessage(message, sequence));
        }
        
        private void processBufferedMessages() {
            while (!buffer.isEmpty()) {
                BufferedMessage peeked = buffer.peek();
                if (peeked != null && peeked.sequence == expectedSequence) {
                    buffer.poll(); // 移除
                    processMessage(peeked.message);
                    expectedSequence++;
                } else {
                    break;
                }
            }
        }
    }
    
    /**
     * 处理消息入口
     */
    public void process(Message message) {
        String key = extractKey(message);
        long sequence = extractSequence(message);
        
        // 获取或创建状态机
        ConsumerStateMachine stateMachine = stateMachines.computeIfAbsent(key,
            k -> new ConsumerStateMachine(k));
        
        // 提交到状态机处理
        stateMachine.submitMessage(message, sequence);
    }
    
    private static class BufferedMessage {
        final Message message;
        final long sequence;
        
        BufferedMessage(Message message, long sequence) {
            this.message = message;
            this.sequence = sequence;
        }
    }
}

五、故障恢复与容错机制

1. 生产者故障恢复

java

复制

下载

// 生产者故障恢复策略
public class ProducerFaultRecovery {
    
    /**
     * 本地消息表模式（确保不丢消息）
     */
    public class LocalMessageTable {
        private final DataSource dataSource;
        private final Producer<Message> mqProducer;
        
        /**
         * 两阶段提交：先存本地，再发MQ
         */
        @Transactional
        public void sendWithLocalTable(Message message) {
            // 1. 生成唯一ID
            String messageId = UUID.randomUUID().toString();
            
            // 2. 插入本地消息表（与业务操作在同一事务）
            jdbcTemplate.update(
                "INSERT INTO local_messages(id, topic, key, body, status, created_time) " +
                "VALUES (?, ?, ?, ?, 'PENDING', NOW())",
                messageId, message.getTopic(), message.getKey(), 
                message.getBody());
            
            // 3. 异步发送到MQ
            CompletableFuture.runAsync(() -> {
                sendToMQWithRetry(messageId, message);
            });
        }
        
        /**
         * 带重试的MQ发送
         */
        private void sendToMQWithRetry(String messageId, Message message) {
            int retryCount = 0;
            boolean success = false;
            
            while (retryCount < MAX_RETRIES && !success) {
                try {
                    SendResult result = mqProducer.send(message);
                    
                    if (result.getSendStatus() == SendStatus.SEND_OK) {
                        // 4. 更新本地状态为已发送
                        jdbcTemplate.update(
                            "UPDATE local_messages SET status = 'SENT', " +
                            "sent_time = NOW() WHERE id = ?", 
                            messageId);
                        success = true;
                    }
                    
                } catch (Exception e) {
                    retryCount++;
                    log.warn("Send failed, retry {}/{}", retryCount, MAX_RETRIES, e);
                    
                    // 指数退避
                    try {
                        Thread.sleep(calculateBackoff(retryCount));
                    } catch (InterruptedException ie) {
                        Thread.currentThread().interrupt();
                        break;
                    }
                }
            }
            
            if (!success) {
                // 5. 标记为失败，需要人工干预
                jdbcTemplate.update(
                    "UPDATE local_messages SET status = 'FAILED', " +
                    "error_msg = ? WHERE id = ?", 
                    "Max retries exceeded", messageId);
                alertManualIntervention(messageId);
            }
        }
        
        /**
         * 定时任务：扫描并重试失败消息
         */
        @Scheduled(fixedDelay = 60000) // 每分钟执行
        public void retryFailedMessages() {
            // 查找PENDING或FAILED状态的消息
            List<LocalMessage> failedMessages = jdbcTemplate.query(
                "SELECT * FROM local_messages WHERE status IN ('PENDING', 'FAILED') " +
                "AND created_time > ? ORDER BY created_time",
                new BeanPropertyRowMapper<>(LocalMessage.class),
                LocalDateTime.now().minusHours(24)); // 只处理24小时内的
            
            for (LocalMessage localMsg : failedMessages) {
                Message mqMsg = convertToMQMessage(localMsg);
                sendToMQWithRetry(localMsg.getId(), mqMsg);
            }
        }
    }
    
    /**
     * 基于WAL（Write-Ahead Log）的恢复机制
     */
    public class WALBasedRecovery {
        private final WriteAheadLog wal;
        private final Producer<Message> producer;
        private final Map<String, Long> inFlightMessages = new ConcurrentHashMap<>();
        
        public void sendWithWAL(Message message) {
            String messageId = generateMessageId();
            
            // 1. 写入WAL（顺序写入，性能高）
            wal.append(new WALEntry(messageId, message, "PENDING"));
            
            // 2. 记录为飞行中
            inFlightMessages.put(messageId, System.currentTimeMillis());
            
            // 3. 异步发送
            CompletableFuture.runAsync(() -> {
                sendAndConfirm(messageId, message);
            });
        }
        
        private void sendAndConfirm(String messageId, Message message) {
            try {
                SendResult result = producer.send(message);
                
                if (result.getSendStatus() == SendStatus.SEND_OK) {
                    // 4. 更新WAL状态
                    wal.append(new WALEntry(messageId, null, "CONFIRMED"));
                    
                    // 5. 从飞行中移除
                    inFlightMessages.remove(messageId);
                }
                
            } catch (Exception e) {
                log.error("Failed to send message: {}", messageId, e);
            }
        }
        
        /**
         * 恢复未确认的消息
         */
        @PostConstruct
        public void recoverOnStartup() {
            // 1. 读取WAL中未确认的消息
            List<WALEntry> unconfirmed = wal.readUnconfirmedEntries();
            
            // 2. 重新发送
            for (WALEntry entry : unconfirmed) {
                if (!inFlightMessages.containsKey(entry.getMessageId())) {
                    // 重新加入飞行中
                    inFlightMessages.put(entry.getMessageId(), System.currentTimeMillis());
                    
                    // 重新发送
                    sendAndConfirm(entry.getMessageId(), entry.getMessage());
                }
            }
            
            // 3. 清理超时的飞行中消息
            cleanupTimeoutMessages();
        }
        
        private void cleanupTimeoutMessages() {
            long cutoff = System.currentTimeMillis() - 300000; // 5分钟超时
            
            inFlightMessages.entrySet().removeIf(entry -> {
                if (entry.getValue() < cutoff) {
                    log.error("Message timeout: {}", entry.getKey());
                    return true;
                }
                return false;
            });
        }
    }
}

2. 消费者故障恢复

java

复制

下载

// 消费者故障恢复策略
public class ConsumerFaultRecovery {
    
    /**
     * 幂等消费与去重机制
     */
    public class IdempotentConsumer {
        private final Cache<String, Boolean> processedMessageCache;
        private final DataSource dataSource;
        
        public IdempotentConsumer() {
            // 内存缓存最近处理的消息（防瞬时重复）
            processedMessageCache = CacheBuilder.newBuilder()
                .maximumSize(100000)
                .expireAfterWrite(10, TimeUnit.MINUTES)
                .build();
        }
        
        public boolean processWithIdempotence(Message message) {
            String messageId = message.getMsgId();
            
            // 1. 检查内存缓存
            if (processedMessageCache.getIfPresent(messageId) != null) {
                log.debug("Duplicate message detected in memory cache: {}", messageId);
                return false;
            }
            
            // 2. 检查数据库（防持久化重复）
            if (checkDatabaseForDuplicate(messageId)) {
                log.warn("Duplicate message detected in database: {}", messageId);
                return false;
            }
            
            Connection conn = null;
            try {
                conn = dataSource.getConnection();
                conn.setAutoCommit(false);
                
                // 3. 插入去重记录
                PreparedStatement insertStmt = conn.prepareStatement(
                    "INSERT INTO consumed_messages(message_id, topic, consume_time) " +
                    "VALUES (?, ?, NOW())");
                insertStmt.setString(1, messageId);
                insertStmt.setString(2, message.getTopic());
                
                try {
                    insertStmt.executeUpdate();
                } catch (SQLIntegrityConstraintViolationException e) {
                    // 唯一键冲突，说明是重复消息
                    conn.rollback();
                    return false;
                }
                
                // 4. 处理业务逻辑
                boolean success = processBusinessLogic(message);
                
                if (success) {
                    conn.commit();
                    
                    // 5. 更新内存缓存
                    processedMessageCache.put(messageId, true);
                    
                    return true;
                } else {
                    conn.rollback();
                    return false;
                }
                
            } catch (SQLException e) {
                log.error("Database error during idempotence check", e);
                return false;
            } finally {
                if (conn != null) {
                    try { conn.close(); } catch (SQLException e) {}
                }
            }
        }
    }
    
    /**
     * 消费者offset管理恢复
     */
    public class OffsetRecoveryManager {
        private final KafkaConsumer<String, String> consumer;
        private final OffsetStorage offsetStorage;
        
        /**
         * 安全的offset提交策略
         */
        public void safeConsumeAndCommit() {
            ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
            
            // 按分区处理，失败的分区不提交offset
            Map<TopicPartition, OffsetAndMetadata> offsetsToCommit = new HashMap<>();
            
            for (TopicPartition partition : records.partitions()) {
                List<ConsumerRecord<String, String>> partitionRecords = 
                    records.records(partition);
                
                try {
                    // 处理分区内的消息
                    boolean success = processPartitionRecords(partition, partitionRecords);
                    
                    if (success) {
                        // 成功则记录要提交的offset
                        long lastOffset = partitionRecords.get(partitionRecords.size() - 1).offset();
                        offsetsToCommit.put(partition, new OffsetAndMetadata(lastOffset + 1));
                    } else {
                        // 失败则跳过该分区的offset提交
                        log.error("Failed to process partition: {}, will retry", partition);
                    }
                    
                } catch (Exception e) {
                    log.error("Error processing partition: {}", partition, e);
                }
            }
            
            // 提交成功的分区的offset
            if (!offsetsToCommit.isEmpty()) {
                consumer.commitSync(offsetsToCommit);
            }
        }
        
        /**
         * 灾难恢复：offset重置与重放
         */
        public void recoverFromDisaster(String topic, int hoursBack) {
            // 1. 获取所有分区
            List<PartitionInfo> partitions = consumer.partitionsFor(topic);
            
            for (PartitionInfo partitionInfo : partitions) {
                TopicPartition partition = new TopicPartition(topic, partitionInfo.partition());
                
                // 2. 获取最早和最新的offset
                Map<TopicPartition, Long> beginningOffsets = 
                    consumer.beginningOffsets(Collections.singleton(partition));
                Map<TopicPartition, Long> endOffsets = 
                    consumer.endOffsets(Collections.singleton(partition));
                
                long beginning = beginningOffsets.get(partition);
                long end = endOffsets.get(partition);
                
                // 3. 计算要重放的时间点offset
                long targetOffset = calculateOffsetByTime(partition, hoursBack);
                targetOffset = Math.max(beginning, Math.min(end, targetOffset));
                
                // 4. 重置offset
                consumer.seek(partition, targetOffset);
                log.info("Reset partition {} offset to {} for recovery", partition, targetOffset);
                
                // 5. 开始重放消费
                recoverConsumeFromOffset(partition, targetOffset);
            }
        }
        
        private long calculateOffsetByTime(TopicPartition partition, int hoursBack) {
            // 查找指定时间之前的offset
            long targetTime = System.currentTimeMillis() - (hoursBack * 3600000L);
            Map<TopicPartition, OffsetAndTimestamp> offsets = 
                consumer.offsetsForTimes(Collections.singletonMap(partition, targetTime));
            
            OffsetAndTimestamp offsetAndTimestamp = offsets.get(partition);
            if (offsetAndTimestamp != null) {
                return offsetAndTimestamp.offset();
            } else {
                // 如果找不到，返回最早offset
                return consumer.beginningOffsets(Collections.singleton(partition))
                    .get(partition);
            }
        }
    }
}

篇幅限制下面就只能给大家展示小册部分内容了。整理了一份核心面试笔记包括了：Java面试、Spring、JVM、MyBatis、Redis、MySQL、并发编程、微服务、Linux、Springboot、SpringCloud、MQ、Kafc

需要全套面试笔记及答案
【点击此处即可/免费获取】​​​

六、顺序消息监控与告警

1. 顺序性监控体系

java

复制

下载

// 顺序消息监控组件
@RestController
@RequestMapping("/monitor/order")
public class OrderMessageMonitor {
    
    @Autowired
    private MetricsCollector metricsCollector;
    @Autowired
    private AlertService alertService;
    
    /**
     * 顺序性指标收集
     */
    @Component
    public class OrderMetricsCollector {
        private final MeterRegistry meterRegistry;
        private final Map<String, AtomicLong> lastSequenceNumbers = new ConcurrentHashMap<>();
        
        /**
         * 记录消息到达顺序
         */
        public void recordMessageArrival(String key, long sequence) {
            AtomicLong lastSeq = lastSequenceNumbers.computeIfAbsent(key, 
                k -> new AtomicLong(0));
            
            long expected = lastSeq.get() + 1;
            
            if (sequence == expected) {
                // 顺序正确
                lastSeq.set(sequence);
                meterRegistry.counter("order.messages.correct").increment();
                
            } else if (sequence > expected) {
                // 消息丢失
                long lostCount = sequence - expected;
                meterRegistry.counter("order.messages.lost", "key", key)
                    .increment(lostCount);
                
                // 告警
                if (lostCount > 10) {
                    alertService.sendAlert(new Alert(
                        "ORDER_MESSAGE_LOST",
                        String.format("Key %s lost %d messages", key, lostCount),
                        AlertSeverity.CRITICAL
                    ));
                }
                
                // 更新为当前序列号
                lastSeq.set(sequence);
                
            } else {
                // 消息乱序
                meterRegistry.counter("order.messages.out_of_order").increment();
                
                // 记录详细乱序信息
                meterRegistry.gauge("order.out_of_order.gap", 
                    Tags.of("key", key), 
                    expected - sequence);
            }
        }
        
        /**
         * 生成顺序性报告
         */
        public OrderReport generateOrderReport(String key, Duration period) {
            OrderReport report = new OrderReport();
            
            // 计算顺序正确率
            double correctRate = meterRegistry.counter("order.messages.correct")
                .count() / (double) meterRegistry.counter("order.messages.total").count();
            report.setCorrectRate(correctRate);
            
            // 计算平均乱序间隔
            Map<String, Double> gauges = metricsCollector.getGauges(
                "order.out_of_order.gap", period);
            double avgGap = gauges.values().stream()
                .mapToDouble(Double::doubleValue)
                .average()
                .orElse(0.0);
            report.setAvgOutOfOrderGap(avgGap);
            
            // 检测热点Key
            report.setHotKeys(detectHotKeys(period));
            
            // 延迟分析
            report.setLatencyAnalysis(analyzeLatency(key, period));
            
            return report;
        }
    }
    
    /**
     * 顺序性验证服务
     */
    @Component
    public class OrderValidator {
        
        /**
         * 实时顺序验证
         */
        public ValidationResult validateInRealTime(Message message) {
            String key = message.getKey();
            long sequence = message.getSequence();
            long timestamp = message.getTimestamp();
            
            ValidationResult result = new ValidationResult();
            
            // 1. 序列号连续性验证
            if (!validateSequenceContinuity(key, sequence)) {
                result.addViolation("SEQUENCE_DISCONTINUITY");
            }
            
            // 2. 时间戳单调性验证
            if (!validateTimestampMonotonicity(key, timestamp)) {
                result.addViolation("TIMESTAMP_NON_MONOTONIC");
            }
            
            // 3. 生产者标识验证
            if (!validateProducerConsistency(key, message.getProducerId())) {
                result.addViolation("PRODUCER_INCONSISTENCY");
            }
            
            // 4. 业务逻辑验证
            if (!validateBusinessLogic(message)) {
                result.addViolation("BUSINESS_LOGIC_VIOLATION");
            }
            
            result.setValid(result.getViolations().isEmpty());
            
            if (!result.isValid()) {
                // 触发告警
                triggerValidationAlert(key, result);
                
                // 记录到死信队列
                sendToDeadLetterQueue(message, result);
            }
            
            return result;
        }
        
        /**
         * 批量回溯验证
         */
        public BatchValidationResult batchValidate(String key, 
                                                  LocalDateTime startTime,
                                                  LocalDateTime endTime) {
            // 从存储中读取该时间段的消息
            List<Message> messages = loadMessagesFromStorage(key, startTime, endTime);
            
            // 排序并验证
            messages.sort(Comparator.comparing(Message::getSequence));
            
            BatchValidationResult batchResult = new BatchValidationResult();
            long expected = 1;
            
            for (Message message : messages) {
                if (message.getSequence() != expected) {
                    batchResult.addGap(expected, message.getSequence());
                    expected = message.getSequence() + 1;
                } else {
                    expected++;
                }
                
                // 验证单条消息
                ValidationResult singleResult = validateInRealTime(message);
                if (!singleResult.isValid()) {
                    batchResult.addInvalidMessage(message, singleResult);
                }
            }
            
            return batchResult;
        }
    }
}

七、最佳实践与调优

1. 配置优化指南

yaml

复制

下载

# RocketMQ顺序消息最佳配置
rocketmq:
  producer:
    # 顺序生产者配置
    ordered:
      sendMessageTimeout: 3000           # 发送超时3秒
      retryTimesWhenSendFailed: 3        # 失败重试次数
      retryTimesWhenSendAsyncFailed: 1   # 异步发送重试
      maxMessageSize: 4096               # 最大消息大小
      compressMessageBodyOverHowmuch: 4096  # 压缩阈值
      
      # 队列选择策略
      messageQueueSelector: "hash"       # 哈希选择
      
  consumer:
    # 顺序消费者配置
    ordered:
      consumeThreadMin: 20               # 最小消费线程
      consumeThreadMax: 20               # 最大消费线程（保持相等，避免动态调整）
      pullBatchSize: 32                  # 每次拉取数量
      consumeMessageBatchMaxSize: 1      # 批量消费最大为1（顺序消费）
      maxReconsumeTimes: 16              # 最大重试次数
      suspendCurrentQueueTimeMillis: 1000 # 队列挂起时间
      
      # 消费位点管理
      consumeFromWhere: "CONSUME_FROM_LAST_OFFSET"  # 从哪里开始消费
      consumeTimestamp: "30分钟前"        # 时间回溯消费

2. 架构设计建议

java

复制

下载

// 顺序消息架构设计模式
public class OrderMessagePatterns {
    
    /**
     * 模式1: 强顺序业务拆分
     */
    public void patternStrongOrderSegregation() {
        // 将需要强顺序的业务拆分为独立的Topic
        Map<BusinessType, String> topicMapping = Map.of(
            BusinessType.ORDER_CREATE, "order-create-topic",
            BusinessType.ORDER_PAY,    "order-pay-topic",
            BusinessType.ORDER_REFUND, "order-refund-topic"
        );
        
        // 每个Topic内部保证顺序，不同Topic间无顺序要求
        // 优点：降低复杂度，提高并行度
    }
    
    /**
     * 模式2: 顺序窗口缓冲
     */
    public void patternOrderedWindowBuffer() {
        // 使用时间窗口缓冲消息，窗口内消息可并行处理
        // 窗口间保持顺序
        TimeWindowBuffer buffer = new TimeWindowBuffer(Duration.ofSeconds(5));
        
        buffer.addMessage(message, (windowMessages) -> {
            // 窗口内的消息可以并行处理
            windowMessages.parallelStream()
                .forEach(this::processInWindow);
            
            // 窗口间串行
        });
    }
    
    /**
     * 模式3: 业务层顺序保证
     */
    public void patternBusinessLayerOrder() {
        // 在业务层实现最终顺序，MQ只负责传输
        // 使用版本号或状态机确保最终一致性
        
        // 示例：订单状态机
        OrderStateMachine stateMachine = new OrderStateMachine(orderId);
        
        // 接收乱序消息，但按状态机顺序处理
        stateMachine.transition(OrderEvent.PAY_SUCCESS);
        stateMachine.transition(OrderEvent.SHIPPED);  // 即使消息乱序到达，状态机保证顺序
    }
    
    /**
     * 模式4: 补偿事务机制
     */
    public void patternCompensatingTransaction() {
        // 对于关键顺序业务，使用补偿事务
        
        // 1. 尝试执行业务
        try {
            executeBusiness(message);
            
            // 2. 发送确认消息
            sendConfirmation(message);
            
        } catch (Exception e) {
            // 3. 失败则执行补偿
            executeCompensation(message);
            
            // 4. 发送补偿完成消息
            sendCompensationComplete(message);
        }
        
        // 消费者端需要处理重复的确认/补偿消息（幂等）
    }
}

3. 故障处理预案

markdown

复制

下载

# 顺序消息故障处理预案

## 场景1: 生产者发送失败
### 现象：
- 消息发送返回失败
- 本地消息表中有PENDING状态消息

### 处理步骤：
1. 检查MQ集群状态
2. 检查网络连接
3. 自动重试（指数退避）
4. 重试超过阈值后告警
5. 人工介入：检查消息内容、修复配置

## 场景2: 消费者顺序错乱
### 现象：
- 监控告警顺序性指标异常
- 业务数据出现状态不一致

### 处理步骤：
1. 暂停问题分区/队列的消费
2. 检查消费者并发配置
3. 检查消费位点是否正确
4. 修复数据状态（可能需要人工干预）
5. 重置消费位点到正确位置
6. 恢复消费，监控顺序性

## 场景3: 消息丢失
### 现象：
- 监控显示消息数量不匹配
- 业务发现数据缺失

### 处理步骤：
1. 确认丢失范围和数量
2. 检查生产者确认机制
3. 检查MQ存储状态
4. 从备份/日志恢复丢失消息
5. 重新投递恢复的消息
6. 验证数据完整性

## 场景4: 消息积压
### 现象：
- 消费延迟增加
- 队列长度持续增长

### 处理步骤：
1. 增加消费者实例（注意顺序性约束）
2. 调整消费批处理大小
3. 优化消费逻辑性能
4. 紧急情况：临时关闭非关键功能
5. 长期方案：重新设计分区策略

顺序消息的保证是一个系统工程，需要从生产者、MQ服务端、消费者三个层面协同工作。在追求顺序性的同时，也要权衡性能、可用性和复杂度。实际应用中应根据业务需求选择合适的顺序保证级别，并在监控、告警、故障恢复等方面做好充分准备。