PHP + Redis 实现简单消息队列
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 的消息进行处理。
低调大师中文资讯倾力打造互联网数据资讯、行业资源、电子商务、移动互联网、网络营销平台。
持续更新报道IT业界、互联网、市场资讯、驱动更新,是最及时权威的产业资讯及硬件资讯报道平台。
转载内容版权归作者及来源网站所有,本站原创内容转载请注明来源。
- 上一篇
缓存雪崩 和 缓存穿透 问题及解决方案----->(浅谈)
缓存雪崩: 缓存同一时间大面积的失败,所以后面的请求都会落到数据库上,造成数据库短时间内承受大量请求而崩掉 解决方案: 事前:尽量保证整个redis集群的高可用性,发现机器宕机尽快补上 事中:本地ehcache缓存+hystrix限流&降级,避免MySQL崩掉 事后:利用redis持久化机制保存的数据尽快恢复缓存 缓存穿透: 简答点说就是大量请求的Key根本不存在于缓存中,导致请求直接到了数据库上,根本没有经过缓存这一次层. 解决方案: 最基本的就是首先做好参数校验,一些不合法的参数请求直接抛出异常信息返回给客户端.比如查询的数据库id不能小于0,传入的邮箱格式不对的时候直接返回错误消息给客户端等等.
- 下一篇
最新版Dubbo-admin 服务器部署
1、下载dubbo-admin服务器版本 2、上传至服务器,并解压tar -zxvf xxxx.tar.gz 3、进入解压后的目录 4、默认启动8080端口,可能会存在冲突, 修改dubbo-admin-server的端口 修改dubbo-admin-ui的端口 4、进行打包mvn clean,mvn package 执行至这个位置可能会报npm run buildfaild;进入dubbo-admin-ui目录执行npm run build 发现会报 building for production...Killed;是服务器内存不够用了,这样就给他配置一个单独的内存出来就解决了; 执行命令: sudo /bin/dd if=/dev/zero of=/var/swap.1 bs=1M count=1024 sudo /sbin/mkswap /var/swap.1 sudo /sbin/swapon /var/swap.1 然后再次 mvn clean, mvn package; 会在这个位置等待安装十多二十分钟,而且再新开一个终端执行命令会花很长时间; 下图表示正在构建node, ...
相关文章
文章评论
共有0条评论来说两句吧...