队列在PHP与MySQL中的消息去重和消息幂等性的处理方法

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

function processMessage($message) {
    $messageId = generateUniqueId($message);
    $timestamp = time();

    // 判断消息是否已经存在
    $existingTimestamp = $redis->zscore('message:deduplication', $messageId);

    // 如果消息存在并且时间戳在一定范围内，则不进行处理
    if ($existingTimestamp && $timestamp - $existingTimestamp <= 60) {
        return;
    }

    // 处理消息
    // ...

    // 将消息的唯一标识和时间戳添加到有序集合中
    $redis->zadd('message:deduplication', $timestamp, $messageId);
}

function generateUniqueId($message) {
    // 生成消息的唯一标识
    // ...
    return $uniqueId;
}

<?php
$mysqli = new mysqli('localhost', 'username', 'password', 'database');

function processMessage($message) {
    $messageId = generateUniqueId($message);

    $sql = "INSERT IGNORE INTO message_deduplication (message_id) VALUES ('$messageId')";

    if ($mysqli->query($sql)) {
        // 插入成功，处理消息
        // ...
    } else {
        // 消息已经存在，不再处理
    }
}

function generateUniqueId($message) {
    // 生成消息的唯一标识
    // ...
    return $uniqueId;
}

<?php
$mysqli = new mysqli('localhost', 'username', 'password', 'database');

function processMessage($message) {
    $messageId = generateUniqueId($message);

    // 查询处理状态
    $sql = "SELECT status FROM message_processing WHERE message_id = '$messageId'";
    $result = $mysqli->query($sql);

    if ($result && $result->num_rows > 0) {
        $row = $result->fetch_assoc();
        $status = $row['status'];

        // 如果处理状态为已处理，则不再处理
        if ($status == 1) {
            return;
        }
    }

    // 处理消息
    // ...

    // 更新处理状态
    $sql = "INSERT INTO message_processing (message_id, status) VALUES ('$messageId', 1) ON DUPLICATE KEY UPDATE status = 1";
    $mysqli->query($sql);
}

function generateUniqueId($message) {
    // 生成消息的唯一标识
    // ...
    return $uniqueId;
}