ci框架如何用redis队列

ci框架用redis队列的示例：

在autoload.php中加入如下配置行：

$autoload['libraries'] = array('redis');

在/application/config中加入文件redis.php，文件内容如下：

// Default connection group
$config['redis_default']['host'] = 'localhost'; // IP address or host
$config['redis_default']['port'] = '6379'; // Default Redis port is 6379
$config['redis_default']['password'] = ''; // Can be left empty when the server does not require AUTH
$config['redis_slave']['host'] = '';
$config['redis_slave']['port'] = '6379';
$config['redis_slave']['password'] = '';
?>

在/application/libraries中加入文件Redis.php，文件内容如下：

class CI_Redis {
private $_ci;
private $_connection;
public $debug = FALSE;
const CRLF = "\r\n";
public function __construct($params = array())
{
log_message('debug', 'Redis Class Initialized');
$this->_ci = get_instance();
$this->_ci->load->config('redis');
// Check for the different styles of configs
if (isset($params['connection_group']))
{
// Specific connection group
$config = $this->_ci->config->item('redis_' . $params['connection_group']);
}
elseif (is_array($this->_ci->config->item('redis_default')))
{
// Default connection group
$config = $this->_ci->config->item('redis_default');
}
else
{
// Original config style
$config = array(
'host' => $this->_ci->config->item('redis_host'),
'port' => $this->_ci->config->item('redis_port'),
'password' => $this->_ci->config->item('redis_password'),
);
}
// Connect to Redis
$this->_connection = @fsockopen($config['host'], $config['port'], $errno, $errstr, 3);
// Display an error message if connection failed
if ( ! $this->_connection)
{
show_error('Could not connect to Redis at ' . $config['host'] . ':' . $config['port']);
}
// Authenticate when needed
$this->_auth($config['password']);
}
public function __call($method, $arguments)
{
$request = $this->_encode_request($method, $arguments);
return $this->_write_request($request);
}
public function command($string)
{
$slices = explode(' ', $string);
$request = $this->_encode_request($slices[0], array_slice($slices, 1));
return $this->_write_request($request);
}
private function _auth($password = NULL)
{
// Authenticate when password is set
if ( ! empty($password))
{
// See if we authenticated successfully
if ($this->command('AUTH ' . $password) !== 'OK')
{
show_error('Could not connect to Redis, invalid password');
}
}
}
public function _clear_socket()
{
// Read one character at a time
fflush($this->_connection);
return NULL;
}
private function _write_request($request)
{
if ($this->debug === TRUE)
{
log_message('debug', 'Redis unified request: ' . $request);
}
// How long is the data we are sending?
$value_length = strlen($request);
// If there isn't any data, just return
if ($value_length <= 0) return NULL;
// Handle reply if data is less than or equal to 8192 bytes, just send it over
if ($value_length <= 8192)
{
fwrite($this->_connection, $request);
}
else
{
while ($value_length > 0)
{
// If we have more than 8192, only take what we can handle
if ($value_length > 8192) {
$send_size = 8192;
}
// Send our chunk
fwrite($this->_connection, $request, $send_size);
// How much is left to send?
$value_length = $value_length - $send_size;
// Remove data sent from outgoing data
$request = substr($request, $send_size, $value_length);
}
}
// Read our request into a variable
$return = $this->_read_request();
// Clear the socket so no data remains in the buffer
$this->_clear_socket();
return $return;
}
private function _read_request()
{
$type = fgetc($this->_connection);
// Times we will attempt to trash bad data in search of a
// valid type indicator
$response_types = array('+', '-', ':', '$', '*');
$type_error_limit = 50;
$try = 0;
while ( ! in_array($type, $response_types) && $try < $type_error_limit)
{
$type = fgetc($this->_connection);
$try++;
}
if ($this->debug === TRUE)
{
log_message('debug', 'Redis response type: ' . $type);
}
switch ($type)
{
case '+':
return $this->_single_line_reply();
break;
case '-':
return $this->_error_reply();
break;
case ':':
return $this->_integer_reply();
break;
case '$':
return $this->_bulk_reply();
break;
case '*':
return $this->_multi_bulk_reply();
break;
default:
return FALSE;
}
}
private function _single_line_reply()
{
$value = rtrim(fgets($this->_connection));
$this->_clear_socket();
return $value;
}
private function _error_reply()
{
// Extract the error message
$error = substr(rtrim(fgets($this->_connection)), 4);
log_message('error', 'Redis server returned an error: ' . $error);
$this->_clear_socket();
return FALSE;
}
private function _integer_reply()
{
return (int) rtrim(fgets($this->_connection));
}
private function _bulk_reply()
{
// How long is the data we are reading? Support waiting for data to
// fully return from redis and enter into socket.
$value_length = (int) fgets($this->_connection);
if ($value_length <= 0) return NULL;
$response = '';
// Handle reply if data is less than or equal to 8192 bytes, just read it
if ($value_length <= 8192)
{
$response = fread($this->_connection, $value_length);
}
else
{
$data_left = $value_length;
// If the data left is greater than 0, keep reading
while ($data_left > 0 ) {
// If we have more than 8192, only take what we can handle
if ($data_left > 8192)
{
$read_size = 8192;
}
else
{
$read_size = $data_left;
}
// Read our chunk
$chunk = fread($this->_connection, $read_size);
// Support reading very long responses that don't come through
// in one fread
$chunk_length = strlen($chunk);
while ($chunk_length < $read_size)
{
$keep_reading = $read_size - $chunk_length;
$chunk .= fread($this->_connection, $keep_reading);
$chunk_length = strlen($chunk);
}
$response .= $chunk;
// Re-calculate how much data is left to read
$data_left = $data_left - $read_size;
}
}
// Clear the socket in case anything remains in there
$this->_clear_socket();
return isset($response) ? $response : FALSE;
}
private function _multi_bulk_reply()
{
// Get the amount of values in the response
$response = array();
$total_values = (int) fgets($this->_connection);
// Loop all values and add them to the response array
for ($i = 0; $i < $total_values; $i++)
{
// Remove the new line and carriage return before reading
// another bulk reply
fgets($this->_connection, 2);
// If this is a second or later pass, we also need to get rid
// of the $ indicating a new bulk reply and its length.
if ($i > 0)
{
fgets($this->_connection);
fgets($this->_connection, 2);
}
$response[] = $this->_bulk_reply();
}
// Clear the socket
$this->_clear_socket();
return isset($response) ? $response : FALSE;
}
private function _encode_request($method, $arguments = array())
{
$request = '$' . strlen($method) . self::CRLF . $method . self::CRLF;
$_args = 1;
// Append all the arguments in the request string
foreach ($arguments as $argument)
{
if (is_array($argument))
{
foreach ($argument as $key => $value)
{
// Prepend the key if we're dealing with a hash
if (!is_int($key))
{
$request .= '$' . strlen($key) . self::CRLF . $key . self::CRLF;
$_args++;
}
$request .= '$' . strlen($value) . self::CRLF . $value . self::CRLF;
$_args++;
}
}
else
{
$request .= '$' . strlen($argument) . self::CRLF . $argument . self::CRLF;
$_args++;
}
}
$request = '*' . $_args . self::CRLF . $request;
return $request;
}
public function info($section = FALSE)
{
if ($section !== FALSE)
{
$response = $this->command('INFO '. $section);
}
else
{
$response = $this->command('INFO');
}
$data = array();
$lines = explode(self::CRLF, $response);
// Extract the key and value
foreach ($lines as $line)
{
$parts = explode(':', $line);
if (isset($parts[1])) $data[$parts[0]] = $parts[1];
}
return $data;
}
public function debug($bool)
{
$this->debug = (bool) $bool;
}
function __destruct()
{
if ($this->_connection) fclose($this->_connection);
}
}
?>

在文件中使用，代码：

if($this->redis->get('mark_'.$gid) === null){ //如果未设置
$this->redis->set('mark_'.$gid, $giftnum); //设置
$this->redis->EXPIRE('mark_'.$gid, 30*60); //设置过期时间 （30 min）
}else{
$giftnum = $this->redis->get('mark_'.$gid); //从缓存中直接读取对应的值
}
?>