Support

中文说明

A comprehensive support component for Hyperf providing essential utilities, helpers, and base classes.

Features

• 🎯 Fluent Dispatch API - Elegant job dispatch with support for async queue, AMQP, and Kafka
• 🔄 Closure Jobs - Execute closures as background jobs with dependency injection
• 🛠️ Helper Functions - Collection of useful helper functions
• 📦 Bus System - Pending dispatch classes for various message systems
• 🧩 Traits & Utilities - Reusable traits and utility classes
• ⏱️ Backoff Strategies - Multiple retry backoff implementations for retry mechanisms

Installation

composer require friendsofhyperf/support

Usage

Dispatch Helper Function

The dispatch() helper function provides a fluent API for dispatching jobs to different systems:

Async Queue (Closure Jobs)

usefunctionFriendsOfHyperf\Support\dispatch;
// Simple closure dispatch to async queuedispatch(function () {
// Your job logic herelogger()->info('Job executed!');
});
// With configurationdispatch(function () {
// Your job logic here
})
->onConnection('high-priority')
->delay(60) // Execute after 60 seconds
->setMaxAttempts(5);
// With dependency injectiondispatch(function (UserService$userService, LoggerInterface$logger) {
$users = $userService->getActiveUsers();
$logger->info('Processing ' . count($users) . ' users');
});

AMQP Producer Messages

useHyperf\Amqp\Message\ProducerMessageInterface;
usefunctionFriendsOfHyperf\Support\dispatch;
// Dispatch AMQP messagedispatch($amqpMessage)
->setPool('default')
->setExchange('my.exchange')
->setRoutingKey('my.routing.key')
->setTimeout(10)
->setConfirm(true);

Kafka Producer Messages

useHyperf\Kafka\Producer\ProduceMessage;
usefunctionFriendsOfHyperf\Support\dispatch;
// Dispatch Kafka messagedispatch($kafkaMessage)
->setPool('default');

CallQueuedClosure

The CallQueuedClosure class allows you to execute closures as async queue jobs:

useFriendsOfHyperf\Support\CallQueuedClosure;
// Create a closure job$job = CallQueuedClosure::create(function () {
// Your job logicreturn'Job completed!';
});
// Configure max attempts$job->setMaxAttempts(3);
// The job can be pushed to queue manually or via dispatch()

Pending Dispatch Classes

PendingAsyncQueueDispatch

Fluent API for async queue job dispatch:

useFriendsOfHyperf\Support\Bus\PendingAsyncQueueDispatch;
$pending = newPendingAsyncQueueDispatch($job);
$pending
->onConnection('default')
->delay(30)
->when($condition, function ($dispatch) {
$dispatch->onConnection('special');
})
->unless($otherCondition, function ($dispatch) {
$dispatch->delay(60);
});
// Job is dispatched when object is destroyed

PendingAmqpProducerMessageDispatch

Fluent API for AMQP message dispatch:

useFriendsOfHyperf\Support\Bus\PendingAmqpProducerMessageDispatch;
$pending = newPendingAmqpProducerMessageDispatch($message);
$pending
->setPool('default')
->setExchange('my.exchange')
->setRoutingKey('my.routing.key')
->setTimeout(5)
->setConfirm(true);
// Message is sent when object is destroyed

PendingKafkaProducerMessageDispatch

Fluent API for Kafka message dispatch:

useFriendsOfHyperf\Support\Bus\PendingKafkaProducerMessageDispatch;
$pending = newPendingKafkaProducerMessageDispatch($message);
$pending->setPool('default');
// Message is sent when object is destroyed

Conditional Execution

All pending dispatch classes support conditional execution:

usefunctionFriendsOfHyperf\Support\dispatch;
dispatch($job)
->when($shouldUseHighPriority, function ($dispatch) {
$dispatch->onConnection('high-priority');
})
->unless($isTestMode, function ($dispatch) {
$dispatch->delay(10);
});

API Reference

dispatch($job)

Creates a pending dispatch instance based on the job type:

• Closure → PendingAsyncQueueDispatch with CallQueuedClosure
• ProducerMessageInterface → PendingAmqpProducerMessageDispatch
• ProduceMessage → PendingKafkaProducerMessageDispatch
• Other objects → PendingAsyncQueueDispatch

PendingAsyncQueueDispatch Methods

• onConnection(string $connection): static - Set queue connection
• delay(int $delay): static - Delay job execution (seconds)
• setMaxAttempts(int $attempts): static - Set max retry attempts
• when(mixed $condition, callable $callback): static - Conditional execution
• unless(mixed $condition, callable $callback): static - Inverse conditional execution

PendingAmqpProducerMessageDispatch Methods

• setPool(string $pool): static - Set AMQP pool name
• setExchange(string $exchange): static - Set exchange name
• setRoutingKey(array|string $routingKey): static - Set routing key(s)
• setTimeout(int $timeout): static - Set timeout (seconds)
• setConfirm(bool $confirm): static - Enable/disable confirm mode
• when(mixed $condition, callable $callback): static - Conditional execution
• unless(mixed $condition, callable $callback): static - Inverse conditional execution

PendingKafkaProducerMessageDispatch Methods

• setPool(string $pool): static - Set Kafka pool name
• when(mixed $condition, callable $callback): static - Conditional execution
• unless(mixed $condition, callable $callback): static - Inverse conditional execution

CallQueuedClosure

• create(Closure $closure): static - Create a new closure job
• setMaxAttempts(int $attempts): void - Set max retry attempts
• handle(): mixed - Execute the closure (called by queue worker)

Backoff Strategies

The component provides various backoff strategies for retry mechanisms:

ArrayBackoff

Use custom delay intervals defined in an array:

useFriendsOfHyperf\Support\Backoff\ArrayBackoff;
// Custom delays$backoff = newArrayBackoff([100, 500, 1000, 2000, 5000]);
// Stop after array is exhausted (returns 0)$backoff = newArrayBackoff([100, 500, 1000], false);
// From comma-separated string$backoff = ArrayBackoff::fromString('100, 500, 1000, 2000');
// From predefined patterns$backoff = ArrayBackoff::fromPattern('short'); // [100, 200, 300, 500, 1000]$backoff = ArrayBackoff::fromPattern('medium'); // [200, 500, 1000, 2000, 5000]$backoff = ArrayBackoff::fromPattern('long'); // [500, 1000, 2000, 5000, 10000, 30000]$backoff = ArrayBackoff::fromPattern('exponential'); // [100, 200, 400, 800, 1600, 3200, 6400]// Usage in retry logic$attempt = 0;
while (true) {
try {
returnperformOperation();
} catch (Exception$e) {
$delay = $backoff->next();
if ($delay === 0) {
throw$e; // No more retries
}
usleep($delay * 1000); // Convert to microseconds
}
}

Available Backoff Implementations

• ArrayBackoff - Custom intervals from an array
• FixedBackoff - Constant delay between retries
• LinearBackoff - Linear growth with configurable step
• ExponentialBackoff - Exponential growth with optional jitter
• FibonacciBackoff - Fibonacci sequence-based delays
• PoissonBackoff - Statistical distribution-based delays
• DecorrelatedJitterBackoff - Decorrelated jitter for better spreading

Common Backoff Interface

All backoff implementations implement BackoffInterface:

interface BackoffInterface
{
publicfunctionnext(): int; // Get next delay in millisecondspublicfunctionreset(): void; // Reset attempt counterpublicfunctiongetAttempt(): int; // Get current attempt countpublicfunctionsleep(): int; // Sleep for calculated delay
}

Contact

• Twitter
• Gmail

License

MIT