问题标签 [symfony-messenger]

我不确定这是否可能：为同一个队列拥有一个信使消费者池？

我尝试使用 Redisconsumer=选项，但事实并非如此。

也许“池”中间件可以对特定传输进行一些路由？

我需要在 Symfony 上编写的应用程序中有多个总线。为此，我正在使用信使包。

所以我需要：

• external_events_bus - 从外部系统接收消息，使用 sqs 传输
• application_events_bus - 用于应用程序事件（目前使用数据库驱动程序）
• domain_events_bus - 用于域事件（目前使用数据库驱动程序）
• command_bus - 用于命令和查询 (CQRS)（当前同步）

所以在我的messenger.yaml我有：

在我的services.yaml

在我只有命令总线和外部总线之前想要澄清一切都很好，但是在我添加域和应用程序总线之后，somethinkg 出错了。

在我使用ExternalEventBus的测试中，测试外部事件是否正确处理，如下所示：

我收到类似的错误

但是当我bin/console debug:messenger同时为开发和测试环境运行时，我可以看到所有事件及其处理程序之间的正确映射。

但是当我调试它可以看到HandlersLocator::getHandlers()出于SomeExternalEvent某种原因返回与命令总线而不是外部事件总线相关的处理程序列表（可能是因为命令总线设置为默认总线）

我知道解释和展示主题有点困难，但我已经花了一整天的时间来寻找它的问题所在。任何帮助表示赞赏，谢谢！！！

抱歉有任何错别字，我在这个例子中更改了类的名称，可能有一些错别字

我正在尝试测试我通过 Symfony 中的 Messenger 处理一些数据的过程。

我尝试以下方法：

首先，我定义了我的工人。

然后在此之后，一些代码运行，向信使发送消息。我测试了这段代码，效果很好。通过转储 $transport->get() 我看到我的消息已成功发送到工作人员和数据库。

现在在我的单元测试结束时，我通过 $worker->run() 运行 worker。由于信使只处理几个动作/查询，它应该只需要一点时间就可以完成。但看起来它只是在那里闲置。我在 $worker->run() 之后直接调用 $worker->stop() 但它没有到达那里。

简短的背景，在应用程序中我们使用doctrine传输失败的队列，所以如果消息失败（所有重试失败）信使将其存储到失败的消息表中，它正在等待进一步的操作。当消息发送到该表并更新普罗米修斯仪表并发送警报时，我需要以某种方式收听。

我尝试以与发送其他与信使相关的指标相同的方式来执行此操作（使用事件订阅者，我发现它比中间件灵活得多），所以基本上通过听其中一个https://symfony.com/doc/current/messenger.html#messenger-events，但我发现当信使时不可能抓住片刻将它发送到失败传输，信使有SendMessageToTransportsEvent但它没有提供传输名称，所以我不明白它是否失败......

我还发现信使有SentToFailureTransportStamp邮票，但由于某种原因，当消息失败时，信封中不存在邮票，并将被发送到失败的队列，但取决于类描述Stamp applied when a message is sent to the failure transport.它应该......

我唯一的想法是制作一个收集器，它将定期查询失败的消息表并将记录数量发送到普罗米修斯，但是如果我将传输更改为其他东西呢？？？？这意味着我需要一直更换收集器...

有没有办法捕捉信使向失败队列发送消息的时刻？？？谢谢指教！！！

我已经实现了 symfony messenger https://symfony.com/doc/current/messenger.html。我想在handler实现的类中获得重试次数MessageHandlerInterface

Messenger 消费者命令抛出错误并根据配置重试 5 次。所以我想在处理程序中获取重试计数，以便我可以相应地添加一些逻辑。

我还尝试创建此答案https://stackoverflow.com/a/61078739中提到的中间件。但是这个中间件没有被执行。

我有一个包装在 docker 中的 symfony 项目，并配置了一个 Redis 队列。

配置：

信使.yaml

处理程序：

但是当我运行命令时

我看到无论我运行哪个接收器以及优先级如何，都会处理 MessagePriorityLow，也就是说，如果我在队列中有两条消息，那么当我运行命令时

首先处理那些首先进入队列的消息

另外，如果我在 Hanlker 中明确指定 from_transport => async_priority_low 为低优先级，并运行命令

我收到没有 async_priority_low 处理程序的错误

我究竟做错了什么？

为了发送邮件等，我们在项目中使用消息。如果消息失败（由于消息处理程序代码中的错误），它将存储在messenger_messages表中。

该messenger_messages表包含一个名为 的列body。这是body专栏的一部分（它包含更多，但它真的很大）。该主体由 Symfony/PHP（或 RabbitMQ 包，不确定究竟是哪个）编码。

我怎样才能解码/反序列化这个身体？我尝试使用unserialize:(请记住，这只是为了测试，这不是最终代码0

但这不起作用并给出此错误，例如：

我们正在使用 Symfony Messenger，并且有这些传输：

当我们向队列发送消息async，并且最后一次重试失败并出现异常时，异常被记录到MessengerMessages表中，并且异常冒泡（在我们的例子中进入Sentry）。这就是我们想要的。

然而，当我们向队列发送消息asyncLowPriority时，我们希望失败的消息发送到：

• 没到failed交通工具
• 不要让异常冒泡

基本上，应该删除异常。

这可能吗，怎么做？

原因是我们使用这个队列来异步下载图像，并且我们已经在命令处理程序的专用数据库表中记录了每个失败。

我有一个 Symfony 应用程序，它在负载均衡器后面的几台服务器上运行。所以我有单独的主机 www1、www2、www3 等。

目前我messenger:consume只在 www1 上运行，担心竞争条件和潜在的消息被处理两次。

现在我有一个场景，我需要在每个主机上执行一个命令。

我正在考虑为每个主机使用单独的传输并messenger:consume在每个主机上运行，​​只使用来自其各自队列的消息。但是我希望配置是动态的，即我不想在添加或删除新主机时使用不同的传输配置进行另一个代码发布。

你能提出一个实现这一目标的策略吗？

编辑：这个问题出现在试图在同一个应用程序中同时拥有同步和同步电子邮件的过程中。没有说清楚。在撰写本文时，这是不可能的，至少不像这里尝试的那样简单。请参阅下面@msg 的评论。

配置为异步发送电子邮件的电子邮件服务，而是立即发送电子邮件。这发生在doctrine或amqp选择为MESSENGER_TRANSPORT_DSN。doctrine传输成功创建messenger_messages表，但没有内容。这告诉我MESSENGER_TRANSPORT_DSN观察到了。使用 RabbitMQ 'Hello World' 教程的简单测试amqp表明它配置正确。

我在下面的代码中遗漏了什么？

如下所示的序列摘要：添加机会 ->OppEmailService创建电子邮件内容 ->TemplatedEmail()从EmailerService（未显示）获取对象 -> 将TemplatedEmail()对象提交到LaterEmailService，配置为异步。

信使.yaml：

OpportunityController：

OppEmailService：

LaterEmailService：