小编给大家分享一下PHP中如何实现基于redis的MessageQueue队列封装，希望大家阅读完这篇文章之后都有所收获，下面让我们一起去探讨吧！
创新互联公司是一家专业提供阿瓦提企业网站建设,专注与网站建设、网站设计、H5高端网站建设、小程序制作等业务。10年已为阿瓦提众多企业、政府机构等服务。创新互联专业网站制作公司优惠进行中。
Redis的链表List可以用来做链表，高并发的特性非常适合做分布式的并行消息传递。
左进右出
$redis->lPush($key, $value); $redis->rPop($key);
以下程序已在生产环境中正式使用。
基于Redis的PHP消息队列封装
'127.0.0.1', 'PORT' => '6379'), $key = 'redis_message_queue', $p_connect = false)
{
if (empty($key))
throw new \Exception('message queue key can not be empty');
$this->server = $server_config['IP'];
$this->port = $server_config['PORT'];
$this->key = $key;
$this->check_environment();
if ($p_connect) {
$this->pconnect();
} else {
$this->connect();
}
}
/**
* 析构函数，关闭redis链接，使用长连接时，最好主动调用关闭
*/
public function __destruct()
{
$this->close();
}
/**
* 短连接
*/
private function connect()
{
$this->redis_server = new \Redis();
$this->redis_server->connect($this->server, $this->port);
}
/**
* 长连接
*/
public function pconnect()
{
$this->redis_server = new \Redis();
$this->redis_server->pconnect($this->server, $this->port);
}
/**
* 关闭链接
*/
public function close()
{
$this->redis_server->close();
}
/**
* 向队列插入一条信息
* @param $message
* @return mixed
*/
public function put($message)
{
return $this->redis_server->lPush($this->key, $message);
}
/**
* 向队列中插入一串信息
* @param $message
* @return mixed
*/
public function puts(){
$params = func_get_args();
$message_array = array_merge(array($this->key), $params);
return call_user_func_array(array($this->redis_server, 'lPush'), $message_array);
}
/**
* 从队列顶部获取一条记录
* @return mixed
*/
public function get()
{
return $this->redis_server->lPop($this->key);
}
/**
* 选择数据库，可以用于区分不同队列
* @param $database
*/
public function select($database)
{
$this->redis_server->select($database);
}
/**
* 获得队列状态，即目前队列中的消息数量
* @return mixed
*/
public function size()
{
return $this->redis_server->lSize($this->key);
}
/**
* 获取某一位置的值，不会删除该位置的值
* @param $pos
* @return mixed
*/
public function view($pos)
{
return $this->redis_server->lGet($this->key, $pos);
}
/**
* 检查Redis扩展
* @throws Exception
*/
protected function check_environment()
{
if (!\extension_loaded('redis')) {
throw new \Exception('Redis extension not loaded');
}
}
}如果需要一次写入多个队列，可以使用如下调用方式：
puts(1, 2, 3, 4); $redis->puts(5, 6, 7, 8, 9);
模仿HTTPSQS输出结果的封装如下，提供了写入位置和读取位置记录的功能：
'127.0.0.1', 'PORT' => '6379'),
$key = 'redis_message_queue',
$p_connect = false,
$record_status=true
){
parent::__construct($server_config, $key, $p_connect);
$this->record_status = $record_status;
$this->put_position = $this->key . '_put_position';
$this->get_position = $this->key . '_get_position';
}
public function get(){
if($queue = parent::get()){
$incr_result = $this->redis_server->incr($this->get_position);
if(!$incr_result) throw new \Exception('can not mark get position,please check the redis server');
return $queue;
}else{
return false;
}
}
public function put($message){
if(parent::put($message)){
$incr_result = $this->redis_server->incr($this->put_position);
if(!$incr_result) throw new \Exception('can not mark put position,please check the redis server');
return true;
}else{
return false;
}
}
public function puts_status(){
$message_array = func_get_args();
$result = call_user_func_array(array($this, 'puts'), $message_array);
if($result){
$this->redis_server->incrBy($this->put_position, count($message_array));
return true;
}
return false;
}
public function size(){
return $this->redis_server->lSize($this->key);
}
public function status(){
$status['put_position'] = ($put_position = $this->redis_server->get($this->put_position)) ? $put_position : 0;
$status['get_position'] = ($get_position = $this->redis_server->get($this->get_position)) ? $get_position : 0;
$status['unread_queue'] = $this->size();
$status['queue_name'] = $this->key;
$status['server'] = $this->server;
$status['port'] = $this->port;
return $status;
}
public function status_normal(){
$status = $this->status();
$message = 'Redis Message Queue' . PHP_EOL;
$message .= '-------------------' . PHP_EOL;
$message .= 'Message queue name:' . $status['queue_name'] . PHP_EOL;
$message .= 'Put position of queue:' . $status['put_position'] . PHP_EOL;
$message .= 'Get position of queue:' . $status['get_position'] . PHP_EOL;
$message .= 'Number of unread queue:' . $status['unread_queue'] . PHP_EOL;
return $message;
}
public function status_json(){
return \json_encode($this->status());
}
}看完了这篇文章，相信你对“PHP中如何实现基于Redis的MessageQueue队列封装”有了一定的了解，如果想了解更多相关知识，欢迎关注创新互联行业资讯频道，感谢各位的阅读！
