Queue message confirmation mechanism and message retry processing method in PHP and MySQL

Message confirmation mechanism and message retry processing method of queue in PHP and MySQL

Introduction:
With the development of Internet applications, many online services A large number of requests need to be processed, and these requests often require an asynchronous processing method. Queues are a common solution that can effectively decouple requests from processing, improving system performance and reliability. This article will introduce the message confirmation mechanism and message retry processing method of queues in PHP and MySQL, and give specific code examples.

1. The concept and function of message queue
Message queue is a common application mode, which stores messages in the queue and then processes them asynchronously. The benefits of message queues are mainly reflected in the following aspects:

1. Decoupling: Decoupling requests and processing improves the scalability and maintainability of the system.
2. Asynchronous processing: Put time-consuming operations in the queue for asynchronous processing to improve the system's response speed.
3. Fault-tolerant mechanism for failed processing: Ensure the reliability of message processing through message confirmation and message retry mechanisms.

2. Message confirmation mechanism
In the queue system, message confirmation is a mechanism to ensure the completion of message processing. The message acknowledgment mechanism helps avoid the problem of message loss or duplicate processing.

Message confirmation in PHP can be achieved by using the ACK mechanism. The specific implementation steps are as follows:

1. The producer sends messages to the queue.
2. Consumers take out messages from the queue for processing.
3. If the message processing is successful, the consumer sends ACK to confirm that the message processing is completed; otherwise, the consumer sends NACK to reject the message processing.
4. The queue acknowledges receipt of the ACK or NACK of the message and deletes the ACK message from the queue.

The following is a sample code using RabbitMQ as a message queue:
Producer:

#!/usr/bin/env php
<?php
require_once __DIR__ . '/../vendor/autoload.php';

use PhpAmqpLibConnectionAMQPStreamConnection;
use PhpAmqpLibMessageAMQPMessage;

$connection = new AMQPStreamConnection('localhost', 5672, 'guest', 'guest');
$channel = $connection->channel();
$channel->queue_declare('queue_name', false, false, false, false);

$msg = new AMQPMessage('Hello World!');
$channel->basic_publish($msg, '', 'queue_name');

$channel->close();
$connection->close();

Consumer:

#!/usr/bin/env php
<?php
require_once __DIR__ . '/../vendor/autoload.php';

use PhpAmqpLibConnectionAMQPStreamConnection;
use PhpAmqpLibMessageAMQPMessage;

$connection = new AMQPStreamConnection('localhost', 5672, 'guest', 'guest');
$channel = $connection->channel();
$channel->queue_declare('queue_name', false, false, false, false);

$callback = function (AMQPMessage $msg) {
    echo 'Received message: ' . $msg->body . PHP_EOL;
    if (processMessage($msg)) {
        $msg->delivery_info['channel']->basic_ack($msg->delivery_info['delivery_tag']); // 消息处理成功，发送ACK确认
    } else {
        $msg->delivery_info['channel']->basic_nack($msg->delivery_info['delivery_tag'], false, true); // 消息处理失败，发送NACK拒绝
    }
};

$channel->basic_consume('queue_name', '', false, false, false, false, $callback);
while (count($channel->callbacks)) {
    $channel->wait();
}

function processMessage(AMQPMessage $msg) {
    // 消息处理逻辑
    if ($msg->body == 'Hello World!') {
        return true;
    } else {
        return false;
    }
}

$channel->close();
$connection->close();

3. Message retry Processing method
In actual applications, message processing may fail, such as network failure, server error, etc. In order to ensure the reliability of messages, messages that fail to be processed can be retried.

MySQL provides transaction and rollback mechanisms, which can be applied in message retry processing. The specific implementation steps are as follows:

1. The producer sends messages to the message table in the database.
2. Consumers retrieve messages from the message table in the database for processing.
3. If the message is processed successfully, delete the message from the message table; otherwise, add one to the number of processing times in the message table and update the processing time.
4. Set up a scheduled task to regularly check the messages in the message table that have been processed less than or equal to the maximum number of retries, and re-deliver them to the consumer.

The following is a sample code that uses MySQL as a message store:
Producer:

<?php
$dsn = 'mysql:dbname=testdb;host=127.0.0.1';
$user = 'root';
$password = '';

try {
    $db = new PDO($dsn, $user, $password);
    $db->setAttribute(PDO::ATTR_ERRMODE, PDO::ERRMODE_EXCEPTION);

    $sql = 'INSERT INTO message_queue (message) VALUES (?)';
    $stmt = $db->prepare($sql);
    $message = 'Hello World!';
    $stmt->bindParam(1, $message);
    $stmt->execute();

} catch (PDOException $e) {
    echo 'Connection failed: ' . $e->getMessage();
}

Consumer:

<?php
$dsn = 'mysql:dbname=testdb;host=127.0.0.1';
$user = 'root';
$password = '';

try {
    $db = new PDO($dsn, $user, $password);
    $db->setAttribute(PDO::ATTR_ERRMODE, PDO::ERRMODE_EXCEPTION);

    $sql = 'SELECT * FROM message_queue';
    $stmt = $db->prepare($sql);
    $stmt->execute();
    $messages = $stmt->fetchAll();

    foreach ($messages as $message) {
        if (processMessage($message)) {
            $deleteSql = 'DELETE FROM message_queue WHERE id = ?';
            $deleteStmt = $db->prepare($deleteSql);
            $deleteStmt->bindParam(1, $message['id']);
            $deleteStmt->execute();
        } else {
            $retrySql = 'UPDATE message_queue SET retries = retries + 1, last_retry_time = ? WHERE id = ?';
            $retryStmt = $db->prepare($retrySql);
            $now = date('Y-m-d H:i:s');
            $retryStmt->bindParam(1, $now);
            $retryStmt->bindParam(2, $message['id']);
            $retryStmt->execute();
        }
    }

} catch (PDOException $e) {
    echo 'Connection failed: ' . $e->getMessage();
}

function processMessage($message) {
    // 消息处理逻辑
    if ($message['message'] == 'Hello World!') {
        return true;
    } else {
        return false;
    }
}

Scheduled task:

<?php
$dsn = 'mysql:dbname=testdb;host=127.0.0.1';
$user = 'root';
$password = '';

try {
    $db = new PDO($dsn, $user, $password);
    $db->setAttribute(PDO::ATTR_ERRMODE, PDO::ERRMODE_EXCEPTION);

    $sql = 'SELECT * FROM message_queue WHERE retries <= ?';
    $stmt = $db->prepare($sql);
    $maxRetries = 3;
    $stmt->bindParam(1, $maxRetries);
    $stmt->execute();
    $messages = $stmt->fetchAll();

    foreach ($messages as $message) {
        // 重新投递消息给消费者
    }

} catch (PDOException $e) {
    echo 'Connection failed: ' . $e->getMessage();
}

Conclusion:
Through the message confirmation mechanism and message retry processing method, we can improve the reliability and stability of the system. As a common decoupling and asynchronous processing tool, queues can effectively implement message confirmation and retry in PHP and MySQL, providing better performance and user experience for our applications.

References:

1. PHP RabbitMQ official documentation: https://github.com/php-amqplib/php-amqplib
2. PHP MySQL official documentation: https ://www.php.net/manual/en/book.pdo.php

The above is the detailed content of Queue message confirmation mechanism and message retry processing method in PHP and MySQL. For more information, please follow other related articles on the PHP Chinese website!