Download the PHP package friendsofhyperf/amqp-job without Composer

Amqp Job

中文说明

Introduction

friendsofhyperf/amqp-job dispatches PHP job objects through hyperf/amqp. A job declares its AMQP binding with the AmqpJob attribute, and the component automatically registers a consumer for each enabled attributed job when the server starts.

The component uses PHP serialization by default, so only dispatch trusted job objects and keep their classes available to consumers.

Installation

The package's ConfigProvider is discovered by Hyperf. It registers the consumer listener and binds the retry-attempt store to Redis and the message packer to PhpSerializerPacker. Configure hyperf/amqp and its selected AMQP pool before dispatching jobs. Setting amqp.enable to false skips all automatic job-consumer registration.

Define And Dispatch A Job

dispatch(JobInterface $payload, ?bool $confirm = null, ?int $timeout = null): bool is a convenience function that creates and publishes a JobMessage. It assigns a unique job ID when the job does not already have one. The optional confirm and timeout arguments override the job's values for that dispatch.

Currently, dispatch() does not copy the job's exchange, routing key, or pool into the JobMessage. The resulting producer message therefore uses an empty exchange, an empty routing key, and Hyperf's default AMQP pool. To publish to the binding declared by AmqpJob, create a JobMessage, set its destination as shown above, and pass it to Hyperf\Amqp\Producer.

Attribute Options

If a job does not use the attribute, its getters fall back to exchange hyperf, routing key hyperf.job, no explicit pool, publisher confirmation disabled, a publish timeout of 5 seconds, and no retries. A subclass may provide protected properties such as $exchange, $routingKey, $poolName, $confirm, $timeout, or $maxAttempts to override those getter values. As noted above, dispatch() only consumes the confirm and timeout getters.

Consumption And Retries

JobConsumer acknowledges a job when handle() returns nothing or an unrecognized string. Returning a Hyperf\Amqp\Result or its string value controls the result directly.

When handle() throws, the consumer logs the exception when a compatible logger is available. With maxAttempts > 0, the Redis-backed attempt counter requeues the job until the configured total attempt count is reached. It then calls fail(Throwable $e) and drops the message. The default fail() method does nothing and can be overridden:

The default Redis attempt keys use the hyperf:amqp-job:attempts: prefix and expire after 86,400 seconds. Redis is therefore required when retries are enabled. Bind FriendsOfHyperf\AmqpJob\Contract\Attempt or FriendsOfHyperf\AmqpJob\Contract\Packer to replace the default implementations. For exception logging, the consumer first resolves FriendsOfHyperf\AmqpJob\Contract\LoggerInterface, then Hyperf\Contract\StdoutLoggerInterface; logging is skipped when neither is available.

Register A Custom Consumer

Automatic consumers are normally sufficient. To consume compatible serialized jobs with a manually declared consumer, extend JobConsumer and use Hyperf's Consumer attribute: