发布于 由admin

Redis Stream在PHP中的应用：构建可靠的消息队列与消费者组消费

Redis Stream在PHP中的应用：构建可靠的消息队列与消费者组消费

大家好，今天我们来深入探讨一下 Redis Stream 在 PHP 中的应用，重点是如何利用它构建可靠的消息队列，并通过消费者组实现高效的并发消费。

1. Redis Stream 简介

Redis Stream 是 Redis 5.0 版本引入的一个强大的数据结构，它提供了一种持久化的、可追加的消息日志，非常适合构建可靠的消息队列。 与传统的发布/订阅模式相比，Stream 提供了更强的持久性和灵活性，它支持：

  • 消息持久化: 消息会被持久化存储，即使消费者离线，消息也不会丢失。
  • 消息顺序保证: 消息按照生产者添加的顺序存储，保证了消息的有序性。
  • 消费者组: 支持多个消费者组成一个组，共同消费 Stream 中的消息，实现并发处理。
  • 确认机制: 消费者可以对已消费的消息进行确认，确保消息被正确处理。
  • 范围查询: 可以根据消息 ID 范围查询消息。
  • 阻塞读取: 消费者可以阻塞等待新消息的到来。

2. 基本概念

在深入代码之前，我们需要了解一些关键概念：

  • Stream (流): 一个有序的消息序列，可以看作一个日志文件。 每个消息都有一个唯一的 ID。
  • Message (消息): Stream 中存储的实际数据，通常是一个键值对的集合。
  • Entry ID (消息 ID): Stream 中每个消息的唯一标识符，通常是 timestamp-sequence 的格式，例如 1678886400000-1。
  • Consumer Group (消费者组): 一组消费者的集合，共同消费 Stream 中的消息。
  • Consumer (消费者): 消费者组中的一个成员，负责处理分配给它的消息。
  • Pending Entries List (PEL, 待处理消息列表): 记录了已发送给消费者但尚未被确认的消息。

3. PHP 中使用 Redis Stream 的准备工作

首先，确保你的 PHP 环境已经安装了 Redis 扩展。 如果没有安装，可以使用以下命令安装：

pecl install redis

然后在 php.ini 文件中启用 Redis 扩展：

extension=redis.so

接下来，我们需要一个 Redis 客户端。 PHP 官方提供了 Redis 类，我们可以直接使用。

<?php

$redis = new Redis();
$redis->connect('127.0.0.1', 6379); // 连接 Redis 服务器

if ($redis->ping()) {
    echo "连接 Redis 服务器成功!n";
} else {
    echo "连接 Redis 服务器失败!n";
}

?>

4. 生产者：向 Stream 中添加消息

生产者负责向 Stream 中添加消息。 在 Redis 中，我们使用 XADD 命令来实现。 在 PHP 中，对应的函数是 xAdd()。

<?php

$redis = new Redis();
$redis->connect('127.0.0.1', 6379);

$streamKey = 'my_stream'; // Stream 的键名
$message = [
    'user_id' => 123,
    'event' => 'product_viewed',
    'product_id' => 456,
    'timestamp' => time()
];

$messageId = $redis->xAdd($streamKey, '*', $message);

if ($messageId) {
    echo "成功添加消息到 Stream，消息 ID: " . $messageId . "n";
} else {
    echo "添加消息到 Stream 失败!n";
}

?>

代码解释：

  • $streamKey = 'my_stream';: 定义 Stream 的键名，可以根据实际业务场景自定义。
  • $message: 定义要添加到 Stream 中的消息，是一个关联数组，包含了 user_id、event、product_id 和 timestamp 等信息。
  • $redis->xAdd($streamKey, '*', $message);: 使用 xAdd() 函数向 Stream 中添加消息。
    • $streamKey: Stream 的键名。
    • '*': 表示让 Redis 自动生成消息 ID。 也可以指定一个特定的消息 ID，但通常让 Redis 自动生成更方便。
    • $message: 要添加的消息内容。
  • $messageId: xAdd() 函数返回新添加消息的 ID。

批量添加消息:

为了提高效率，我们可以批量添加消息：

<?php

$redis = new Redis();
$redis->connect('127.0.0.1', 6379);

$streamKey = 'my_stream';
$messages = [
    ['user_id' => 1, 'event' => 'order_placed', 'product_id' => 101],
    ['user_id' => 2, 'event' => 'product_viewed', 'product_id' => 102],
    ['user_id' => 1, 'event' => 'cart_updated', 'product_id' => 103],
];

foreach ($messages as $message) {
    $message['timestamp'] = time(); // 添加时间戳
    $messageId = $redis->xAdd($streamKey, '*', $message);
    if (!$messageId) {
        echo "添加消息到 Stream 失败!n";
    } else {
         echo "成功添加消息到 Stream，消息 ID: " . $messageId . "n";
    }
}

echo "批量添加消息完成!n";
?>

5. 消费者组：创建和消费消息

消费者组允许我们创建多个消费者共同消费 Stream 中的消息，实现并发处理。

5.1 创建消费者组

我们需要使用 XGROUP CREATE 命令来创建消费者组。 在 PHP 中，对应的函数是 xGroup()， 并且需要指定操作为 CREATE。

<?php

$redis = new Redis();
$redis->connect('127.0.0.1', 6379);

$streamKey = 'my_stream';
$groupName = 'my_group';
$startId = '0'; // 从 Stream 的起始位置开始消费

try {
    $redis->xGroup('CREATE', $streamKey, $groupName, $startId, true); // 创建消费者组，并设置创建时自动创建Stream
    echo "消费者组 " . $groupName . " 创建成功!n";
} catch (Exception $e) {
    echo "创建消费者组失败: " . $e->getMessage() . "n";
}

?>

代码解释：

  • $groupName = 'my_group';: 定义消费者组的名称。
  • $startId = '0';: 指定从 Stream 的哪个位置开始消费。 '0' 表示从 Stream 的起始位置开始消费。 '$' 表示从最新的消息开始消费。
  • $redis->xGroup('CREATE', $streamKey, $groupName, $startId, true);: 使用 xGroup() 函数创建消费者组。
    • 'CREATE': 指定操作类型为创建消费者组。
    • $streamKey: Stream 的键名。
    • $groupName: 消费者组的名称。
    • $startId: 从哪个位置开始消费。
    • true: 如果Stream不存在，则自动创建。

5.2 消费者消费消息

消费者使用 XREADGROUP 命令从 Stream 中读取消息。在 PHP 中，我们使用 xReadGroup() 函数。

<?php

$redis = new Redis();
$redis->connect('127.0.0.1', 6379);

$streamKey = 'my_stream';
$groupName = 'my_group';
$consumerName = 'consumer_1';
$count = 5; // 每次读取的消息数量
$block = 10000; // 阻塞等待的时间，单位是毫秒

while (true) {
    $messages = $redis->xReadGroup($groupName, $consumerName, [$streamKey => '>'], $count, $block);

    if ($messages) {
        foreach ($messages[$streamKey] as $messageId => $message) {
            echo "消费者 " . $consumerName . " 消费了消息，消息 ID: " . $messageId . ", 内容: " . json_encode($message) . "n";

            // 处理消息...
            processMessage($message);

            // 确认消息
            $redis->xAck($streamKey, $groupName, [$messageId]);
            echo "消息 " . $messageId . " 已确认!n";
        }
    } else {
        echo "没有新消息，继续等待...n";
    }
}

function processMessage($message) {
    // 这里编写处理消息的逻辑
    sleep(1); // 模拟处理消息的时间
    echo "处理消息完成!n";
}

?>

代码解释：

  • $consumerName = 'consumer_1';: 定义消费者的名称，在同一个消费者组内，每个消费者必须有唯一的名称。
  • $count = 5;: 每次读取的消息数量。
  • $block = 10000;: 阻塞等待的时间，单位是毫秒。 如果设置为 0，则立即返回，如果没有消息则返回空数组。
  • $redis->xReadGroup($groupName, $consumerName, [$streamKey => '>'], $count, $block);: 使用 xReadGroup() 函数从 Stream 中读取消息。
    • $groupName: 消费者组的名称。
    • $consumerName: 消费者的名称。
    • [$streamKey => '>']: 指定要读取的 Stream 和起始位置。 '>' 表示读取消费者组中尚未被消费的消息。
    • $count: 每次读取的消息数量。
    • $block: 阻塞等待的时间。
  • $redis->xAck($streamKey, $groupName, [$messageId]);: 使用 xAck() 函数确认消息。 确认消息后，该消息将从 PEL 中移除。

多个消费者:

可以启动多个消费者实例，并使用相同的 $streamKey 和 $groupName，但每个消费者必须使用不同的 $consumerName，这样就可以实现多个消费者并发地消费 Stream 中的消息。

5.3 处理 Pending Entries (PEL)

如果消费者在处理消息的过程中崩溃，消息将不会被确认，这些未被确认的消息会留在 PEL 中。 当消费者重新启动时，需要处理这些 PEL 中的消息。

<?php

$redis = new Redis();
$redis->connect('127.0.0.1', 6379);

$streamKey = 'my_stream';
$groupName = 'my_group';
$consumerName = 'consumer_1';
$count = 5;
$block = 10000;
$retryInterval = 60000; // 重试间隔，单位是毫秒

while (true) {
    // 1. 先尝试读取 PEL 中的消息
    $pendingMessages = $redis->xReadGroup($groupName, $consumerName, [$streamKey => '0'], $count, 0); // 从 PEL 中读取，不阻塞

    if ($pendingMessages) {
        foreach ($pendingMessages[$streamKey] as $messageId => $message) {
            // 检查消息是否已经超过重试间隔
            $messageTimestamp = explode('-', $messageId)[0];
            if ((time() * 1000 - $messageTimestamp) > $retryInterval) {
                echo "消费者 " . $consumerName . " 重新处理 PEL 中的消息，消息 ID: " . $messageId . ", 内容: " . json_encode($message) . "n";
                processMessage($message);
                $redis->xAck($streamKey, $groupName, [$messageId]);
                echo "消息 " . $messageId . " 已确认!n";
            } else {
                echo "消息 " . $messageId . " 还在重试间隔内，稍后重试n";
            }
        }
    }

    // 2. 再读取新的消息
    $newMessages = $redis->xReadGroup($groupName, $consumerName, [$streamKey => '>'], $count, $block);

    if ($newMessages) {
        foreach ($newMessages[$streamKey] as $messageId => $message) {
            echo "消费者 " . $consumerName . " 消费了新消息，消息 ID: " . $messageId . ", 内容: " . json_encode($message) . "n";
            processMessage($message);
            $redis->xAck($streamKey, $groupName, [$messageId]);
            echo "消息 " . $messageId . " 已确认!n";
        }
    } else {
        echo "没有新消息，继续等待...n";
    }
}

function processMessage($message) {
    // 这里编写处理消息的逻辑
    sleep(1); // 模拟处理消息的时间
    echo "处理消息完成!n";
}

?>

代码解释：

  • 首先尝试从 PEL 中读取消息，使用 xReadGroup() 函数，并指定起始位置为 '0' (表示从 Stream 的起始位置开始读取，实际上只会读取 PEL 中的消息，因为已经确认的消息不会再被读取)。
  • 检查消息的时间戳，如果消息的时间戳距离当前时间超过了重试间隔 $retryInterval，则重新处理该消息。
  • 如果 PEL 中没有消息，或者消息还在重试间隔内，则读取新的消息，使用 xReadGroup() 函数，并指定起始位置为 '>'。

5.4 查看消费者组信息

我们可以使用 XINFO GROUPS 命令查看消费者组的信息。 在 PHP 中，使用 xInfo()函数，并指定操作为GROUPS。

<?php

$redis = new Redis();
$redis->connect('127.0.0.1', 6379);

$streamKey = 'my_stream';

$groupInfo = $redis->xInfo('GROUPS', $streamKey);

if ($groupInfo) {
    print_r($groupInfo);
} else {
    echo "获取消费者组信息失败!n";
}

?>

5.5 查看 Stream 信息

我们可以使用 XINFO STREAM 命令查看 Stream 的信息，例如 Stream 的长度、第一个消息的 ID、最后一个消息的 ID 等。 在 PHP 中，使用 xInfo() 函数，并指定操作为STREAM。

<?php

$redis = new Redis();
$redis->connect('127.0.0.1', 6379);

$streamKey = 'my_stream';

$streamInfo = $redis->xInfo('STREAM', $streamKey);

if ($streamInfo) {
    print_r($streamInfo);
} else {
    echo "获取 Stream 信息失败!n";
}

?>

5.6 删除消费者组

当我们不再需要某个消费者组时，可以使用 XGROUP DESTROY 命令来删除它。在 PHP 中，使用 xGroup()函数，并指定操作为DESTROY。

<?php

$redis = new Redis();
$redis->connect('127.0.0.1', 6379);

$streamKey = 'my_stream';
$groupName = 'my_group';

$result = $redis->xGroup('DESTROY', $streamKey, $groupName);

if ($result) {
    echo "消费者组 " . $groupName . " 删除成功!n";
} else {
    echo "删除消费者组失败!n";
}

?>

6. 实际应用场景

Redis Stream 可以应用于各种需要可靠消息队列的场景，例如：

  • 异步任务处理: 将耗时的任务放入 Stream 中，由消费者异步处理，提高系统响应速度。
  • 事件驱动架构: 将系统中的事件发布到 Stream 中，由订阅这些事件的消费者进行处理，实现系统组件之间的解耦。
  • 数据同步: 将数据库的变更记录放入 Stream 中，由消费者同步到其他数据库或缓存。
  • 日志收集: 将应用程序的日志放入 Stream 中，由消费者进行分析和存储。

例如： 电商订单处理

  1. 用户下单后，将订单信息放入 Stream 中。
  2. 订单服务、库存服务、支付服务等作为消费者，分别处理订单相关的业务逻辑。
  3. 如果某个服务处理失败，可以重新从 PEL 中读取订单信息进行重试。

7. 性能优化

  • 批量操作: 使用 xAdd() 函数批量添加消息，使用 xReadGroup() 函数批量读取消息，可以减少网络开销，提高性能。
  • 连接池: 使用 Redis 连接池，避免频繁地创建和关闭连接。
  • 合理设置阻塞时间: 根据实际情况调整 xReadGroup() 函数的阻塞时间，避免消费者长时间空转。
  • 监控和告警: 监控 Stream 的长度、消费者组的活跃度、PEL 的大小等指标，及时发现和解决问题。

8. 总结

Redis Stream 提供了一种强大的、可靠的、可扩展的消息队列解决方案。 通过合理地使用 Stream 和消费者组，我们可以构建出高效的、健壮的异步处理系统。 理解其基本概念和关键命令，掌握在 PHP 中操作 Stream 的方法，可以帮助我们更好地利用 Redis Stream 来解决实际问题。

发表回复

您的邮箱地址不会被公开。 必填项已用 * 标注