PHP + Redis 实现简单消息队列-低调大师

PHP + Redis 实现简单消息队列

2020-08-15 679

Redis做消息队列的好处在于它的轻量级，高并发，延迟敏感，应用场景有即时数据分析、秒杀计数器、缓存等

Redis做消息队列待解决的问题：

1、消息的可靠性：没有相应的机制保证消息的消费，当消费者消费失败的时候，消息体丢失，需要手动处理。生产者只管向队列中插入数据，不管消费者是否成功消费。

2、消费者挂掉消息不会丢失，但是需要重新触发一下消费者，才能够继续消费消息。

代码如下：

lib.php 是工具文件，里面有数据库的连接、Redis的连接：

<?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);
}

register.php 是消息发布者，注释的是将消息存入数据库部分的代码。

首先想消息存入 register_users 队列中，存入的 key是register_users；value是一个list，消息全部存入其中。用 redis-cli 查看数据的命令是：

LRANGE register_users 0 -1

register.php：

<?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();

work.php 做为消息的消费者

<?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);
            }
        }
    }
});

register.php将消息放入redis 的 register_users队列中，然后再使用 publish 将 register_success 消息发不出去。work.php 使用 subscribe 订阅register_success 的消息。接收到 register_success 消息之后，读取 register_users 的消息进行处理。

微信关注我们

原文链接：https://my.oschina.net/u/4479011/blog/4496701

转载内容版权归作者及来源网站所有！

低调大师中文资讯倾力打造互联网数据资讯、行业资源、电子商务、移动互联网、网络营销平台。持续更新报道IT业界、互联网、市场资讯、驱动更新,是最及时权威的产业资讯及硬件资讯报道平台。

缓存雪崩和缓存穿透问题及解决方案----->(浅谈)

缓存雪崩: 缓存同一时间大面积的失败,所以后面的请求都会落到数据库上,造成数据库短时间内承受大量请求而崩掉解决方案: 事前:尽量保证整个redis集群的高可用性,发现机器宕机尽快补上事中:本地ehcache缓存+hystrix限流&降级,避免MySQL崩掉事后:利用redis持久化机制保存的数据尽快恢复缓存缓存穿透: 简答点说就是大量请求的Key根本不存在于缓存中,导致请求直接到了数据库上,根本没有经过缓存这一次层. 解决方案: 最基本的就是首先做好参数校验,一些不合法的参数请求直接抛出异常信息返回给客户端.比如查询的数据库id不能小于0,传入的邮箱格式不对的时候直接返回错误消息给客户端等等.

2020-08-15

718

资源下载

更多资源

优质分享App

近一个月的开发和优化，本站点的第一个app全新上线。该app采用极致压缩，本体才4.36MB。系统里面做了大量数据访问、缓存优化。方便用户在手机上查看文章。后续会推出HarmonyOS的适配版本。

腾讯云软件源

为解决软件依赖安装时官方源访问速度慢的问题，腾讯云为一些软件搭建了缓存服务。您可以通过使用腾讯云软件源站来提升依赖包的安装速度。为了方便用户自由搭建服务架构，目前腾讯云软件源站支持公网访问和内网访问。

Nacos

Nacos /nɑ:kəʊs/ 是 Dynamic Naming and Configuration Service 的首字母简称，一个易于构建 AI Agent 应用的动态服务发现、配置管理和AI智能体管理平台。Nacos 致力于帮助您发现、配置和管理微服务及AI智能体应用。Nacos 提供了一组简单易用的特性集，帮助您快速实现动态服务发现、服务配置、服务元数据、流量管理。Nacos 帮助您更敏捷和容易地构建、交付和管理微服务平台。

Rocky Linux

Rocky Linux（中文名：洛基）是由Gregory Kurtzer于2020年12月发起的企业级Linux发行版，作为CentOS稳定版停止维护后与RHEL（Red Hat Enterprise Linux）完全兼容的开源替代方案，由社区拥有并管理，支持x86_64、aarch64等架构。其通过重新编译RHEL源代码提供长期稳定性，采用模块化包装和SELinux安全架构，默认包含GNOME桌面环境及XFS文件系统，支持十年生命周期更新。