amqp 相关问题

AMQP是一种开放式消息传递标准。它为高性能企业消息传递定义了线级协议和语义框架。如果您正在使用特定代理（例如：[rabbitmq]），则AMQP标记可能是多余的。 AMQP不应与Apache [ActiveMQ]混淆

我如何在Symfony 5.2上正确配置RabbitMQ，Messenger和AMQP，而无需使用Docker_compose.yml？我想使用Messenger组件和RabbitMQ提交使用异步方法提交表单。提交此表格后，发送电子邮件给我。但是，当我实施RabbitMQ时，它行不通。我肯定有一个

symfony amqp symfony-messenger

回答 0 投票 0

PHP和RABBITMQ 504 GATEWAWAY超时错误在消费者中

我的班级消费者，兔子的等待消息： <?php namespace App\Infrastructure; use App\Application\Interfaces\BankServiceInterface; use App\Application\Interfaces\ConsumerInterface; use App\Application\Interfaces\MailAgentInterface; use PhpAmqpLib\Channel\AMQPChannel; use PhpAmqpLib\Connection\AMQPStreamConnection; class Consumer implements ConsumerInterface { private AMQPStreamConnection $connection; private AMQPChannel $channel; private BankServiceInterface $service; private MailAgentInterface $mailAgent; public function __construct(AMQPStreamConnection $connection, BankServiceInterface $service, MailAgentInterface $mailAgent) { $this->connection = $connection; $this->channel = $connection->channel(); $this->service = $service; $this->mailAgent = $mailAgent; } public function runFromQueue(string $queueName): void { $this->channel->queue_declare( $queueName, false, true, false, false ); $this->channel->basic_qos(null, 1, null); $this->channel->basic_consume( $queueName, '', false, false, false, false, $this->onConsume() ); while(count($this->channel->callbacks)) { $this->channel->wait(); } $this->closeConnection(); } public function closeConnection(): void { $this->channel->close(); $this->connection->close(); } private function onConsume() { return function ($request) { echo '<pre>'; var_dump($request->getBody()); echo '</pre>'; }; } } 初始化这堂课时，他的作品在504 Gateway Tim-Out中失败。但是，如果从方法删除调用方法getBody()：： onConsume 这是工作，并在属性中打印带有消息主体的消息对象。另外，我会收到504个网关超时错误，如果尝试访问此对象中的访问属性private function onConsume() { return function ($request) { echo '<pre>'; var_dump($request); echo '</pre>'; }; ： body 它像方法private function onConsume() { return function ($request) { echo '<pre>'; var_dump($request->body); echo '</pre>'; }; 中的循环一样，循环和服务器按超时范围掉落。尝试用方法runFromQueue()替换loop之后：runFromQueue()但结果将像循环一样，如果调用方法consume()或属性public function runFromQueue(string $queueName): void { $this->channel->queue_declare( $queueName, false, true, false, false ); $this->channel->basic_qos(null, 1, null); $this->channel->basic_consume( $queueName, '', false, false, false, false, $this->onConsume() ); try { $this->channel->consume(); } catch (\Throwable $exception) { echo $exception->getMessage(); } $this->closeConnection(); } ，我会收到504 Gateway超时错误，并且此功能与简单打印getBody()。然后我进行简单的修改，添加类新属性： body 将其放在循环条件下： $request 值此属性更改为private $isMessageRead = true; ： public function runFromQueue(string $queueName): void { $this->channel->queue_declare( $queueName, false, true, false, false ); $this->channel->basic_qos(null, 1, null); $this->channel->basic_consume( $queueName, '', false, false, false, false, $this->onConsume() ); while($this->isMessageRead) { $this->channel->wait(); } $this->closeConnection(); } 这项工作，但我对此解决方案不满意。有什么问题？如果没有财产onConsume()？？这个问题似乎源于消费者如何处理消息确认和连接的生命周期。当您调用private function onConsume() { return function ($request) { echo '<pre>'; var_dump($request->getBody()); echo '</pre>'; $this->isMessageRead = false; }; } 时，它可能正在阻止或无法正确确认消息，从而导致服务器超时。正在发生的事情：方法用于处理传入消息。但是，如果您不正确确认消息（即使用isMessageRead），则服务器可以保持连接打开或处于无响应状态，从而导致超时。您的解决方案是$request->getBody()的，因为您正在手动控制流动，但这不是理想的。要正确修复此此操作而无需使用诸如wait()的标志，请确保您在处理后确认消息并将控件返回到队列。任期代码： $request->ack() 这种方式，在处理消息后，您将确认它，它告诉RabbitMQ已消耗消息，并且连接可以继续进行下一条消息而无需无限期地等待。这应该防止504网关超时问题。通过确保正确的确认，脚本应平稳运行，而无需自定义标志或循环。

php rabbitmq amqp

回答 1 投票 0

单元测试 Symfony Messenger

我一直在我最新的项目中使用 Symfony Messenger 来实现 AMQP。虽然代码的工作让我非常高兴，但我无法为发送消息的方法编写单元测试......

php symfony phpunit amqp symfony-messenger

回答 2 投票 0

如何强制 RabbitMQ 代理 NACK 消息进行测试？

我正在通过 Spring AMQP 使用 RabbitMQ 开发 Spring Boot 项目。我们希望能够在消息发布被 NACK 时测试应用程序的行为（特别是当

java spring rabbitmq amqp

回答 3 投票 0

Spring RabbitMQ - RPC - CorrelationId 不匹配 - TopicExchange - 客户端 - 服务器 - 模型

我有 Spring boot - RabbitMQ 应用程序。使用的交换器是topicexchange。公共静态最终字符串RPC_REQ_QUEUE =“req.queue”; public static Final String RPC_RES_QUEUE = "res.queue&q...