消息队列（redis）-订单处理

Daichongweb

2019-09-16

一、简单的介绍

Redis是一个开源的使用ANSI C语言编写、支持网络、可基于内存亦可持久化的日志型、Key-Value数据库，并提供多种语言的API。
MySQLi是一个开源使用的 MySQL连接扩展，用于对数据库的操作和管理。
php-redis是一个开源的使用C语言编写的php扩展，用于php连接、管理redis。

二、业务介绍

某商城搞促销活动，会有大量的订单涌入数据库，每个订单15分钟未支付即为过期订单并且不能继续支付。

二、需求分析

很显然此业务需求在不使用redis队列的情况下，大量的订单并发会造成阻塞，甚至有可能超时（php默认的超时时间为30s）。

那么就会有以下步骤：
（1）用户下单后把数据直接写入消息队列（list）而不是数据库，写入队列成功即返回用户成功消息。
（2）创建订单队列：启动脚本来处理订单队列，逐一写入数据库，并设置一个key为订单号的hash过期时间为15分钟。
（3）用户支付时首先拿这个订单号去redis里查是存在，如存在订单状态为正常状态，否则为过期状态。
（4）订单支付成功队列：支付成功后，把订单数据写入另一个队列中，并启动脚本处理订单状态都为‘支付成功’的队列（如发送消息，修改订单状态等操作）。

注意：这期间可能会出现数据量过大，订单创建的队列尚未执行完毕，订单支付成功后查找该订单却未找到(lpop)，那么就需要把这个未找到的订单重新添加到队列的最后（rpush）。

三、代码实现
（1）创建订单：create_order.php

<?php
include "./RedisManager.php";

/**
 * 模拟下单
 */
$post = $_POST;
if (!empty($post)) {
    // 声明状态
    $status = 0;
    $msg = 'error';

    $order = [
        'order_id' => 'HD' . time(),
        'name' => $post['name'] . '_' . $post['id'],
        'price' => $post['price'],
        'date' => date('Y-m-d H:i:s')
    ];

    // 用户下单但未支付加入队列等待支付
    return createOrder($order);
}

/**
 * 创建订单 createOrder
 *
 * @return void
 */
function createOrder($order)
{
    try {
        $orderList = 'order:temp:list'; // 临时订单列表
        // 把订单写入队列 守护进程启动一个 watch_order.php 来处理订单消息
        $result = RedisConnect::getRedis()->lpush($orderList, json_encode($order));

        // 记录日志
        if (!$result) {
            createLog('订单写入队列失败', $order);
            returnMsg('订单写入队列失败');
        }
        returnMsg('下单成功', $order);
    } catch (Exception $e) {
        createLog($e->getMessage(), $order);
        returnMsg($e->getMessage());
    }
}

/**
 * 写入日志 createLog
 *
 * @param string $msg
 * @param array $data
 * @return void
 */
function createLog(string $msg, array $data = [])
{
    $orderMsg = 'order:msg'; // 订单消息
    $data['errorMsg'] = $msg;
    $data['errorTime'] = date('Y-m-d H:i:s');

    RedisConnect::getRedis()->select(1);
    RedisConnect::getRedis()->lPush($orderMsg, json_encode($data));
}

/**
 * 输出消息 returnMsg
 *
 * @param string $msg
 * @param array $data
 * @return void
 */
function returnMsg(string $msg, array $data = [])
{
    echo json_encode([
        'status' => 200,
        'message' => $msg,
        'data' => $data
    ]);
    exit;
}

（2）订单入库insert_db.php（以守护进程启动）

<?php
ini_set('default_socket_timeout', -1);
include './RedisManager.php';
include './MysqlManager.php';

$orderExpire = 'order:temp:list';
while (true) {
	try {
        $orderList = 'order:temp:list'; // 临时订单列表
		if (RedisConnect::getRedis()->lSize($orderList) < 1) {
			var_dump('订单已处理完毕') . '\n';
			sleep(1);
		}
		// 处理消息
		$order = RedisConnect::getRedis()->lPop($orderList);

		if ($order) {
			$order = json_decode($order, true);
			$order_id = $order['order_id'];
			$name = $order['name'];
			$price = $order['price'];
			$sql = "INSERT INTO `order`(`order_id`,`name`,`price`) VALUES('{$order_id}','{$name}',{$price})";
			$result = MysqlConnect::getMysql()->query($sql);

			//给每一个订单设置一个过期时间
			$expireResult = RedisConnect::getRedis()->hMset($orderExpire . ':' . $order['order_id'], $order);
			RedisConnect::getRedis()->expire($orderExpire . ':' . $order['order_id'], 900);
			var_dump($result) . '\n';
		} else {
			sleep(1);
		}
	} catch (Exception $e) {
		echo $e->getMessage();
	}
}

（3）监听过期的订单watch_order.php（以守护进程启动）

<?php
ini_set('default_socket_timeout', -1);

include './RedisManager.php';
include './MysqlManager.php';

RedisConnect::getRedis()->psubscribe(array('__keyevent@0__:expired'), 'keyCallback');

function keyCallback($redis, $pattern, $channel, $message)
{
    // 把过期的订单写入队列进行处理（这里可以处理为不同的订单写入不同的队列）
    $msgType = [
        'HD' => 'home:order', // 住房订单
        'DD' => 'ding:order' // 餐饮订单
    ];

    $keyArr = explode(':', $message);
    $key = array_pop($keyArr);
    $id = substr($key, 0, 2);
    switch($id){
        case 'HD':
            $sql = "UPDATE `order` SET `status` = 1 WHERE `order_id` = '{$key}'";
        break;
    }
    $result = MysqlConnect::getMysql()->query($sql);
    var_dump($result);
}