Redis 研究

admin

`BRPOP` 指令详解

BRPOP 是 Redis List 的阻塞弹出命令，作用是从一个或多个指定的列表尾部弹出元素。如果所有指定的列表都为空，客户端会被阻塞，直到有元素可弹出或达到超时时间。

语法

BRPOP key [key ...] timeout

key [key ...]：一个或多个列表键名，Redis会按顺序检查这些列表。
timeout：阻塞等待的秒数，0 表示无限阻塞。

返回值

成功时返回一个两元素的数组：
- 第一个元素是弹出元素所在的列表键名；
- 第二个元素是被弹出的元素值。
超时或所有列表为空且无元素弹出时，返回 nil。

举例

假设有两个列表 queue1 和 queue2：

BRPOP queue1 queue2 5

此命令会先检查 queue1 的尾部是否有元素可弹出，如果有，立即返回该元素；如果没有，则检查 queue2，如果也没有元素，则阻塞最多5秒等待新元素入队。如果5秒内仍无元素，则返回 nil。

`BRPOP` 在消息队列中的应用

消费者阻塞等待：消费者使用 BRPOP 命令阻塞等待消息，避免轮询带来的资源浪费。
多队列监听：消费者可以同时监听多个队列，提高灵活性和容错能力。
先进先出保证：结合 BRPOP 弹出尾部元素，配合生产者使用 LPUSH 插入头部，确保消息顺序。

与 `BLPOP` 的对比

BLPOP 从列表头部弹出元素，适合生产者使用 RPUSH 插入尾部的场景。
BRPOP 从列表尾部弹出元素，适合生产者使用 LPUSH 插入头部的场景。

两者都支持阻塞等待，具体使用哪个，取决于生产者入队元素的位置。

总结

BRPOP 是 Redis List 实现阻塞消费的重要命令，适合需要消费者等待消息且支持多队列监听的场景。在构建简单高效的消息队列时，BRPOP 能帮助减少资源浪费并及时响应新消息。

admin

下面我给出一个基于 BRPOP 的简单消息消费者示例，演示如何阻塞等待队列中的消息并处理它们。

基于 `BRPOP` 的消息消费者示例（伪代码）

-- 假设使用Lua脚本调用Redis客户端库

local redis = require "redis"  -- 伪代码，需替换为实际Redis客户端
local client = redis.connect("127.0.0.1", 6379)

local queues = {"queue1", "queue2"}  -- 多个队列监听
local timeout = 0  -- 0表示无限阻塞

while true do
    -- BRPOP 返回格式: {queue_name, message} 或 nil（超时）
    local result = client:brpop(unpack(queues), timeout)
    if result then
        local queue_name = result[1]
        local message = result[2]

        print(string.format("从队列 %s 收到消息: %s", queue_name, message))
        -- 此处处理消息逻辑
        -- 例如：业务处理，写日志，调用其他服务等
    else
        -- 超时返回 nil，这里timeout=0一般不会执行到
        print("等待消息超时，无新消息")
    end
end

说明

brpop 命令会阻塞直到任意一个指定队列有消息可消费。
这样消费者线程不会空转轮询，节省资源。
可以监听多个队列，实现灵活的消息分发和消费。
业务代码写在收到消息后，保证及时响应。

admin

下面我先给出一个简单的生产者示例代码，然后介绍如何结合 RPOPLPUSH 来实现可靠消费和消息重试机制。

生产者示例（Lua伪代码）

local redis = require "redis"
local client = redis.connect("127.0.0.1", 6379)

local queue = "queue1"

-- 模拟发送消息函数
local function send_message(msg)
    -- 使用 LPUSH 将消息推入队列头部（生产者入队）
    client:lpush(queue, msg)
    print("消息已发送到队列：" .. msg)
end

-- 示例：发送几个消息
send_message("任务1")
send_message("任务2")
send_message("任务3")

结合 `RPOPLPUSH` 实现可靠消费的思路

RPOPLPUSH source destination 命令会从 source 列表尾部弹出一个元素，并将它推入到 destination 列表头部，原子操作。

应用场景

工作队列（Work Queue）模式中，消费者从主队列弹出消息后，先把消息放到一个“处理中”队列，防止消息丢失。
消费者处理完成后，再从“处理中”队列移除该消息，确认消费。
如果消费者处理失败或宕机，消息仍在“处理中”队列，可以通过定时检查“处理中”队列实现消息重试。

流程示意

生产者将消息使用 LPUSH 或 RPUSH 放入主队列。
消费者使用 RPOPLPUSH 主队列处理中队列，将消息从主队列尾部弹出并推入处理中队列头部。
消费者处理消息。
处理成功后，使用 LREM 处理中队列 1 消息 删除该消息，完成确认。
处理失败可决定是否将消息重新放回主队列或记录日志。

消费者可靠消费示例（Lua伪代码）

local redis = require "redis"
local client = redis.connect("127.0.0.1", 6379)

local main_queue = "queue1"
local processing_queue = "queue1:processing"

while true do
    -- 原子地从主队列尾部弹出一个消息并推入处理中队列头部
    local msg = client:rpoplpush(main_queue, processing_queue)
    
    if msg then
        print("开始处理消息: " .. msg)
        local success = false
        -- 处理消息 (示例：模拟处理)
        -- 这里写你的业务逻辑
        if math.random() > 0.2 then  -- 模拟80%成功率
            success = true
        end
        
        if success then
            -- 处理成功，删除处理中队列中的消息
            client:lrem(processing_queue, 1, msg)
            print("消息处理成功: " .. msg)
        else
            -- 处理失败，决定是否重试或记录日志
            print("消息处理失败，保留在处理中队列等待重试: " .. msg)
            -- 可选：休眠一段时间后重试，或者将消息移回主队列
        end
    else
        -- 主队列为空，休眠等待
        print("队列为空，等待新消息")
        os.execute("sleep 1")
    end
end

额外说明

通过 RPOPLPUSH 保证消息在处理时不会丢失，除非处理成功明确删除。
需要设计一个定时任务，扫描“处理中队列”，对长时间未确认的消息进行重试或告警，防止消息永久丢失。
结合 BRPOPLPUSH 命令还能实现阻塞等待，进一步优化性能。

admin

接下来我给你演示如何使用 BRPOPLPUSH 实现一个阻塞且可靠的消息消费者。相比之前的 RPOPLPUSH 方案，BRPOPLPUSH 支持阻塞等待消息，避免空轮询，进一步提升性能和资源利用率。

基于 `BRPOPLPUSH` 的阻塞可靠消费者示例（Lua伪代码）

local redis = require "redis"
local client = redis.connect("127.0.0.1", 6379)

local main_queue = "queue1"
local processing_queue = "queue1:processing"
local timeout = 0  -- 0 表示无限阻塞，等待消息

while true do
    -- 阻塞等待主队列有消息，将消息从主队列尾部弹出并推入处理中队列头部
    local msg = client:brpoplpush(main_queue, processing_queue, timeout)
    if msg then
        print("收到消息开始处理: " .. msg)
        local success = false

        -- 处理消息业务逻辑（这里示例随机模拟）
        if math.random() > 0.2 then
            success = true
        end

        if success then
            -- 处理成功，从处理中队列删除消息（确认消费）
            client:lrem(processing_queue, 1, msg)
            print("消息处理成功并确认: " .. msg)
        else
            -- 失败时不删除消息，保留在处理中队列等待重试
            -- 可以考虑休眠或记录日志，防止死循环
            print("消息处理失败，留在处理中队列: " .. msg)
            os.execute("sleep 1")  -- 简单休眠，防止CPU空转
        end
    else
        -- 理论上不会到这里，除非网络断开或关闭连接
        print("BRPOPLPUSH 返回 nil，可能断开连接")
        os.execute("sleep 1")
    end
end

使用建议

阻塞等待：timeout=0 使消费者线程无限阻塞，只有当有消息时才执行后续逻辑，节省资源。
消息确认：处理成功后删除处理中队列消息，保证消息不丢失且不会重复消费。
消息重试：失败消息保留在处理中队列，结合定时任务扫描处理超时消息，重试或报警。
异常恢复：程序异常退出时，处理中队列中的消息未确认，可以通过后台任务将它们重新放回主队列，保证消息不丢失。

定时重试示例思路（伪代码）

-- 定时任务脚本，扫描 processing_queue
local function retry_failed_messages()
    local processing_queue = "queue1:processing"
    local main_queue = "queue1"
    
    -- 获取处理中队列全部消息（小队列适用）
    local msgs = client:lrange(processing_queue, 0, -1)
    
    for _, msg in ipairs(msgs) do
        -- 判断消息处理超时逻辑（这里简化为全部重试）
        -- 将消息重新推回主队列，供消费者重新消费
        client:lrem(processing_queue, 1, msg)
        client:rpush(main_queue, msg)
        print("消息重新放回主队列重试: " .. msg)
    end
end

admin

接下来介绍Redis 5.0及以后版本引入的更高级消息队列数据结构—— Redis Stream ，以及它的消费者组（Consumer Group）功能，这对构建复杂且可靠的消息队列系统非常有帮助。

一、Redis Stream 简介

Redis Stream 是一种日志型数据结构，类似于Kafka、RabbitMQ里的消息流，支持消息持久化、自动消息ID生成、消息排序和基于时间的消息消费。

主要特点：

消息ID自动生成，保证唯一且有序。
支持多个消费者组，并且组内消费者可协作消费消息。
内置消息确认机制，支持消息重试和消息追踪。
支持消息持久化，结合Redis的RDB/AOF机制。
支持阻塞式消费和消费组分组消费。

二、基本概念

Stream：消息流，类似一个有序的消息列表。
消费者组（Consumer Group）：一组消费者共同消费Stream消息，消息在组内分发给不同消费者。
消费者（Consumer）：消费者组中的成员，负责拉取和处理消息。
Pending Entries List (PEL)：未确认的消息列表，用于追踪已被消费者取出但未确认的消息。

三、核心命令示例

1. 生产者写入消息

XADD mystream * field1 value1 field2 value2

*表示自动生成消息ID。
field1 value1 是消息内容的键值对，支持多字段。

2. 创建消费者组

XGROUP CREATE mystream mygroup 0

0 表示从流的起始ID开始消费。
如果流不存在会自动创建。

3. 消费者读取消息

XREADGROUP GROUP mygroup consumer1 COUNT 1 BLOCK 5000 STREAMS mystream >

GROUP mygroup consumer1：指定消费组和消费者名称。
COUNT 1：最多读取1条消息。
BLOCK 5000：阻塞等待最多5秒。
STREAMS mystream >：>表示只获取新的待分发消息。

4. 消息确认

XACK mystream mygroup 1526985058136-0

消费者处理完消息后调用，确认消息已消费，消息会从Pending List移除。

四、消费者组的工作流程

消息写入Stream：生产者调用XADD。
消费者组创建：调用XGROUP CREATE创建消费组。
消费者读取消息：使用XREADGROUP阻塞或非阻塞拉取消息。
消息处理：消费者处理消息。
消息确认：调用XACK确认消息。
消息重试：通过检查XPENDING命令查看未确认消息，重试处理。

五、Redis Stream 相比传统 List 的优势

特性List + BRPOP/BRPOPLPUSHStream + Consumer Group 消息ID无自动唯一递增ID 消息确认需自定义（RPOPLPUSH方案）内置确认机制（XACK）消费者分组需人工管理原生支持多个消费者组，负载均衡消息持久化依赖Redis持久化机制依赖持久化机制，且支持消息追踪阻塞消费支持（BRPOP、BRPOPLPUSH）支持（XREADGROUP BLOCK）消息重试需要自己实现支持Pending List重试消息排序通过List顺序保证通过消息ID保证顺序

六、示例应用场景

需要多消费者并行处理消息的任务队列。
需要消息可靠传递、确认和可追踪的分布式系统。
复杂的事件流处理和日志收集。

七、简单示例流程

# 生产者写入消息
XADD mystream * task "email" user "alice"

# 创建消费者组（只需执行一次）
XGROUP CREATE mystream mygroup 0

# 消费者读取消息
XREADGROUP GROUP mygroup consumer1 COUNT 1 BLOCK 0 STREAMS mystream >

# 消费者处理完成后确认消息
XACK mystream mygroup <消息ID>

admin

下面我给出一个基于 Redis Stream 的完整示例，包括生产者和消费者两个部分，演示如何写入消息、创建消费者组、消费消息并确认处理。

依赖环境

此示例使用 Lua 伪代码，假设你有一个支持 Redis 命令的客户端库，例如 redis-lua、python-redis 或者其他语言的客户端，调用方式类似即可。

1. 生产者示例代码

local redis = require "redis"
local client = redis.connect("127.0.0.1", 6379)

local stream_key = "mystream"

-- 生产者发送消息函数
local function produce_message(task_type, user)
    -- XADD mystream * task <task_type> user <user>
    local message_id = client:xadd(stream_key, "*", "task", task_type, "user", user)
    print("已发送消息，ID: " .. message_id)
end

-- 示例发送几条消息
produce_message("email", "alice")
produce_message("sms", "bob")
produce_message("push", "carol")

2. 创建消费者组（只需运行一次）

你可以用命令行或者代码创建消费者组，下面是Lua调用示例：

local redis = require "redis"
local client = redis.connect("127.0.0.1", 6379)

local stream_key = "mystream"
local group_name = "mygroup"

-- 创建消费者组，如果已存在会报错，可以捕获忽略
local ok, err = pcall(function()
    client:xgroup("CREATE", stream_key, group_name, "0")
end)

if ok then
    print("消费者组创建成功")
else
    print("消费者组可能已存在，错误信息：" .. tostring(err))
end

3. 消费者示例代码

消费者会阻塞等待新消息，读取后处理并确认。

local redis = require "redis"
local client = redis.connect("127.0.0.1", 6379)

local stream_key = "mystream"
local group_name = "mygroup"
local consumer_name = "consumer1"
local block_time = 5000  -- 毫秒，阻塞等待时间

while true do
    -- XREADGROUP GROUP mygroup consumer1 COUNT 1 BLOCK 5000 STREAMS mystream >
    local streams = client:xreadgroup("GROUP", group_name, consumer_name,
                                     "COUNT", 1,
                                     "BLOCK", block_time,
                                     "STREAMS", stream_key, ">")

    if streams then
        -- streams 格式: { {stream_key, { {message_id, {field1, value1, ...}}, ... } } }
        for _, stream_data in ipairs(streams) do
            local key = stream_data[1]
            local messages = stream_data[2]
            for _, message in ipairs(messages) do
                local message_id = message[1]
                local fields = message[2]  -- 是一个数组，奇数位是字段名，偶数位是值

                -- 解析字段（示例只取task和user）
                local task_type, user
                for i = 1, #fields, 2 do
                    if fields[i] == "task" then
                        task_type = fields[i+1]
                    elseif fields[i] == "user" then
                        user = fields[i+1]
                    end
                end

                print(string.format("收到消息ID=%s，任务类型=%s，用户=%s", message_id, task_type, user))

                -- 模拟业务处理
                local success = true  -- 这里可以替换成实际处理逻辑及异常捕获

                if success then
                    -- 处理成功，确认消息
                    client:xack(stream_key, group_name, message_id)
                    print("消息已确认，ID=" .. message_id)
                else
                    -- 处理失败，可记录日志或重试
                    print("消息处理失败，ID=" .. message_id)
                end
            end
        end
    else
        -- 阻塞超时无消息
        print("无新消息，继续等待...")
    end
end

4. 说明

生产者通过 XADD 写入消息，消息自动带有ID。
消费者组只需创建一次，并且支持多消费者并发消费。
消费者通过 XREADGROUP 阻塞等待消息，拉取未分发的消息。
处理完成后调用 XACK 确认消息，消息才会被移出 Pending List。
如果消费者异常退出，未确认消息会一直留在 Pending List，可以通过 XPENDING 和 XCLAIM 命令实现消息重试。