Apache骆驼写到rabbitmq

Question

我正在尝试使用 apache camel 从文件中读取数据并写入 rabbitmq 队列，但最终出现错误

    Exception in thread "main" org.apache.camel.FailedToCreateRouteException: Failed to create route route1 at: >>> To[rabbitmq://localhost:15672?queue=hello] <<< in route: Route(route1)[[From[file://target/?fileName=doctor.txt&chars... because of Failed to resolve endpoint: rabbitmq://localhost:15672?queue=hello due to: No URI path as the exchangeName for the RabbitMQEndpoint, the URI is rabbitmq://localhost:15672?queue=hello
    at org.apache.camel.model.RouteDefinition.addRoutes(RouteDefinition.java:945)
    at org.apache.camel.model.RouteDefinition.addRoutes(RouteDefinition.java:187)
    at org.apache.camel.impl.DefaultCamelContext.startRoute(DefaultCamelContext.java:794)
    at org.apache.camel.impl.DefaultCamelContext.startRouteDefinitions(DefaultCamelContext.java:2184)
    at org.apache.camel.impl.DefaultCamelContext.doStartCamel(DefaultCamelContext.java:1916)
    at org.apache.camel.impl.DefaultCamelContext.doStart(DefaultCamelContext.java:1777)
    at org.apache.camel.support.ServiceSupport.start(ServiceSupport.java:61)
    at org.apache.camel.impl.DefaultCamelContext.start(DefaultCamelContext.java:1745)
    at test.RMQCamelSender.main(RMQCamelSender.java:38)
Caused by: org.apache.camel.ResolveEndpointFailedException: Failed to resolve endpoint: rabbitmq://localhost:15672?queue=hello due to: No URI path as the exchangeName for the RabbitMQEndpoint, the URI is rabbitmq://localhost:15672?queue=hello
    at org.apache.camel.impl.DefaultCamelContext.getEndpoint(DefaultCamelContext.java:545)
    at org.apache.camel.util.CamelContextHelper.getMandatoryEndpoint(CamelContextHelper.java:71)
    at org.apache.camel.model.RouteDefinition.resolveEndpoint(RouteDefinition.java:202)
    at org.apache.camel.impl.DefaultRouteContext.resolveEndpoint(DefaultRouteContext.java:106)
    at org.apache.camel.impl.DefaultRouteContext.resolveEndpoint(DefaultRouteContext.java:112)
    at org.apache.camel.model.SendDefinition.resolveEndpoint(SendDefinition.java:61)
    at org.apache.camel.model.SendDefinition.createProcessor(SendDefinition.java:55)
    at org.apache.camel.model.ProcessorDefinition.makeProcessor(ProcessorDefinition.java:500)
    at org.apache.camel.model.ProcessorDefinition.addRoutes(ProcessorDefinition.java:213)
    at org.apache.camel.model.RouteDefinition.addRoutes(RouteDefinition.java:942)
    ... 8 more
Caused by: java.lang.IllegalArgumentException: No URI path as the exchangeName for the RabbitMQEndpoint, the URI is rabbitmq://localhost:15672?queue=hello
    at org.apache.camel.component.rabbitmq.RabbitMQComponent.createEndpoint(RabbitMQComponent.java:50)
    at org.apache.camel.component.rabbitmq.RabbitMQComponent.createEndpoint(RabbitMQComponent.java:31)
    at org.apache.camel.impl.DefaultComponent.createEndpoint(DefaultComponent.java:122)
    at org.apache.camel.impl.DefaultCamelContext.getEndpoint(DefaultCamelContext.java:525)
    ... 17 more

以下是我对创建骆驼上下文和rabbitmq队列的类的实现。

RMQCamleSender.java

package test;

import org.apache.camel.CamelContext;
import org.apache.camel.Exchange;
import org.apache.camel.Processor;
import org.apache.camel.builder.RouteBuilder;
import org.apache.camel.impl.DefaultCamelContext;
import test.Producer;

public class RMQCamelSender {
    public static void main(String[] argv) throws Exception {

        Producer queueProd = new Producer();
        queueProd.setupConnection();
        System.out.println(queueProd.toString());

        CamelContext context = new DefaultCamelContext();

        context.addRoutes(new RouteBuilder() {

            public void configure() throws Exception {
                System.out.println("hello world");
                from("file://target/?fileName=doctor.txt&charset=utf-8")
                        .process(new Processor() {
                            public void process(Exchange msg) throws Exception {
                                System.out.println(msg.getIn().getBody(
                                        String.class));

                            }

                        }).to("rabbitmq://localhost:15672?queue=hello");

            }
        });

        context.start();
        Thread.sleep(4000);
        context.stop();
    }
}

Producer.java

package test;

import java.io.IOException;
import java.util.concurrent.TimeoutException;

import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.Channel;

public class Producer {
    public static final String QUEUE_NAME = "hello";
    public static Connection connection;

    public void setupConnection() throws IOException, TimeoutException {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost");

        connection = factory.newConnection();
        Channel channel = connection.createChannel();
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);
    }

}

如果我不使用 Camel 并尝试使用标准 rabbitmq 库联系队列，程序运行良好。

Answer 1

错误消息非常具有指示性：您在端点 URI 中缺少交换名称，它不能为空，因为 the official Camel documentation 表示它必须遵循给定的格式：

rabbitmq://hostname[:port]/exchangeName?[options]

我建议您尝试使用amq.direct 交换（更多详细信息，请参阅official RabbitMQ documentation），如下所示：

public void configure() throws Exception {
    System.out.println("hello world");
    from("file://target/?fileName=doctor.txt&charset=utf-8")
        .process(new Processor() {
            public void process(Exchange msg) throws Exception {
                System.out.println(msg.getIn().getBody(String.class));
            }

        }).to("rabbitmq://localhost:5672/amq.direct?routingKey=hello");
}

另外，你正在使用15672端口，这是默认情况下RabbitMQ的Web管理控制台的端口，我猜你没有更改默认设置，所以端口需要是5672。您还必须使用 routingKey 参数而不是 queue 参数，因为在 RabbitMQ 语义中，您发布到 exchange:routingKey 并且仅从队列中消费。

Answer 2

您需要包含 rabbitmq 交换名称。此处不见了；

.to("rabbitmq://localhost:15672?queue=hello");

应该是这样的：

.to("rabbitmq://localhost:15672/exchangeName?routingKey=hello");

另外，如果你要发送到交易所，为什么要指定一个队列？您只需要指定路由键，如果该路由键有绑定，交换将发送到该队列。