Download the PHP package pdffiller/qless-php without Composer

qless-php

PHP Bindings for qless.

Qless is a powerful Redis-based job queueing system inspired by resque, but built on a collection of Lua scripts, maintained in the qless-core repo. Be sure to check the change log.

A big thank you to our contributors; you rock!

NOTE: This library is fully reworked and separately developed version of Contatta's qless-php. The copyright to the Contatta/qless-php code belongs to Ryver, Inc. For more see the Contatta/qless-php license.

Documentation is borrowed from seomoz/qless.

Contents

• Philosophy and Nomenclature
• Features
• Installation
  • Requirements
• Usage
  • Enqueuing Jobs
  • Running A Worker
  • Forking Worker
  • Non-Forking Worker
  • Signal Handling
  • Job Reservers
  • Custom Job Handler
  • Web Interface
  • Job Dependencies
  • Priority
  • Scheduled Jobs
  • Recurring Jobs
  • Topics
  • Configuration Options
  • Tagging / Tracking
  • Event System
  • Per-Job Events
  • List of Events
  • Sync job processing
  • Heartbeating
  • Stats
  • Time
  • Ensuring Job Uniqueness
  • Setting Default Job Options
  • Testing Jobs
• Contributing and Developing
• License

Philosophy and Nomenclature

A job is a unit of work identified by a job id or jid. A queue can contain several jobs that are scheduled to be run at a certain time, several jobs that are waiting to run, and jobs that are currently running. A worker is a process on a host, identified uniquely, that asks for jobs from the queue, performs some process associated with that job, and then marks it as complete. When it's completed, it can be put into another queue.

Jobs can only be in one queue at a time. That queue is whatever queue they were last put in. So if a worker is working on a job, and you move it, the worker's request to complete the job will be ignored.

A job can be canceled, which means it disappears into the ether, and we'll never pay it any mind ever again. A job can be dropped, which is when a worker fails to heartbeat or complete the job in a timely fashion, or a job can be failed, which is when a host recognizes some systematically problematic state about the job. A worker should only fail a job if the error is likely not a transient one; otherwise, that worker should just drop it and let the system reclaim it.

Features

• Jobs don't get dropped on the floor — Sometimes workers drop jobs. Qless automatically picks them back up and gives them to another worker.
• Tagging / Tracking — Some jobs are more interesting than others. Track those jobs to get updates on their progress. Tag jobs with meaningful identifiers to find them quickly in the UI.
• Topics — Deliver a job to multiple queues.
• Job Dependencies — One job might need to wait for another job to complete,
• Stats — qless automatically keeps statistics about how long jobs wait to be processed and how long they take to be processed. Currently, we keep track of the count, mean, standard deviation, and a histogram of these times.
• Job data is stored temporarily — Job info sticks around for a configurable amount of time, so you can still look back on a job's history, data, etc.
• Priority — Jobs with the same priority get popped in the order they were inserted; a higher priority means that it gets popped faster.
• Retry logic — Every job has a number of retries associated with it, which are renewed when it is put into a new queue or completed. If a job is repeatedly dropped, then it is presumed to be problematic, and is automatically failed.
• Web App — With the advent of a Ruby client, there is a Sinatra-based web app that gives you control over certain operational issues.
• Scheduled Work — Until a job waits for a specified delay (defaults to 0), jobs cannot be popped by workers.
• Recurring Jobs — Scheduling is all well and good, but we also support jobs that need to recur periodically.
• Notifications — Tracked jobs emit events on pubsub channels as they get completed, failed, put, popped, etc. Use these events to get notified of progress on jobs you're interested in.

Installation

Requirements

Prerequisite PHP extensions are:

• json
• pcntl
• posix
• pcre
• sockets

Supported PHP versions are: 7.1, 7.2, 7.3, 8.0 and 8.1.

Qless PHP can be installed via Composer:

Alternatively, install qless-php from source by checking it out from GitHub:

NOTE: The master branch will always contain the latest unstable version. If you wish to check older versions or formal, tagged release, please switch to the relevant release.

The adm directory contains the configuration examples useful for system administrators.

Usage

Enqueuing Jobs

First things first, create a Qless Client. The Client accepts all the same arguments that you'd use when constructing a Predis\Client client.

Jobs should be classes that define a perform method, which must accept a single Qless\Jobs\BaseJob argument:

Now you can access a queue, and add a job to that queue.

The job data must be serializable to JSON, and it is recommended that you use a hash for it. See below for a list of the supported job options.

The argument returned by queue->put() is the jid (Job ID). Every Qless job has a unique jid, and it provides a means to interact with an existing job:

Running A Worker

The Qless PHP non-forking worker (e.g. if you'd rather save memory by not forking the worker for each job).

Forking Worker

As with resque...

• The worker forks a child process for each job in order to provide resilience against memory leaks (Pass the RUN_AS_SINGLE_PROCESS environment variable to force Qless to not fork the child process. Single process mode should only be used in some test/dev environments.)
• The worker updates its procline with its status, so you can see what workers are doing using ps
• The worker registers signal handlers so that you can control it by sending it signals
• The worker is given a list of queues to pop jobs off of
• The worker logs out put based on setting of the Psr\Log\LoggerInterface instance passed to worker

Resque uses queues for its notion of priority. In contrast, qless has priority support built-in. Thus, the worker supports two strategies for what order to pop jobs off the queues: ordered and round-robin. The ordered reserver will keep popping jobs off the first queue until it is empty, before trying to pop job off the second queue. The round-robin reserver will pop a job off the first queue, then the second queue, and so on. You could also easily implement your own.

To start a worker, write a bit of PHP code that instantiates a worker and runs it. You could write a simple script to do this, for example:

Non-Forking Worker

Qless PHP also includes a non-forking worker. This can alleviate a number of issues related to external connections (such as to Redis or MySQL) from worker processes, and may give better results when using an external service manager such as systemd to manage the worker process. Usage is very similar to the forking worker:

Signal Handling

The following POSIX-compliant signals are supported in the parent process:

• TERM: Shutdown immediately, stop processing jobs
• INT: Shutdown immediately, stop processing jobs
• QUIT: Shutdown after the current job has finished processing
• USR1: Forking Worker: Kill the forked child immediately, continue processing jobs
• USR1: Non-Forking Worker: Abandon progress on the current job immediately, continue processing jobs
• USR2: Don't process any new jobs, and dump the current backtrace
• CONT: Start processing jobs again after a USR2

For detailed info regarding the signals refer to signal(7).

When using the Forking Worker, you should send these to the master process, not the child.

The child process supports the USR2 signal, which causes it to dump its current backtrace.

When using the Non-Forking worker, proper handling of USR1 signals requires that Exceptions of class \Qless\Exceptions\SimpleWorkerContinuationException are not caught. If your Job class, or JobPerformer catches all Exceptions or all Throwables, you will need to re-throw instances of \Qless\Exceptions\SimpleWorkerContinuationException, or USR1 signals will be ignored.

If you wish to change how signals are handled, you can sub-class the Worker class you wish to use, and override the handleSignal method.

Job Reservers

There are different job reservers.

• DefaultReserver: A default job reserver
• OrderedReserver: Orders queues by its name
• PriorityReserver: Orders queues by its priority
• RoundRobinReserver: Round-robins through all the provided queues
• ShuffledRoundRobin: Like RoundRobinReserver but shuffles the order of the queues

Custom Job Handler

There is an ability to set custom Job Handler to process jobs. To do this call \Qless\Workers\WorkerInterface::registerJobPerformHandler method. Its argument should implement the \Qless\Jobs\PerformAwareInterface interface. This approach is handy when Job Handler is complicated service and/or has dependencies. Let's look at an example in which we need to get a custom Job Handler created by external factory:

Web Interface

The Qless PHP does not ship with a web app. However, there is a resque-inspired web app provided by seomoz/qless. In addition, you can take advantage of docker based dashboard. We're planning to create a robust and elegant web interface using PHP framework, but that task does not have the highest priority.

Job Dependencies

Let's say you have one job that depends on another, but the task definitions are fundamentally different. You need to bake a turkey, and you need to make stuffing, but you can't make the turkey until the stuffing is made:

When the stuffing job completes, the turkey job is unlocked and free to be processed.

Priority

Some jobs need to get popped sooner than others. Whether it's a trouble ticket, or debugging, you can do this pretty easily when you put a job in a queue:

What happens when you want to adjust a job's priority while it's still waiting in a queue?

Scheduled Jobs

If you don't want a job to be run right away but some time in the future, you can specify a delay:

This doesn't guarantee that job will be run exactly at 10 minutes. You can accomplish this by changing the job's priority so that once 10 minutes has elapsed, it's put before lesser-priority jobs:

Recurring Jobs

Sometimes it's not enough simply to schedule one job, but you want to run jobs regularly. In particular, maybe you have some batch operation that needs to get run once an hour, and you don't care what worker runs it. Recurring jobs are specified much like other jobs:

You can even access them in much the same way as you would normal jobs:

Changing the interval at which it runs after the fact is trivial:

If you want it to run every hour on the hour, but it's 2:37 right now, you can specify an offset which is how long it should wait before popping the first job:

Recurring jobs also have priority, a configurable number of retries, and tags. These settings don't apply to the recurring jobs, but rather the jobs that they create. In the case where more than one interval passes before a worker tries to pop the job, more than one job is created. The thinking is that while it's completely client-managed, the state should not be dependent on how often workers are trying to pop jobs.

Topics

Topic help you to put job to different queues. First, you must create subscription. You can use pattern for a name of topics. Symbol * - one word, # - few words divided by point .. Examples: first.second.*, *.second.*, #.third.

Then you can put job to all subscribers.

You can call all Queue's public methods for Topic.

Configuration Options

You can get and set global (read: in the context of the same Redis instance) configuration to change the behavior for heartbeating, and so forth. There aren't a tremendous number of configuration options, but an important one is how long job data is kept around. Job data is expired after it has been completed for jobs-history seconds, but is limited to the last jobs-history-count completed jobs. These default to 50k jobs, and 30 days, but depending on volume, your needs may change. There is also possible to configure a failed job in history, using jobs-failed-history parameter. To only keep the last 500 jobs for up to 7 days and failed jobs for 3 days:

Tagging / Tracking

In qless, 'tracking' means flagging a job as important. Tracked jobs have a tab reserved for them in the web interface, and they also emit subscribable events as they make progress (more on that below). You can flag a job from the web interface, or the corresponding code:

Jobs can be tagged with strings which are indexed for quick searches. For example, jobs might be associated with customer accounts, or some other key that makes sense for your project.

This makes them searchable in the web interface, or from code:

You can add or remove tags at will, too:

Notifications

Tracked jobs emit events on specific pubsub channels as things happen to them. Whether it's getting popped off of a queue, completed by a worker, etc.

An example of this is:

Those familiar with redis pubsub will note that a redis connection can only be used for pubsub-y commands once listening. For this reason, invoking Client->events actually creates a second connection so that Client can still be used as it normally would be:

The possible event types match those defined by Qless-core and are defined as constants on the \Qless\PubSub\Manager class: EVENT_CANCELED, EVENT_COMPLETED, EVENT_FAILED, EVENT_POPPED, EVENT_STALLED, EVENT_PUT, EVENT_TRACK, EVENT_UNTRACK.

Event System

Qless also has a basic event system that can be used by your application to customize how some qless internals behave. Events can be used to inject logic before, after or around the processing of a single job in the child process. This can be useful, for example, when you need to re-establish a connection to your database for each job.

Event has few main concepts - entity, happening, source.

• entity is an object type (component) with whom event is taking place, for example entity can be job, worker, queue.
• happening is an act that is taking place, for example it can be beforeFork or beforePerform
• source is an object who fired an event

In code Event is represented by some class. All events classes are descendants of \Qless\Events\User\AbstractEvent class. You can get entity and happening of event by calling static methods getEntityName() and getHappening(), you can get full name of event (made of entity and happening) by calling static method getName() source you can get with getSource().

Also, there are subscribers for events. Subscriber can be any class with methods named as event's happening (example: beforeFork(AbstractEvent $event)). Also, subscriber can be a closure. Handling method of subscriber will receive only one parameter - event, you can get all data you need from that event.

You can attach a subscriber to a specific event or events group (grouped bye events entity)

Example: Define a subscriber with an beforeFork method that will be called where you want the job to be processed:

Then, attach a subscriber to the worker events group:

To attach a subscriber to a specific event you can do:

You can attach subscribers as many as you want. Qless events system supports priories, so you can change default priority:

Per-Job Events

As it was mentioned above, Qless supports events on a per-entity basis. So per-job events are available if you have some orthogonal logic to run in the context of some (but not all) jobs. Every Job Event class is a descendant of \Qless\Events\User\Job\AbstractJobEvent and contains entity of \Qless\Jobs\BaseJob. To get the job from event you can use $event->getJob() method.

Per-job subscribes can be defined the same way as worker's subscribers:

To add them to a job class, you first have to make your job class events-aware by subscribing on the required events group. To achieve this just implement setUp method and subscribe to the desired events:

Note: In this scenario your job class must implement Qless\EventsManagerAwareInterface.

Yet another example. Let's assume that job's payload should always contain additional data from the current context. You can easily amend it using the BeforeEnqueue subscriber:

List of Events

Full list of events available in Qless:

Sync job processing

If you want your job to be processed without worker, you can set sync mode for qless client. In configuration of your project write code like this:

Now you all job will be process without a worker, synchronously.

Note: Use this feature for testing your job in development environment.

Heartbeating

When a worker is given a job, it is given an exclusive lock to that job. That means that job won't be given to any other worker, so long as the worker checks in with progress on the job. By default, jobs have to either report back progress every 60 seconds, or complete it, but that's a configurable option. For longer jobs, this may not make sense.

If you want to set the heartbeat in all queues,

Also, you can set heartbeat for a separate queue

Stats

One nice feature of qless is that you can get statistics about usage. Stats are aggregated by day, so when you want stats about a queue, you need to say what queue and what day you're talking about. By default, you just get the stats for today. These stats include information about the mean job wait time, standard deviation, and histogram. This same data is also provided for job completion:

Time

Redis doesn't allow access to the system time if you're going to be making any manipulations to data. However, Qless have heartbeating. When the client making most requests, it actually sends the current time. So, all workers must be synchronized.

Ensuring Job Uniqueness

Qless generate Job ID automatically, but you can set it manually.

For example, Job ID can be based on className and payload. It'll guaranteed that Qless won't have multiple jobs with the same class and data. Also, it helps for debugging on dev environment.

Setting Default Job Options

• jid
• delay
• priority
• tags
• retries
• depends

All of these options have default value. Also, you can define default job options directly on the job class:

Testing Jobs

You can use synchronize to handle jobs for testing. In this regime, all jobs will be running immediately.

Contributing and Developing

Please see CONTRIBUTING.md.

License

qless-php is open-sourced software licensed under the MIT License. See the LICENSE.txt file for more.

© 2018-2022 PDFfiller
© 2013-2015 Ryver, Inc

All rights reserved.