PHP实现基于Redis的MessageQueue队列封装操作示例

$redis->lPush($key, $value);$redis->rPop($key);

<?php/** * Created by PhpStorm. * User: huyanping * Date: 14-8-19 * Time: 下午12:10 * * 基于Redis的消息队列封装 */namespace Zebra/MessageQueue;class RedisMessageQueue implements IMessageQueue{  protected $redis_server;  protected $server;  protected $port;  /**   * @var 消息队列标志   */  protected $key;  /**   * 构造队列，创建redis链接   * @param $server_config   * @param $key   * @param bool $p_connect   */  public function __construct($server_config = array('IP' => '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');    }  }}

<?php$redis = new RedisMessageQueue();$redis->puts(1, 2, 3, 4);$redis->puts(5, 6, 7, 8, 9);

<?php/** * Created by PhpStorm. * User: huyanping * Date: 14-9-5 * Time: 下午2:16 * * 附加了队列状态信息的RedisMessageQueue */namespace Zebra/MessageQueue;class RedisMessageQueueStatus extends RedisMessageQueue {  protected $record_status;  protected $put_position;  protected $get_position;  public function __construct(    $server_config = array('IP' => '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());  }}