面试题：让你设计一个消息队列，怎么设计？

设计一个消息队列（Message Queue）是一个经典的面试题，涉及分布式系统、高并发、数据持久化等多个方面。以下是设计一个简化版消息队列的详细思路和实现方案。

1. 需求分析

核心功能

消息发布：生产者将消息发送到队列。
消息消费：消费者从队列中获取消息并处理。
消息存储：持久化消息，确保消息不丢失。
消息确认：消费者处理完消息后，确认消息已被消费。
消息重试：当消息处理失败时，支持重试。

非功能需求

高性能：支持高并发发布和消费消息。
高可用：确保消息队列的可用性。
可扩展性：支持水平扩展以应对流量增长。
数据一致性：确保消息不丢失、不重复。

2. 系统设计

核心组件

生产者（Producer）：发布消息到队列。
消费者（Consumer）：从队列中消费消息。
消息队列（Queue）：存储消息的队列。
消息存储（Storage）：持久化消息（如数据库、文件系统）。
消息确认机制（Ack）：消费者处理完消息后确认。
消息重试机制（Retry）：处理失败的消息重新入队。

3. 详细设计

消息队列数据结构

使用链表或数组实现队列。
支持先进先出（FIFO）的消息处理。

消息存储

内存存储：适合高性能场景，但数据易丢失。
持久化存储：使用数据库（如 MySQL）或文件系统（如 Kafka 的日志文件）存储消息。

消息确认机制

消费者处理完消息后，发送确认（ACK）给队列。
队列收到 ACK 后，删除消息。

消息重试机制

如果消费者处理失败，消息重新入队。
设置最大重试次数，超过次数后进入死信队列。

高可用设计

使用主从复制（如 Kafka 的副本机制）确保数据不丢失。
使用分布式一致性协议（如 Raft）选举主节点。

4. 关键问题与解决方案

1. 消息丢失

解决方案：
- 使用持久化存储（如数据库、文件系统）。
- 生产者等待消息存储成功后再返回。

2. 消息重复

解决方案：
- 消费者实现幂等性（Idempotency）。
- 使用唯一消息 ID 去重。

3. 消息顺序

解决方案：
- 使用单线程消费队列。
- 在分布式场景下，使用分区（Partition）保证分区内消息顺序。

4. 高并发支持

解决方案：
- 使用多线程或多进程消费队列。
- 使用分布式队列（如 Kafka）水平扩展。

5. 示例实现

消息队列核心类

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;

public class MessageQueue {
    private final BlockingQueue<String> queue; // 消息队列
    private final Storage storage; // 消息存储

    public MessageQueue(Storage storage) {
        this.queue = new LinkedBlockingQueue<>();
        this.storage = storage;
    }

    // 发布消息
    public void publish(String message) {
        try {
            storage.save(message); // 持久化消息
            queue.put(message); // 加入队列
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
    }

    // 消费消息
    public String consume() {
        try {
            return queue.take(); // 从队列中获取消息
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            return null;
        }
    }

    // 确认消息
    public void ack(String message) {
        storage.delete(message); // 删除已消费的消息
    }
}

消息存储接口

public interface Storage {
    void save(String message); // 保存消息
    void delete(String message); // 删除消息
}

文件系统存储实现

import java.io.*;

public class FileStorage implements Storage {
    private final String filePath;

    public FileStorage(String filePath) {
        this.filePath = filePath;
    }

    @Override
    public void save(String message) {
        try (BufferedWriter writer = new BufferedWriter(new FileWriter(filePath, true))) {
            writer.write(message);
            writer.newLine();
        } catch (IOException e) {
            throw new RuntimeException("Failed to save message", e);
        }
    }

    @Override
    public void delete(String message) {
        // 实现消息删除逻辑（如标记删除）
    }
}

生产者

public class Producer {
    private final MessageQueue queue;

    public Producer(MessageQueue queue) {
        this.queue = queue;
    }

    public void send(String message) {
        queue.publish(message);
    }
}

消费者

public class Consumer {
    private final MessageQueue queue;

    public Consumer(MessageQueue queue) {
        this.queue = queue;
    }

    public void start() {
        new Thread(() -> {
            while (true) {
                String message = queue.consume();
                if (message != null) {
                    process(message); // 处理消息
                    queue.ack(message); // 确认消息
                }
            }
        }).start();
    }

    private void process(String message) {
        System.out.println("Processing message: " + message);
    }
}

测试代码

public class Main {
    public static void main(String[] args) {
        Storage storage = new FileStorage("messages.txt");
        MessageQueue queue = new MessageQueue(storage);

        Producer producer = new Producer(queue);
        Consumer consumer = new Consumer(queue);

        consumer.start(); // 启动消费者

        // 生产者发送消息
        producer.send("Message 1");
        producer.send("Message 2");
        producer.send("Message 3");
    }
}

6. 扩展功能

消息重试机制

public class MessageQueue {
    private final BlockingQueue<String> queue;
    private final Storage storage;
    private final int maxRetries;

    public MessageQueue(Storage storage, int maxRetries) {
        this.queue = new LinkedBlockingQueue<>();
        this.storage = storage;
        this.maxRetries = maxRetries;
    }

    public void publish(String message) {
        try {
            storage.save(message);
            queue.put(message);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
    }

    public String consume() {
        try {
            return queue.take();
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            return null;
        }
    }

    public void ack(String message) {
        storage.delete(message);
    }

    public void retry(String message, int retryCount) {
        if (retryCount < maxRetries) {
            queue.offer(message); // 重新入队
        } else {
            // 进入死信队列
            System.out.println("Message moved to dead letter queue: " + message);
        }
    }
}