基于redis+mysql+php的简单队列实现

消息队列在是分布式系统中必不可少的中间件，目前在生产环境，使用较多的消息队列有ActiveMQ，RabbitMQ，Kafka，RocketMQ等。然而对于一个要求不高的小型系统来说，单独使用维护这些队列系统代价有点大。而redis可以在做缓存的同时也满足队列的需求。redis的list是有序的列表，加上其出队入队函数，利用其特性很简单的就能实现一个消息队列。

一、业务层邮件入队列

入队列使用redis的lpush指令

LPUSH key value1 [value2]

将一个或多个值插入到列表头部

先将邮件详情写入mysql存储，再将该邮件id push到redis队列头部

function sendSysMail($userId, $type, $content)
{
$sysSmsLogModel = M('SysEmailLog');
$data = array(
'user_id' => $userId,
'content' => $content,

'add_time' => time(),

'date_time' => date('Y-m-d H:i:s', time()),
'add_ip' => get_client_ip(),
);
if($sysSmsLogModel->create($data))

{

$emailId = $sysSmsLogModel->add(); //邮件ID

//添加邮件队列
$redis = new Redis();

$redis->connect(C('REDIS_HOST'), C('REDIS_PORT'));

$redis->lPush('mailQueue', $emailId); //lPush将邮件ID入到队列头部

return TRUE;
}
else
{
return FALSE;
}
}

二、后台队列监听

后台执行一个PHP进程，监听队列，对队列循环进行出队列操作。一旦队列有数据，就从mysql中取对应的邮件详细情况进行发送。

出队操作使用BRPOP命令

BRPOP key1 [key2 ] timeout

移出并获取列表的最后一个元素，如果列表没有元素会阻塞列表直到等待超时或发现可弹出元素为止。

使用过程中发生了件奇怪的事情，后台进程开启后开始执行的好好的，并没有什么异常，但跑了一晚上，第二天再来看时便不正常了。

看了下redis的数据，都正常的出入队列了，但是邮件并未发送，日志中也没有发送失败的记录。

一番折腾，发现是mysql连接这块存在问题，mysql默认的连接超时时间是28800秒，即8小时，超过8小时，mysql连接就断开了，过了一晚上,mysql自然就go away了。

于是在没有队列没有数据的时候执行select 1来延长mysql连接的超时时间点，select 1执行异常时重新建立一个mysql的连接替换之前的连接，以此来保持mysql的连接始终可用。

<?php

//mailQueue.php

$config = include('config.php');

$redis = new Redis();

$redis->connect($config['REDIS_HOST'], $config['REDIS_PORT']);

$pdo= new PDO($config['DSN'], $config['DB_USER'], $config['DB_PWD'], array(PDO::ATTR_ERRMODE => PDO::ERRMODE_EXCEPTION));

while (true)

{

$task = $redis->brPop('mailQueue', 10); //brPop从队列尾部出队

if ($task)

{

$emailId = $task[1];

$sql = "SELECT a.*, b.email FROM sys_email_log AS a, user_main AS b WHERE a.user_id = b.user_id AND a.email_id = {$emailId}";

$res = $pdo->query($sql);

foreach ($res as $row)

{

$to = $row['email'];

$title = '邮件提醒';

$content = $row['content'];

$result = send($to, $title, $content); //send邮件发送函数，调用外部系统邮件接口

if ($result)

{

echo date("Y-m-d H:i:s", time()) . ": " . $emailId . "发送成功\n";

}

else

{

echo date("Y-m-d H:i:s", time()) . ": " . $emailId . "发送失败\n";

}

else

{

try

{

$pdo->query("SELECT 1");

echo date("Y-m-d H:i:s", time()) . ": 保持连接\n";

}

catch (PDOException $e)

{

$pdo = new PDO($config['DSN'], $config['DB_USER'], $config['DB_PWD'], array(

PDO::ATTR_ERRMODE => PDO::ERRMODE_EXCEPTION));

echo date("Y-m-d H:i:s", time()) . ": mysql重连\n";

}

三、队列监听程序的守护进程

用shell写的守护进程，方便队列监听程序的开启关闭以及状态查看

#!/bin/bash

#开启
function start()
{
#先检测程序是否已经开启
pid=`ps -ef | grep "php -f mailQueue.php" | grep -v grep | awk '{print $2}'`

if [ "$pid" == "" ]
then
php -f mailQueue.php >> mailLog &
echo "程序启动成功"
else
echo "程序已经开启过"
fi

}

#关闭
function stop()
{
pid=`ps -ef | grep "php -f mailQueue.php" | grep -v grep | awk '{print $2}'`

if [ "$pid" == "" ]
then
echo "程序未开启"
else
kill -9 $pid
echo "程序关闭成功"
fi
}

#查看开启状态
function status()
{
pid=`ps -ef | grep "php -f mailQueue.php" | grep -v grep | awk '{print $2}'`

if [ "$pid" == "" ]
then
echo "程序未开启"
else
echo "程序运行中,pid: $pid"
fi
}

#主程序
case "$1" in
"start")
start
;;

"stop" )
stop
;;

* )
echo "参数错误! Usage: mailQueue [start|stop|status]"
;;