面试题：Redis消息队列在MySQL文件上传系统中如何保障消息的可靠性

保障消息不丢失的方法及利用Redis特性实现方式

生产者端
- 持久化发送消息：在将消息发送到Redis队列前，先将消息持久化到本地文件或数据库。利用Redis的原子性操作，使用 LPUSH 等命令将消息安全地添加到队列中。即使生产者崩溃，也可从持久化存储中恢复消息重新发送。
- 发送确认机制：生产者发送消息后，等待Redis返回确认信息。Redis的响应机制可确保消息成功入队。若未收到确认，可进行重试。例如在Python中使用 redis - py 库发送消息时：

import redis

r = redis.Redis(host='localhost', port=6379, db = 0)
message = "file_upload_info"
result = r.lpush('upload_queue', message)
if result:
    print("Message sent successfully")
else:
    print("Failed to send message, retry...")

Redis自身
- AOF和RDB持久化：
  - RDB：Redis可定期将内存中的数据快照保存到磁盘，利用 SAVE 或 BGSAVE 命令。当Redis重启时，可通过加载RDB文件恢复数据，包括消息队列中的数据。
  - AOF：以日志形式记录Redis执行的写操作，使用 appendfsync 配置参数控制刷盘策略。选择 always 策略可保证每条写命令都同步到磁盘，最大程度保障消息不丢失。配置示例：

appendonly yes
appendfsync always

- **主从复制**：通过设置主从复制，主节点将数据同步到从节点。即使主节点故障，从节点可晋升为主节点继续提供服务，消息队列数据不会丢失。配置从节点时，在从节点的 `redis.conf` 文件中设置 `replicaof <masterip> <masterport>`。

3. 消费者端 - 手动确认机制：消费者从Redis队列取出消息后，不立即删除，而是在成功处理完消息（如完成MySQL文件上传）后，再使用 RPOP 或 LREM 命令从队列中删除消息。例如在Java中使用Jedis库：

Jedis jedis = new Jedis("localhost", 6379);
String message = jedis.lpop("upload_queue");
if (message != null) {
    try {
        // 处理文件上传到MySQL逻辑
        System.out.println("Processed message: " + message);
        jedis.lrem("upload_queue", 1, message);
    } catch (Exception e) {
        // 处理失败，消息保留在队列，可进行重试
        System.out.println("Failed to process message: " + message);
    }
}

- **设置合理的重试机制**：当处理消息失败时，利用Redis的 `ZSET` 数据结构实现延迟重试队列。将失败消息及其重试时间戳存入 `ZSET`，消费者定期检查 `ZSET` 中达到重试时间的消息，重新处理。例如在Python中：

import redis
import time

r = redis.Redis(host='localhost', port=6379, db = 0)
failed_message = "failed_upload_info"
retry_time = time.time() + 60  # 60秒后重试
r.zadd('retry_queue', {failed_message: retry_time})

# 消费者检查重试队列
while True:
    current_time = time.time()
    messages = r.zrangebyscore('retry_queue', 0, current_time)
    for message in messages:
        try:
            # 处理文件上传到MySQL逻辑
            print("Retrying message: " + message.decode('utf - 8'))
            r.zrem('retry_queue', message)
        except Exception e:
            print("Retry failed for message: " + message.decode('utf - 8'))
    time.sleep(10)

星途面试题库

面试题：Redis消息队列在MySQL文件上传系统中如何保障消息的可靠性

知识考点

面试题答案

保障消息不丢失的方法及利用Redis特性实现方式