Stars: 167
Forks: 9
Pull Requests: 3
Issues: 21
Watchers: 8
Last Updated: 2020-12-08 14:28:05
$queue->push('I can be popped off after', '10 minutes');
License: MIT License
Languages: PHP, PLpgSQL, TSQL
Phive Queue is a time-based scheduling queue with multiple backend support.
The recommended way to install Phive Queue is through Composer:
$ composer require rybakit/phive-queue
use Phive\Queue\InMemoryQueue;
use Phive\Queue\NoItemAvailableException;
$queue = new InMemoryQueue();
$queue->push('item1');
$queue->push('item2', new DateTime());
$queue->push('item3', time());
$queue->push('item4', '+5 seconds');
$queue->push('item5', 'next Monday');
// get the queue size
$count = $queue->count(); // 5
// pop items off the queue
// note that is not guaranteed that the items with the same scheduled time
// will be received in the same order in which they were added
$item123 = $queue->pop();
$item123 = $queue->pop();
$item123 = $queue->pop();
try {
$item4 = $queue->pop();
} catch (NoItemAvailableException $e) {
// item4 is not yet available
}
sleep(5);
$item4 = $queue->pop();
// clear the queue (will remove 'item5')
$queue->clear();
Currently, there are the following queues available:
The MongoQueue
requires the Mongo PECL extension (v1.3.0 or higher).
Tip: Before making use of the queue, it's highly recommended to create an index on a eta
field:
$ mongo my_db --eval 'db.my_coll.ensureIndex({ eta: 1 })'
public MongoQueue::__construct(MongoClient $mongoClient, string $dbName, string $collName)
Parameters:
mongoClient The MongoClient instance
dbName The database name
collName The collection name
use Phive\Queue\MongoQueue;
$client = new MongoClient();
$queue = new MongoQueue($client, 'my_db', 'my_coll');
For the RedisQueue
you have to install the Redis PECL extension (v2.2.3 or higher).
public RedisQueue::__construct(Redis $redis)
Parameters:
redis The Redis instance
use Phive\Queue\RedisQueue;
$redis = new Redis();
$redis->connect('127.0.0.1');
$redis->setOption(Redis::OPT_PREFIX, 'my_prefix:');
// Since the Redis client v2.2.5 the RedisQueue has the ability to utilize serialization:
// $redis->setOption(Redis::OPT_SERIALIZER, Redis::SERIALIZER_PHP);
$queue = new RedisQueue($redis);
To use the TarantoolQueue
you have to install the Tarantool PECL
extension and a Lua script for managing queues.
public TarantoolQueue::__construct(Tarantool $tarantool, string $tubeName [, int $space = null ])
Parameters:
tarantool The Tarantool instance
tubeName The tube name
space Optional. The space number. Default to 0
use Phive\Queue\TarantoolQueue;
$tarantool = new Tarantool('127.0.0.1', 33020);
$queue = new TarantoolQueue($tarantool, 'my_tube');
The PheanstalkQueue
requires the Pheanstalk
library (Beanstalk client) to be installed:
$ composer require pda/pheanstalk:~3.0
public PheanstalkQueue::__construct(Pheanstalk\PheanstalkInterface $pheanstalk, string $tubeName)
Parameters:
pheanstalk The Pheanstalk\PheanstalkInterface instance
tubeName The tube name
use Pheanstalk\Pheanstalk;
use Phive\Queue\PheanstalkQueue;
$pheanstalk = new Pheanstalk('127.0.0.1');
$queue = new PheanstalkQueue($pheanstalk, 'my_tube');
The GenericPdoQueue
is intended for PDO drivers whose databases support stored procedures/functions
(in fact all drivers except SQLite).
The GenericPdoQueue
requires PDO and a PDO driver
for a particular database be installed. On top of that PDO error mode must be set to throw
exceptions (PDO::ERRMODE_EXCEPTION
).
SQL files to create the table and the stored routine can be found in the res directory.
public GenericPdoQueue::__construct(PDO $pdo, string $tableName [, string $routineName = null ] )
Parameters:
pdo The PDO instance
tableName The table name
routineName Optional. The routine name. Default to tableName_pop
use Phive\Queue\Pdo\GenericPdoQueue;
$pdo = new PDO('pgsql:host=127.0.0.1;port=5432;dbname=my_db', 'db_user', 'db_pass');
$pdo->setAttribute(PDO::ATTR_ERRMODE, PDO::ERRMODE_EXCEPTION);
$queue = new GenericPdoQueue($pdo, 'my_table', 'my_routine');
The SqlitePdoQueue
requires PDO and SQLite PDO driver.
On top of that PDO error mode must be set to throw exceptions (PDO::ERRMODE_EXCEPTION
).
SQL file to create the table can be found in the res/sqlite directory.
Tip: For performance reasons it's highly recommended to activate WAL mode:
$pdo->exec('PRAGMA journal_mode=WAL');
public SqlitePdoQueue::__construct(PDO $pdo, string $tableName)
Parameters:
pdo The PDO instance
tableName The table name
use Phive\Queue\Pdo\SqlitePdoQueue;
$pdo = new PDO('sqlite:/opt/databases/my_db.sq3');
$pdo->setAttribute(PDO::ATTR_ERRMODE, PDO::ERRMODE_EXCEPTION);
$pdo->exec('PRAGMA journal_mode=WAL');
$queue = new SqlitePdoQueue($pdo, 'my_table');
The SysVQueue
requires PHP to be compiled with the option --enable-sysvmsg.
public SysVQueue::__construct(int $key [, bool $serialize = null [, int $perms = null ]] )
Parameters:
key The message queue numeric ID
serialize Optional. Whether to serialize an item or not. Default to false
perms Optional. The queue permissions. Default to 0666
use Phive\Queue\SysVQueue;
$queue = new SysVQueue(123456);
The InMemoryQueue
can be useful in cases where the persistence is not needed. It exists only in RAM
and therefore operates faster than other queues.
public InMemoryQueue::__construct()
use Phive\Queue\InMemoryQueue;
$queue = new InMemoryQueue();
The following table details the various item types supported across queues.
Queue/Type | string | binary string | null | bool | int | float | array | object |
---|---|---|---|---|---|---|---|---|
MongoQueue | ✓ | ✓ | ✓ | ✓ | ✓ | ✓ | ||
RedisQueue | ✓ | ✓ | ✓ | ✓ | ✓ | ✓ | ✓* | ✓* |
TarantoolQueue | ✓ | ✓ | ✓ | ✓ | ✓ | ✓ | ||
PheanstalkQueue | ✓ | ✓ | ✓ | ✓ | ✓ | ✓ | ||
GenericPdoQueue | ✓ | ✓ | ✓ | ✓ | ✓ | |||
SqlitePdoQueue | ✓ | ✓ | ✓ | ✓ | ✓ | |||
SysVQueue | ✓ | ✓ | ✓* | ✓ | ✓ | ✓ | ✓* | ✓* |
InMemoryQueue | ✓ | ✓ | ✓ | ✓ | ✓ | ✓ | ✓ | ✓ |
✓* — supported if the serializer is enabled.
To bypass the limitation of unsupported types for the particular queue you could convert an item
to a non-binary string before pushing it and then back after popping. The library ships with
the TypeSafeQueue
decorator which does that for you:
use Phive\Queue\GenericPdoQueue;
use Phive\Queue\TypeSafeQueue;
$queue = new GenericPdoQueue(...);
$queue = new TypeSafeQueue($queue);
$queue->push(['foo' => 'bar']);
$array = $queue->pop(); // ['foo' => 'bar'];
Every queue method declared in the Queue interface will throw an exception if a run-time error occurs at the time the method is called.
For example, in the code below, the push()
call will fail with a MongoConnectionException
exception in a case a remote server unreachable:
use Phive\Queue\MongoQueue;
$queue = new MongoQueue(...);
// mongodb server goes down here
$queue->push('item'); // throws MongoConnectionException
But sometimes you may want to catch exceptions coming from a queue regardless of the underlying driver.
To do this just wrap your queue object with the ExceptionalQueue
decorator:
use Phive\Queue\ExceptionalQueue;
use Phive\Queue\MongoQueue;
$queue = new MongoQueue(...);
$queue = new ExceptionalQueue($queue);
// mongodb server goes down here
$queue->push('item'); // throws Phive\Queue\QueueException
And then, to catch queue level exceptions use the QueueException
class:
use Phive\Queue\QueueException;
...
try {
do_something_with_a_queue();
} catch (QueueException $e) {
// handle queue exception
} catch (\Exception $e) {
// handle base exception
}
Phive Queue uses PHPUnit for unit and integration testing. In order to run the tests, you'll first need to install the library dependencies using composer:
$ composer install
You can then run the tests:
$ phpunit
You may also wish to specify your own default values of some tests (db names, passwords, queue sizes, etc.). You can do it by setting environment variables from the command line:
$ export PHIVE_PDO_PGSQL_PASSWORD="pgsql_password"
$ export PHIVE_PDO_MYSQL_PASSWORD="mysql_password"
$ phpunit
You may also create your own phpunit.xml
file by copying the phpunit.xml.dist
file and customize to your needs.
To check the performance of queues run:
$ phpunit --group performance
This test inserts a number of items (1000 by default) into a queue, and then retrieves them back.
It measures the average time for push
and pop
operations and outputs the resulting stats, e.g.:
RedisQueue::push()
Total operations: 1000
Operations per second: 14031.762 [#/sec]
Time per operation: 71.267 [ms]
Time taken for test: 0.071 [sec]
RedisQueue::pop()
Total operations: 1000
Operations per second: 16869.390 [#/sec]
Time per operation: 59.279 [ms]
Time taken for test: 0.059 [sec]
.
RedisQueue::push() (delayed)
Total operations: 1000
Operations per second: 15106.226 [#/sec]
Time per operation: 66.198 [ms]
Time taken for test: 0.066 [sec]
RedisQueue::pop() (delayed)
Total operations: 1000
Operations per second: 14096.416 [#/sec]
Time per operation: 70.940 [ms]
Time taken for test: 0.071 [sec]
You may also change the number of items involved in the test by changing the PHIVE_PERF_QUEUE_SIZE
value in your phpunit.xml
file or by setting the environment variable from the command line:
$ PHIVE_PERF_QUEUE_SIZE=5000 phpunit --group performance
In order to check the concurrency you'll have to install the Gearman server and the German PECL extension. Once the server has been installed and started, create a number of processes (workers) by running:
$ php tests/worker.php
Then run the tests:
$ phpunit --group concurrency
This test inserts a number of items (100 by default) into a queue, and then each worker tries to retrieve them in parallel.
You may also change the number of items involved in the test by changing the PHIVE_CONCUR_QUEUE_SIZE
value in your phpunit.xml
file or by setting the environment variable from the command line:
$ PHIVE_CONCUR_QUEUE_SIZE=500 phpunit --group concurrency
Phive Queue is released under the MIT License. See the bundled LICENSE file for details.