hyperf Error 处理 和 队列异常重试处理的问题

消息队列配置

config/autoload/async_queue.php

这里可以直接参考 default 拷贝多个队列配置即可

每个新的配置都为一个新的队列，与其他队列互不干扰

注意点: 如果多个项目 并且使用同个队列名称 且同个 redis 库，会造成读取互相的数据，造成意料之外的错误
建议不同的项目使用不同的 DB

列出常用的代码

use Hyperf\AsyncQueue\Driver\DriverFactory;
use Hyperf\Utils\ApplicationContext;

...

/**
 * 获取一个实例
 * DateTime: 2021/11/30 9:34 上午
 */
public static function getDriver($driver = 'default')
{
    return ApplicationContext::getContainer()->get(DriverFactory::class)->get($driver);
}

----
use Hyperf\Logger\LoggerFactory;
use Hyperf\Utils\ApplicationContext;
use Psr\Log\LoggerInterface;

/**
 * 获取 Logger 实例
 * DateTime: 2021/12/3 9:47 上午
 * @param string $name
 * @return LoggerInterface
 */
public static function get(string $name = 'default'): LoggerInterface
{
    return ApplicationContext::getContainer()->get(LoggerFactory::class)->get($name, $name);
}

---
use Hyperf\Redis\RedisFactory;
use Hyperf\Utils\ApplicationContext;

/**
 * 获取 Logger 实例
 * DateTime: 2021/12/3 9:47 上午
 * @param string $name
 * @return LoggerInterface
 */
public static function get(string $name = 'default'): LoggerInterface
{
    return ApplicationContext::getContainer()->get(LoggerFactory::class)->get($name, $name);
}

以上为获取常用的几个实例的方式

设置异常接管

# AppExceptionHandler.php

/**
 * @var StdoutLoggerInterface
 */
protected $logger;

public function __construct(StdoutLoggerInterface $logger)
{
    $this->logger = $logger;
}

public function handle(Throwable $throwable, ResponseInterface $response)
{
    $this->logger->error(sprintf('%s[%s] in %s', $throwable->getMessage(), $throwable->getLine(), $throwable->getFile()));
    $this->logger->error($throwable->getTraceAsString());
    $this->errorToFile($throwable);
    return $response->withHeader('Server', 'Hyperf')->withStatus(500)->withBody(new SwooleStream('Internal Server Error.'));
}

public function isValid(Throwable $throwable): bool
{
    dump("---------------------");
    return true;
}

public function errorToFile(Throwable $throwable)
{
    $logger = Logger::get("error");
    $logger->error(sprintf('%s[%s] in %s', $throwable->getMessage(), $throwable->getLine(), $throwable->getFile()));
    $logger->error($throwable->getTraceAsString());
    ErrorHandle::handle($throwable);
}

这里演示的结果包含 错误投递的处理 及 重试

namespace App\Job;

use App\Container\Logger;
use Hyperf\AsyncQueue\Job;

class TestJob extends Job
{
protected $maxAttempts = 2;
protected $params = [];

    public function __construct(array $params)
    {
        // 这里最好是普通数据，不要使用携带 IO 的对象，比如 PDO 对象
        // 即为 不要传递 类似如 new Job() | New User() 这种实例化对象进来，这个地方如 队列的时候 会进行 serialize 序列化，不然后续读取需要解析其命名空间和实例对象 无法进行解析的
        $this->params = $params;
    }

    public function handle()
    {
        // ... 执行我们的业务逻辑
        Logger::get('error')->info('手动触发 Error' . format_now(), $this->params);
        throw new \Exception("手动触发 Error");
    }
}

测试后发现，我们无法无法在 AppExceptionHandler 监听到我们的异常 【这个其实是正常的，他针对的是 http 的异常接管】

我们需要手动创建事件监听器，这里就不再细说，直接贴代码吧

抛出异常 和 监听 队列异常

想要监听到队列的异常，我们只需要在我们队列的业务逻辑处理中 抛出我们的 error 即可

public function handle()
{
    // ... 执行我们的业务逻辑
    Logger::get('error')->info('手动触发 Error' . format_now(), $this->params);
    // 抛出 error
    throw new \Exception("手动触发 Error");
}


# App\Listener

namespace App\Listener;

use App\Container\Logger;
use App\Services\Notify\ErrorHandle;
use Hyperf\AsyncQueue\Event\FailedHandle;
use Hyperf\Crontab\Event\FailToExecute;
use Hyperf\Event\Annotation\Listener;
use Hyperf\Event\Contract\ListenerInterface;

#[Listener]
class QueueCrontabListener implements ListenerInterface
{
public function listen(): array
{
return [
    FailedHandle::class,
];
}

    /**
     * @param FailedHandle $event
     */
    public function process(object $event)
    {
        $throwable = $event->getThrowable();
        $message = sprintf("[%s] [Queue Error] [%s]", format_now(), $throwable->getMessage());
        ErrorHandle::handle($throwable, [], $message);
    }
}

ErrorHandle::handle 为我写的全局 error 处理器

class ErrorHandle
...
public static function handle(\Throwable $throwable, array $params = [], $message = null)
{
    $exportHtml = ExceptionReport($throwable);
    Logger::get('error')->error("error handle", compact('exportHtml','params', 'message'));
    $message && dingtoo('error')->text($message);

    if(env('APP_ENV') == 'dev') { return; }
    try {
        $html = ExceptionReportHtml($exportHtml, $throwable->getTrace());
        $html = $html . json_encode($params);
        $emails = explode( ',', env('DEVELOP_EMAIL'));
        $service = new MailerService();
        $emails && array_map(function ($email) use ($service, $html){
            $service->smtpMail($email, 'Error', $html);
        }, $emails);

    }catch (\Throwable $e) {
        Logger::get('error')->error("error handle", ExceptionReport($e));
    }
}

• queue配置

本文由 邓尘锋 创作，采用 知识共享署名4.0 国际许可协议进行许可
本站文章除注明转载/出处外，均为本站原创或翻译，转载前请务必署名
最后编辑时间为: Dec 31, 2021 at 11:04 am