问题标签 [symfony-messenger]

我正在使用微服务方法构建应用程序。对于服务之间的通信，我使用带有 RMQ 传输的 Symfony Messenger。基本上一切正常，但我所有的服务都必须在同一个命名空间中。一旦我试图将它们分成自己的命名空间，比如App\Mail,App\Auth等等，messenger 抱怨缺少 Event 类，因为整个命名空间是在发送到 RMQ 的消息的标头中提供的。有什么方法可以映射来自两个不同名称空间的事件？

例如，Auth应用程序调度事件UserRegistered，因此消息的类型为App\Auth\Event\UserRegistered. 我想在我的邮件应用程序中处理该事件，但信使无法使用它，因为我的事件和处理程序位于App\Mail命名空间下，因此它无法App\Auth\Event\UserRegistered在“邮件”应用程序中找到类。

我得到的示例错误：

在这个确切的示例中，我正在从命名空间下的应用程序发送事件 UserRequestedPasswordReset App，并且我正在尝试将它与App\Mail命名空间下的应用程序一起使用。

我在文档或互联网上找不到任何有用的东西。我试图在容器中别名App\Event\UserRequestedPasswordReset，App\Mail\Event\UserRequestedPasswordReset但没有运气。我猜这对 Denormalizers 来说是可行的，但在互联网上也找不到任何有用的东西。

通信本身正在工作，消息被发送到 RMQ 并在其他服务中接收。我对 RMQ 的设置是：我有多个队列，每个服务一个。我与那些绑定的队列进行了扇出交换。每当我生成事件时，我都会将其发布以交换以将其填充到所有队列中，以便感兴趣的服务可以处理它们。

我的一项服务中的示例信使配置。除了 event 我使用 messenger 来处理 CQRS 命令和查询，所以我使用三种不同的总线。

我想将我的应用程序保留在不同的命名空间中，并且仍然能够处理来自其他服务的事件

我正在尝试使用 Messenger 组件在 rabbit 和 Symfony 中实现交换。

假设我们有交通工具：

消费者如何知道它应该使用哪个处理程序来处理消息？

我正在使用带有 Symfony 4 messenger 组件的工作人员。

这个工人是

• 接收消息（来自rabbitMQ）
• 启动 ffmpeg
• 对视频进行治疗
• 并将某些内容保存在数据库中。

为了在 Symfony 上配置这个 worker，我已经完成了这个（中间件很重要）：

然后为了了解可能导致此问题的原因，我尝试调试信使以查看中间件是否配置正确

一切似乎都很好，对吧？

那么这怎么可能看到这个：

[2019-08-23 10:25:26] messenger.ERROR：重试 App\Api\Message\AMQPvideoFFMPEG - 重试 #1。{"message":"[object] (App\Api\Message\AMQPvideoFFMPEG: {})","class":"App\Api\Message\AMQPvideoFFMPEG","re​​tryCount":1,"error":"[object ]（Doctrine\DBAL\Exception\ConnectionException（代码：0）：驱动程序中发生异常：SQLSTATE [HY000] [2002] 连接超时在 /var/www/html/vendor/doctrine/dbal/lib/Doctrine/DBAL /Driver/AbstractMySQLDriver.php:93, Doctrine\DBAL\Driver\PDOException(code: 2002): SQLSTATE[HY000] [2002] Connection timed out at /var/www/html/vendor/doctrine/dbal/lib/Doctrine/ DBAL/Driver/PDOConnection.php:31, PDOException(code: 2002): SQLSTATE[HY000] [2002] Connection timed out at /var/www/html/vendor/doctrine/dbal/lib/Doctrine/DBAL/Driver/PDOConnection .php:27)"} []

我完全迷路了，我错过了什么吗？

这有时会发生，但它在大多数情况下都有效，我想当我的工作人员失去与数据库的连接时会发生这个错误，特别是如果 ffmpeg 处理持续 7 分钟或更长时间，但这应该通过 ping 和关闭连接的中间件来避免。所以我不清楚这里有什么问题。

似乎很难，并且没有太多关于此的文档（我正在使用 FirendsOfBehat Symfony 扩展）。我想Transport通过使用方法测试是否携带任何事件，get()但我没有得到任何结果。感觉它没有路由正确的总线。

我的packages/messenger.yaml配置：

这是调度我的事件的类

这是我的 service_test.yaml 文件，在运行 Behat 测试时会考虑该文件：

检查我的日志，我可以看到信使确实发送了事件：

[2019-08-30 14:14:50] messenger.INFO: 使用 Symfony\Component\Messenger\Transport\InMemoryTransport 发送消息 App\EventAdapter {"message":"[object] (App\EventAdapter: {})", "类":"App\EventAdapter","sender":"Symfony\Component\Messenger\Transport\InMemoryTransport"} []

我已经集成了 symfony messenger 包，我正在尝试对该消息进行编码。它在 Symfony 3.4.4 版本中工作。但是它在 Symfony 3.4.28 版本中给出了上述错误。我在 symfony serialzer 组件中进行了跟踪，似乎 jsonEncoder 没有在 serialzer.php 中列出，这导致了这个问题。

在 symfony 序列化程序组件中排除 json 编码器的原因是什么。见下文 DoctrineTransportSender：

尝试配置 symfony messenger 时总是出错No transport supports the given Messenger DSN "doctrine://default".

symfony 框架：4.3.4

信使：4.3.0

在这个应用程序中，只安装了学说包和 symfony-messenger。

教义：

信使：

我创建消息组（例如 G1、G2、...）。

在每个组内，所有消息都可以并行处理，顺序无关紧要。

但是，要处理 G2，来自 G1 的所有消息必须已完成处理。

是否有可能在rabbitmq中实现这一点？（最后我将在 symfony 中实现它......）

我的问题与具有“消息屏障”功能的消息队列有关？

我正在使用 Symfony Messenger，我想在MyHandler.

这就是我进行的方式，并且效果很好：

问题是，当我尝试$message像这样直接将 my 传递给处理程序时：

我最终得到以下错误：

调用未定义的方法 App\Tests\MyTestClass::handler()

有人知道幕后发生了什么吗？

Messenger通过 Symfony 组件分派事件和通过 Symfony组件分派事件之间的基本区别是什么EventDispatcher？

我希望能够轻松地同步和异步处理 api 调用。

所以我用相应的处理程序和基类创建了一个空接口

好的，那么我有空子类来区分异步和同步

选择发生在messenger.yaml

好的，这确实有效，但是在调度同步有效负载时出现错误：

在我的日志文件中，我看到了几十条这样的消息：

[2019-10-27 22:11:14] messenger.CRITICAL：处理消息 App\Entity\Messages\ApiPayloadSync 时引发错误。重试 3 次后从传输中移除。错误：“没有消息“App\Entity\Messages\ApiPayloadSync”的处理程序。” {"message":"[object] (App\Entity\Messages\ApiPayloadSync: {"payload":{"action":"mdcadd",...})","class":"App\Entity\Messages\ ApiPayloadSync","re​​tryCount":3,"error":"没有消息处理程序 "App\Entity\Messages\ApiPayloadSync"。","exception":"[object] (Symfony\Component\Messenger\Exception\NoHandlerForMessageException(code : 0): 没有消息“App\Entity\Messages\ApiPayloadSync”的处理程序。在 /var/www/xxx/vendor/symfony/messenger/Middleware/HandleMessageMiddleware.php:75)"} []

因此，尽管我将框架配置为不执行任何操作（对于同步传输），但框架确实开始重试