<?php
/*** 获取数据库连接** @param $host* @param $username* @param $password* @param $database* @return mysqli*/
function getDBConnection($host, $username, $password, $database){$connection = new mysqli('p:'.$host, $username, $password, $database);if (!$connection) {echo "Error: Unable to connect to MySQL." . PHP_EOL;echo "Debugging errno: " . mysqli_connect_errno() . PHP_EOL;echo "Debugging error: " . mysqli_connect_error() . PHP_EOL;exit;}mysqli_query($connection, "set names 'utf8'");return $connection;
}/*** 获取Redis连接** @param $host* @param $port* @param string $password* @param int $database* @return Redis*/
function getRedis($host='127.0.0.1', $port='6379', $password=null, $database=0){$redis = new Redis();if(!$redis->connect($host, $port)){die("Redis连接失败:IP或端口有误");}if(!empty($password) && !$redis->auth($password)){die("Redis连接失败:密码错误");}if($database){$redis->select($database);}// work中 subscribe 如果一段时间没有接到消息，就会停掉然后停掉，所以加这个语句让其永不超时$redis->setOption(Redis::OPT_READ_TIMEOUT, -1);return $redis;
}/*** 打印消息日志** @param $msg*/
function stdout($msg=null){$msg = '['.date('Y-m-d H:i:s').']'.$msg.chr(10);;fwrite(STDOUT, $msg);
}

LRANGE register_users 0 -1

<?php
require './lib.php';
$name = $argv[1];
$mobile = $argv[2];
if(empty($name) || empty($mobile)){die("参数错误");
}
// $connection = getDBConnection('localhost:3306', 'root', 'root', 'blog');
// // 开启事务
// mysqli_begin_transaction($connection);
// $sql = "insert into mq_user(name, mobile) values ('$name', '$mobile')";
// if(!mysqli_query($connection, $sql)){
//     die("写入用户信息失败,原因:".$connection->error);
// }
$redis = getRedis();
// 添加消息
$result = $redis->lpush('register_users', json_encode(array('name'=>$name, 'mobile'=>$mobile), JSON_UNESCAPED_UNICODE));
if($result === false){mysqli_rollback($connection);die("添加消息队列失败");
}
// 发布消息
$redis->publish('register_success', 'ok');
// 所有操作完成后提交事务
// mysqli_commit($connection);
// $connection->close();
$redis->close();

<?php
require './lib.php';
$redis = getRedis();
$redis->subscribe(['register_success'], function ($instance, $channelName, $message) {if($channelName == "register_success" && $message = "ok") {$redis = getRedis();while($redis->lsize("register_users")>0) {$arr = $redis->brPop(['register_users'], 20);if(count($arr)) {$userInfo = json_decode($arr[1], true);stdout("新注册用户信息:");stdout("姓名:".$userInfo['name']);stdout("手机号:".$userInfo['mobile']);stdout();sleep(3);}}}
});