问题标签 [spring-amqp]

我看过 AmqpJobSample

http://docs.spring.io/spring-batch/spring-batch-samples/#AmqpJobFunctionalTests

它使用 AmqpItemReader 从 rabbitmq 消息队列中读取，使用消息处理程序对其进行处理，然后使用 AmqpItemWriter 写回队列。

我的用例是，从我已经实现的 listItemReader 读取项目，然后将项目发送到消息队列，该队列将由运行在不同服务器（rabbitmq 集群）上的消费者处理，并且这些消费者将响应 iterm 是否成功处理每一个项目。

然后我将从消息队列中读取它（在 itemWriter 中），如果检查项目处理的状态，如果成功则从 itemWriter 返回状态，如果失败则抛出异常。

我的应用程序在 PHP Symfony 框架中，但我正在为当前在 shell 脚本中的批处理引入 Spring Batch。

我正在使用 Spring Batch 来生成消息，但我的消费者使用的是在不同服务器上运行的 PHP。我只想将 id 作为消息传递，并希望在不同服务器上使用 PHP 消费者处理 id。

如何进行远程分块或分区？

http://docs.spring.io/spring-batch/reference/html/scalability.html

是否有任何使用消息（例如 RabbiyMq）进行远程分块和分区的示例？

我怎样才能使用 AmqpItemReader 和 AmqpItemWriter ？我需要使用请求/回复吗？我可以设置回复队列吗？最好的方法是什么？

我正在尝试使用 spring amqp 使用rabbitmq，下面是我的配置。

这是一个简单的消息监听器类，

这是生产者（即春季批次的 itemWriter），

当我运行我的春季批处理作业时， ImportItemWriter.write 被调用。但是 ImportMessageListener.onMessage 不起作用。它不打印消息。我得到控制台上所有项目的低于输出

我正在尝试简单的 spring XD 应用程序在 HDFS 中加载日志事件。spring-ampq/rabbit log4j appender我已经使用（类）配置了目标应用程序，org.springframework.amqp.rabbit.log4j.AmqpAppender以将日志消息泵送到预先配置的交换器。我设置了以下流以从 HDFS 中提取这些消息并将它们推送到 HDFS，其中 soruce 和 sink 模块都是现成的 XD 模块，

流定义，

创建并部署了新的流“demoQ1”

AMQP Appender 正在发布要交换的消息并将其路由到 demoQ1 队列，其中 rabbit 源正在接收第一条消息然后卡住，因为它不确认消息。可能是什么原因？

我正在使用弹簧批处理和弹簧 amqp。我在我的春季批处理 ItemWriter 中使用

在消费者方面，我使用 POJO 使用 MessageListenerAdapter 处理消息。

我想要 template.convertSendAndReceive(item.toString()); 抛出消息处理程序抛出的相同异常，以便spring批处理可以停止批处理，将其标记为失败并记录异常？

我看过这个但不知道如何实现上述用例？

http://docs.spring.io/spring-amqp/docs/1.3.1.RELEASE/reference/html/amqp.html

我该怎么做呢 ？

我正在尝试使用 spring amqp 使用rabbitmq，下面是我的配置。

这是一个简单的消息监听器类，

这是生产者（即春季批次的 itemWriter），

当我运行我的春季批处理作业时，每条消息都会被一一发送和处理，我在响应中得到了回复

它应该发送 5 条消息并将它们排队，并且 5 个消费者线程（并发 = 5）应该同时处理它们，并且应该在它完成后立即响应

我不希望生产者等待第一条消息的回复来排队第二条消息。

我尝试使用 convertAndSend 使其异步（不等待回复）但是如何在 itemWriter 中获得回复消息，就像我可以使用 convertSendAndReceive 一样？

如果我将模板配置更改为

如果我使用 template.convertAndSend(item.toString()); 那我怎样才能得到回复消息？

我无法将自己的消息处理程序附加到此侦听器以获取回复消息，就像我们可以在消费者端附加的方式一样。对于回复，它采用默认的 RabbitmqTemplate 处理程序。

我正在使用 spring amqp rabbitmq，并使用发送消息

我的消息处理程序是

我已经使用

但我越来越低于执行

并且类似的问题也在生产者端replyHandler

另外，如果 importExchange 有任何异常，我如何在 replyHandler 中获得异常？

我正在将所有消息排队到 rabbitmq 队列并在远程服务器上处理这些消息。下面是我在同一个班级的制作人和回复处理程序。

我正在以 write 方法发送所有消息并在 onMessage 中获得回复。这工作正常，但 write 不等待回复，它返回给调用者，并且 spring-batch 步骤标记为已完成。

但我希望进程在发送所有消息后等待回复，直到我们在 onMessage 中获得所有回复。我们应该怎么做 ？

我在我的 spring 批处理项目编写器中使用 spring amqp 模板将消息添加到 rabbitmq 队列。

消息将在不同的远程服务器上处理。我正在尝试处理如果我的消息处理失败（由于某些异常）步骤执行将停止的用例。

我想清除该队列中的所有剩余消息，以便队列中的剩余消息不应被消耗和处理，因为它们也会失败。

如果该步骤失败，我的项目编写器将再次对所有消息进行排队，因此我需要在任何异常情况下清除所有剩余消息。

如何使用 spring amqp 清除队列？

我的容器 XML 配置：

并且myListener只是一个类

我已经concurrency="10"在我的 XML 中指定了。这究竟是什么意思？

我找到了一些文档。他们没有那么有帮助，说明：

指定要创建的并发使用者的数量。默认值为 1。

我感兴趣的是是否MyListener必须是线程安全的，即

• 是否有许多线程创建了许多实例或使用了单个实例？
• 我可以访问没有同步的实例字段吗？
• 是SomeDependency dependency 实例化一次还是为每个线程/实例实例化？
• 是否dependency需要线程安全？

我使用RabbitMQ和Spring AMQP发送消息。

我有这个片段：

当我对单个VHOST.

我必须从 1 个队列接收消息并将它们重新发送到 N 个交换器（在许多不同VHOST的 s 上）。我该怎么做Spring AMQP (ver 1.2)？

在我的 xml 上下文中，有什么比手动定义<rabbit:connection-factory>（对于每个 VHOST）和相关元素更好的东西吗？