我在做这样的事
from(rabbitmq:pollingQueue?prefetchSize=1&concurrentConsumer=10)
.process(pollingRequestStatus) // check status of the request, if not ready, requeue = true
.Choice
.when(requeue == true) // request not ready
.to(rabbitmq:pollingQueue)//back to the
需要在多线程环境中执行@Handler(导入org.apache.camel.Handler)方法。下面是我的当前代码和camelroute.xml文件。任何想法或建议都是值得赞赏的。
@Component("messagehandler")
public class HandleMessages {
@Handler
public void handle(String body, Exchange exchange) throws Exception {
// do some business operation
}
}
这是我编写的从队列中读取消息的类。
public class Sender {
public static void main(String[] args) throws Exception {
CamelContext camelContext = new DefaultCamelContext();
camelContext.addRoutes( new RouteBuilder() {
@Override
public void configure() throws Exception {
我目前正在尝试使用Talend open studio for ESB为RabbitMQ队列构建一个使用者。在阅读了我能找到的关于这个主题的每一篇文章后,我有两个不同的版本,但没有一个是实际有效的。
版本1:
我的路线:
cMessagingEndpoint1 URI:
"rabbitmq://host:port/exchange?username=xxxxx&password=xxxx&queue=xxxx“
我还尝试向此URI添加耐久= true和no_declare = true,但随后收到未知参数错误。所以我再次删除了这两个。
控制台输出:这似乎是旧语法,
有一个问题,我已经注意到,当配置InOut能力的路线在骆驼骆驼-狂犬病扩展。当我将主队列配置设置为autoAck=false时,对临时应答队列也会复制相同的配置(它甚至使用相同的预取(5)设置,很容易在RabbitMQ控制台中看到)。这将导致临时队列中的消息无限期地停留在那里,直到服务器重新启动为止。
Virtual host Name Features State Ready Unacked Total incoming deliver / get ack
/test amq.gen-Hkdx9mckIfMc6JhDI6d-JA A