PHP parallel task multi-thread message based library
Library for running parallel php tasks in a simple and intuitive way across multiple computers.
What can be done with it:
- Delegate tasks in order to improve application response time. For example, instead of sending an email inline, it can be sent asynchronously.
- Delegate tasks whose result are not required to be immediately visible or for which a response is not needed in the current script. The Examples here can range from simple things like logging up to complex application logic that can be processed with delay in order to improve high peak performance.
- Parallel processing. An algorithm that can be parallelized can benefit from this library by starting multiple workers on different machines and aggregate the results after all finished.
How to use it:
- Choose a queue implementation and create an instance of it. Existing implementations: Redis and RabbitMQ. Help with more implementations is appreciated.
- Build the worker using queue and start it in a cli environment. You can start multiple workers.
$workerBuilder = new ExecutorWorkerBuilder() $worker = $workerBuilder->withConsumeQueue($queue)->buildWorker(); $worker->work($type);
- Build the executor using queue
$executorBuilder = new ExecutorWorkerBuilder() $executor = $executorBuilder->withPublishQueue($queue)->buildExecutor();
- Define a task by implementing the
- Submit task and get results
$futureResult = $executor->submit($type, MyTask::class, [$param1, $param2]); //...some other time consuming tasks $result = $futureResult->getResult();
Technically, the library is splitted into three parts:
Queue module. Handles queue implementation details. From outside, it contains two public interfaces
ConsumeQueueand the input/output entities from it's methods:
InputMessageand store it for processing
InputMessage, store it for processing and returns and
InputMessageIdentifierfor fetching the result.
InputMessageIdentifierand returns the
OutputMessageresult for it. It's blocking until the result is available.
runreceives a callable. It should fetch an
InputMessage, run to callable with it as an input and stores the received
Task module. Uses Queue module adding a layer of
Taskimplementation. From the outside, it has two classes with two interfaces:
TaskRunnereach of them used for scheduling a task for asynchronous run and respectively running tasks asynchronously.
The interfaces of
executereceives a new Task and the inputs for it. It should be used for methods that don't have a return type.
submitreceives a new Task and the inputs for it and returns a
FutureTaskResult. Should be used for methods that returns something.
The interfaces of
runruns tasks asynchronously in a loop
runOnceruns only one task asynchronously
Executor/Worker facade. It is just a wrapper around Task module to allow easier composition and usage.
There is a builder that helps creating
Workerallowing to configure implementation for
ConsumeQueueand to change default implementation of