Symfony Messenger+Doctrine报‘7 no connection to the server’异常求助

问题描述 #

在Symfony 7.0.4项目中使用Messenger处理消息时出现异常：消息携带实体ID存储于AWS SQS，Handler从远程PostgreSQL数据库加载实体，业务逻辑执行完成后数据库结果正确，但日志抛出SQLSTATE[HY000]: General error: 7 no connection to the server异常，消息被转入失败队列。将相同逻辑移至Symfony命令通过Cron执行时一切正常。

Messenger配置 #

framework:
    messenger:
        failure_transport: failed

        transports:
            async:
                dsn: 'https://sqs.%env(AWS_REGION)%.amazonaws.com/test-messenger.fifo?access_key=%env(AWS_ACCESS_KEY)%&secret_key=%env(AWS_URLENCODED_SECRET_KEY)%'
                retry_strategy:
                    max_retries: 0
                    multiplier: 2
            failed: 'https://sqs.%env(AWS_REGION)%.amazonaws.com/test-messenger-failed.fifo?access_key=%env(AWS_ACCESS_KEY)%&secret_key=%env(AWS_URLENCODED_SECRET_KEY)%'

        routing:
            App\Messenger\Message\NewValidCertificateForCustomDomainMessage: async

Worker配置（Supervisor） #

[program:messenger-consume]
command=php /var/www/bin/console messenger:consume async --memory-limit=128M -vv --time-limit=600
user=root
numprocs=4
startsecs=0
autostart=true
autorestart=true
startretries=10
stopwaitsecs=5
process_name=%(program_name)s_%(process_num)02d

消息类 #

<?php

declare(strict_types=1);

namespace App\Messenger\Message;

class NewValidCertificateForCustomDomainMessage
{
    public function __construct(
        private string $accessTokenId,
    ) {
    }

    public function getAccessTokenId(): string
    {
        return $this->accessTokenId;
    }
}

Handler类 #

<?php

declare(strict_types=1);

namespace App\Messenger\Handler;

use App\Entity\AccessToken;
use App\Messenger\Message\NewValidCertificateForCustomDomainMessage;
use App\Service\CustomDomainService;
use Doctrine\ORM\EntityManagerInterface;
use Psr\Log\LoggerInterface;
use Symfony\Component\Messenger\Attribute\AsMessageHandler;
use Symfony\Component\Uid\Uuid;

#[AsMessageHandler]
class NewValidCertificateForCustomDomainHandler
{
    private const CONTEXT = 'NewValidCertificateForCustomDomainHandler';

    public function __construct(
        private CustomDomainService $customDomainService,
        private EntityManagerInterface $entityManager,
        private LoggerInterface $logger,
    ) {
    }

    public function __invoke(NewValidCertificateForCustomDomainMessage $newValidCertificateForCustomDomainMessage): void
    {
        if ($this->entityManager->getConnection()->isConnected() === false) {
            $this->entityManager->getConnection()
                ->close();
        }

        $accessToken = $this->entityManager->getRepository(AccessToken::class)->find(
            Uuid::fromString($newValidCertificateForCustomDomainMessage->getAccessTokenId())
        );

        if (!$accessToken instanceof AccessToken) {
            $this->logger->error(
                'AccessToken not found!',
                [
                    'context' => self::CONTEXT,
                    'accessTokenId' => $newValidCertificateForCustomDomainMessage->getAccessTokenId(),
                ]
            );

            throw new \RuntimeException('AccessToken not found!');
        }

        if (!$this->customDomainService->addCustomDomainToDistribution($accessToken)) {
            throw new \RuntimeException('Failed to add custom domain to CloudFront distribution!');
        }
    }
}

异常日志 #

{
    "message": "Error thrown while handling message {class}. Removing from transport after {retryCount} retries. Error: \"{error}\"",
    "context": {
        "class": "App\\Messenger\\Message\\NewValidCertificateForCustomDomainMessage",
        "retryCount": 0,
        "error": "Handling \"App\\Messenger\\Message\\NewValidCertificateForCustomDomainMessage\" failed: An exception occurred while executing a query: SQLSTATE[HY000]: General error: 7 no connection to the server",
        "exception": {
            "class": "Symfony\\Component\\Messenger\\Exception\\HandlerFailedException",
            "message": "Handling \"App\\Messenger\\Message\\NewValidCertificateForCustomDomainMessage\" failed: An exception occurred while executing a query: SQLSTATE[HY000]: General error: 7 no connection to the server",
            "code": 7,
            "file": "/var/www/vendor/symfony/messenger/Middleware/HandleMessageMiddleware.php:124",
            "previous": {
                "class": "Doctrine\\DBAL\\Exception\\DriverException",
                "message": "An exception occurred while executing a query: SQLSTATE[HY000]: General error: 7 no connection to the server",
                "code": 7,
                "file": "/var/www/vendor/doctrine/dbal/src/Driver/API/PostgreSQL/ExceptionConverter.php:80",
                "previous": {
                    "class": "Doctrine\\DBAL\\Driver\\PDO\\Exception",
                    "message": "SQLSTATE[HY000]: General error: 7 no connection to the server",
                    "code": 7,
                    "file": "/var/www/vendor/doctrine/dbal/src/Driver/PDO/Exception.php:28",
                    "previous": {
                        "class": "PDOException",
                        "message": "SQLSTATE[HY000]: General error: 7 no connection to the server",
                        "code": 0,
                        "file": "/var/www/vendor/doctrine/dbal/src/Driver/PDO/Statement.php:55"
                    }
                }
            }
        }
    },
    "level": 500,
    "level_name": "CRITICAL",
    "channel": "messenger",
    "datetime": "2024-03-25T00:05:02.362953+00:00",
    "extra": {}
}

已尝试无效的方法 #

• 在Handler中重置数据库连接（仅判断未连接时关闭，未重新连接）
• 缩短Worker超时时间至600秒

解决方案 #

1. 修复Handler中的连接处理逻辑 #

当前代码仅在未连接时关闭连接，未重新建立连接，改为：

// 替换原连接判断代码
$connection = $this->entityManager->getConnection();
if (!$connection->isConnected()) {
    $connection->connect();
}
// 或者强制重新连接（避免连接超时失效）
$connection->close();
$connection->connect();

2. 配置Doctrine连接的空闲超时自动重连 #

在config/packages/doctrine.yaml中添加连接参数，让Doctrine自动检测并重建失效连接：

doctrine:
    dbal:
        default_connection: default
        connections:
            default:
                # 原有配置...
                keepalives: true
                keepalives_idle: 60
                keepalives_interval: 10
                keepalives_count: 5
                # 开启Doctrine自动重连
                retry_attempts: 3

3. 调整Worker的进程生命周期 #

结合--limit参数限制每个Worker处理的消息数，避免长期运行导致连接失效：

command=php /var/www/bin/console messenger:consume async --memory-limit=128M -vv --time-limit=600 --limit=100

4. 添加Messenger中间件处理连接 #

自定义中间件，在每个消息处理前确保数据库连接有效：

<?php

namespace App\Messenger\Middleware;

use Doctrine\ORM\EntityManagerInterface;
use Symfony\Component\Messenger\Middleware\MiddlewareInterface;
use Symfony\Component\Messenger\Envelope;
use Symfony\Component\Messenger\Middleware\StackInterface;

class EnsureDatabaseConnectionMiddleware implements MiddlewareInterface
{
    public function __construct(private EntityManagerInterface $entityManager)
    {
    }

    public function handle(Envelope $envelope, StackInterface $stack): Envelope
    {
        $connection = $this->entityManager->getConnection();
        if (!$connection->isConnected()) {
            $connection->connect();
        }

        return $stack->next()->handle($envelope, $stack);
    }
}

然后在config/packages/messenger.yaml中注册中间件：

framework:
    messenger:
        # 原有配置...
        middleware:
            - App\Messenger\Middleware\EnsureDatabaseConnectionMiddleware
            # 保留其他默认中间件

内容的提问来源于stack exchange，提问作者KdoSiOdJinud