115 lines
3.1 KiB
PHP
115 lines
3.1 KiB
PHP
|
<?php
|
||
|
|
|
||
|
|
declare(strict_types=1);
|
||
|
|
|
||
|
|
namespace App\Platform;
|
||
|
|
|
||
|
|
use Hyperf\Amqp\Annotation\Producer;
|
||
|
|
use Hyperf\Amqp\Message\ProducerMessage;
|
||
|
|
use PhpAmqpLib\Message\AMQPMessage;
|
||
|
|
use Throwable;
|
||
|
|
|
||
|
|
/**
|
||
|
|
* 错误消息生产者
|
||
|
|
*
|
||
|
|
* 用于将处理失败且超过重试次数的消息发送到 errors.queue
|
||
|
|
* 使用 errors.exchange 进行路由
|
||
|
|
*/
|
||
|
|
#[Producer(exchange: 'errors.exchange', routingKey: 'error')]
|
||
|
|
class ErrorProducer extends ProducerMessage
|
||
|
|
{
|
||
|
|
/**
|
||
|
|
* 使用 consumer 连接池（因为是在 consumer 中调用）
|
||
|
|
*/
|
||
|
|
protected string $poolName = 'default_consumer';
|
||
|
|
|
||
|
|
/**
|
||
|
|
* VHost
|
||
|
|
*/
|
||
|
|
protected string $vhost = 'dataflow';
|
||
|
|
|
||
|
|
/**
|
||
|
|
* 消息持久化
|
||
|
|
*/
|
||
|
|
protected array $properties = [
|
||
|
|
'delivery_mode' => 2, // 持久化消息
|
||
|
|
];
|
||
|
|
|
||
|
|
/**
|
||
|
|
* Exchange 配置
|
||
|
|
*/
|
||
|
|
protected string $exchange = 'errors.exchange';
|
||
|
|
|
||
|
|
/**
|
||
|
|
* Routing Key 配置
|
||
|
|
*/
|
||
|
|
protected string|array $routingKey = 'error';
|
||
|
|
|
||
|
|
/**
|
||
|
|
* 构造错误消息
|
||
|
|
*
|
||
|
|
* @param AMQPMessage $originalMessage 原始失败的消息
|
||
|
|
* @param Throwable $error 错误异常
|
||
|
|
* @param int $retryCount 重试次数
|
||
|
|
*/
|
||
|
|
public function __construct(
|
||
|
|
AMQPMessage $originalMessage,
|
||
|
|
Throwable $error,
|
||
|
|
int $retryCount
|
||
|
|
) {
|
||
|
|
$this->payload = $this->buildErrorMessage($originalMessage, $error, $retryCount);
|
||
|
|
}
|
||
|
|
|
||
|
|
/**
|
||
|
|
* 构建错误消息格式
|
||
|
|
*
|
||
|
|
* @param AMQPMessage $originalMessage 原始消息
|
||
|
|
* @param Throwable $error 错误信息
|
||
|
|
* @param int $retryCount 重试次数
|
||
|
|
* @return array
|
||
|
|
*/
|
||
|
|
protected function buildErrorMessage(
|
||
|
|
AMQPMessage $originalMessage,
|
||
|
|
Throwable $error,
|
||
|
|
int $retryCount
|
||
|
|
): array {
|
||
|
|
$originalData = json_decode($originalMessage->getBody(), true);
|
||
|
|
|
||
|
|
return [
|
||
|
|
'error_id' => $this->generateErrorId(),
|
||
|
|
'original_message' => $originalData,
|
||
|
|
'error' => [
|
||
|
|
'type' => get_class($error),
|
||
|
|
'message' => $error->getMessage(),
|
||
|
|
'code' => $error->getCode(),
|
||
|
|
'file' => $error->getFile(),
|
||
|
|
'line' => $error->getLine(),
|
||
|
|
'trace' => $error->getTraceAsString(),
|
||
|
|
'timestamp' => date('c'),
|
||
|
|
],
|
||
|
|
'metadata' => [
|
||
|
|
'platform' => $originalData['platform'] ?? 'unknown',
|
||
|
|
'platform_id' => $originalData['meta']['platform_id'] ?? null,
|
||
|
|
'company_id' => $originalData['meta']['company_id'] ?? null,
|
||
|
|
'store_id' => $originalData['meta']['store_id'] ?? null,
|
||
|
|
'data_type' => $originalData['data_type'] ?? 'unknown',
|
||
|
|
'message_id' => $originalData['message_id'] ?? null,
|
||
|
|
'failed_at' => date('c'),
|
||
|
|
'retry_count' => $retryCount,
|
||
|
|
],
|
||
|
|
];
|
||
|
|
}
|
||
|
|
|
||
|
|
/**
|
||
|
|
* 生成错误ID
|
||
|
|
*
|
||
|
|
* 格式: err_{timestamp}_{random}
|
||
|
|
*
|
||
|
|
* @return string
|
||
|
|
*/
|
||
|
|
protected function generateErrorId(): string
|
||
|
|
{
|
||
|
|
return uniqid('err_', true);
|
||
|
|
}
|
||
|
|
}
|