java Apache camel、RabbitMQ 如何发送消息/对象

Apache camel,RabbitMQ how to send messages/objects

提问by cpu2007

I hope someone can provide some help on this matter.

我希望有人可以在这个问题上提供一些帮助。

I am using camel rabbitmq and for testing purpose I am trying to send a message to the queue, which I'm trying to display in rabbitmq interface and then also read it back.

我正在使用骆驼 rabbitmq 并且出于测试目的，我试图向队列发送一条消息，我试图在 rabbitmq 界面中显示该消息，然后还将其读回。

However I can't get this working.

但是我不能让它工作。

What I believe works is that I created, in the exchange tab of rabbitmq management interface, a new exchange. In my java code I send the message to that exchange. When the code is executed, I can see a spike in the web interface showing that something has been received but I can't see what has been received. When I try to read, I can't read and get the following errror: < in route: Route(route2)[[From[rabbitmq://192.168.59.103:5672/rt... because of Route route2 has no output processors. You need to add outputs to the route such as to("log:foo").

我认为有效的是我在 rabbitmq 管理界面的交换选项卡中创建了一个新的交换。在我的 Java 代码中，我将消息发送到该交易所。执行代码时，我可以在 Web 界面中看到一个尖峰，显示已收到某些内容，但我看不到已收到的内容。当我尝试阅读时，我无法阅读并得到以下错误： < in route: Route(route2)[[From[rabbitmq://192.168.59.103:5672/rt... 因为 Route route2 没有输出处理器。您需要向路由添加输出，例如 to("log:foo")。

Can someone provide me a practical example on how to send a message,see it in the web interace and also read it? any tutorial showing this process will be also appreciated.

有人可以为我提供一个关于如何发送消息的实际示例，在网络界面中查看并阅读它吗？任何展示此过程的教程也将不胜感激。

Thank you

谢谢

================= SECOND PART

================== 第二部分

The error I'm getting now is the following:

我现在得到的错误如下：

Caused by: com.rabbitmq.client.ShutdownSignalException: channel error; reason: {#method<channel.close>(reply-code=406, reply-text=PRECONDITION_FAILED - cannot redeclare exchange 'rhSearchExchange' in vhost '/' with different type, durable, internal or autodelete value, class-id=40, method-id=10), null, ""}
    at com.rabbitmq.utility.ValueOrException.getValue(ValueOrException.java:67)
    at com.rabbitmq.utility.BlockingValueOrException.uninterruptibleGetValue(BlockingValueOrException.java:33)
    at com.rabbitmq.client.impl.AMQChannel$BlockingRpcContinuation.getReply(AMQChannel.java:343)
    at com.rabbitmq.client.impl.AMQChannel.privateRpc(AMQChannel.java:216)
    at com.rabbitmq.client.impl.AMQChannel.exnWrappingRpc(AMQChannel.java:118)
    ... 47 more

I have the following settings:

我有以下设置：

I get this error, I believe I'm doing something wrong with the URI and I have to define some extra parameters that I'm missing My exchange is of direct type My queue is of durable type And my uri is : rabbitmq://192.168.59.105:5672/rhSearchExchange?username=guest&password=guest&routingKey=rhSearchQueue

我收到这个错误，我相信我的 URI 有问题，我必须定义一些我遗漏的额外参数我的交换是直接类型我的队列是持久类型我的 uri 是：rabbitmq:// 192.168.59.105:5672/rhSearchExchange?username=guest&password=guest&routingKey=rhSearchQueue

any input on this?

对此有何意见？

Thanks

谢谢

回答by Jeff

So I was able to figure this out yesterday, I had the same (or at least similar) problems you were having.

所以我昨天能够解决这个问题，我遇到了与您相同（或至少相似）的问题。

The options you have in the RabbitMQ URI must exactly match the options that your exchange was created with. For example, in my configuration, I had an exchange called tasksthat was a direct type, was durable, and was not configured to autodelete. Note that the default value for the autodelete option in the rabbitmq camel component is true. Additionally, I wanted to get the messages with the routing key camel. That means my rabbitmq URI needed to look like:

您在 RabbitMQ URI 中的选项必须与创建您的交换所使用的选项完全匹配。例如，在我的配置中，我有一个称为tasks直接类型的交换，是持久的，并且没有配置为自动删除。请注意，rabbitmq camel 组件中自动删除选项的默认值为true. 此外，我想获取带有路由键的消息camel。这意味着我的 rabbitmq URI 需要看起来像：

rabbitmq:localhost:5672/tasks?username=guest&password=guest&autoDelete=false&routingKey=camel

Additionally, I wanted to read from an existing queue, called task_queuerather than have the rabbitmq camel component declare it's own queue. Therefore, I also needed to add an additional query parameter, so my rabbitmq URI was

此外，我想从现有的队列中读取，调用task_queue而不是让 rabbitmq camel 组件声明它自己的队列。因此，我还需要添加一个额外的查询参数，所以我的rabbitmq URI 是

rabbitmq:localhost:5672/tasks?username=guest&password=guest&autoDelete=false&routingKey=camel&queue=task_queue

This configuration worked for me. Below, I added some Java code snippets from the code that configures the exchange and queue and sends a message, and my Camel Route configuration.

这个配置对我有用。下面，我从配置交换和队列以及发送消息的代码中添加了一些 Java 代码片段，以及我的 Camel Route 配置。

Exchange and Queue configuration:

交换和队列配置：

rabbitConnFactory = new ConnectionFactory();
rabbitConnFactory.setHost("localhost");
final Connection conn = rabbitConnFactory.newConnection();
final Channel channel = conn.createChannel();

// declare a direct, durable, non autodelete exchange named 'tasks'    
channel.exchangeDeclare("tasks", "direct", true); 
// declare a durable, non exclusive, non autodelete queue named 'task_queue'
channel.queueDeclare("task_queue", true, false, false, null); 
// bind 'task_queue' to the 'tasks' exchange with the routing key 'camel'
channel.queueBind("task_queue", "tasks", "camel"); 

Sending a message:

发送消息：

channel.basicPublish("tasks", "camel", MessageProperties.PERSISTENT_TEXT_PLAIN, "hello, world!".getBytes());

Camel Route:

骆驼路线：

@Override
public void configure() throws Exception {
    from("rabbitmq:localhost:5672/tasks?username=guest&password=guest&autoDelete=false&routingKey=camel&queue=task_queue")
        .to("mock:result");
}

I hope this helps!

我希望这有帮助！

回答by slf

Because this it the top hit on Google for rabbitmq/camel integration I feel the need to add a bit more to the subject. The lack of simplecamel examples is astonishing to me.

因为这是在 Google 上对 rabbitmq/camel 集成的热门话题，所以我觉得有必要为该主题添加更多内容。缺乏简单的骆驼例子让我感到惊讶。

import org.apache.camel.CamelContext;
import org.apache.camel.ConsumerTemplate;
import org.apache.camel.Endpoint;
import org.apache.camel.Exchange;
import org.apache.camel.ProducerTemplate;
import org.apache.camel.impl.DefaultCamelContext;
import org.junit.Test;

public class CamelTests {
    CamelContext context;
    ProducerTemplate producer;
    ConsumerTemplate consumer;
    Endpoint endpoint;

    @Test
    public void camelRabbitMq() throws Exception {
        context = new DefaultCamelContext();

        context.start();

        endpoint = context.getEndpoint("rabbitmq://192.168.56.11:5672/tasks?username=benchmark&password=benchmark&autoDelete=false&routingKey=camel&queue=task_queue");

        producer = context.createProducerTemplate();

        producer.setDefaultEndpoint(endpoint);
        producer.sendBody("one");
        producer.sendBody("two");
        producer.sendBody("three");
        producer.sendBody("four");
        producer.sendBody("done");

        consumer = context.createConsumerTemplate();
        String body = null;
        while (!"done".equals(body)) {
            Exchange receive = consumer.receive(endpoint);
            body = receive.getIn().getBody(String.class);
            System.out.println(body);
        }

        context.stop();

    }

}