Reception of messages in RabbitMQ, Java

Question

In rabbitmq tutorials there is an example of a code describing the reception of messages. part of the code is not entirely clear:

Consumer consumer = new DefaultConsumer(channel) { @Override public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException { String message = new String(body, "UTF-8"); System.out.println(" [x] Received '" + message + "'"); } };

not entirely clear what is in brackets. we seem to have created a consumer object, and here the body goes in curly braces ... I don’t understand what the body is. Is this the implementation of the constructor?
Is it possible to receive messages somehow easier?

AdamSkywalker AdamSkywalker 399 one eight · Accepted Answer · 2016-03-09T13:40:52

About anonymous classes have already explained everything, but more about admission:

Usually in a rabbitmq listener, queues are started as a separate thread so that it hangs on this queue and processes everything that comes in:

Initialization:

 Connection connection = connectionFactory.newConnection(); Channel channel = connection.createChannel(); channel.queueDeclare(QUEUE_NAME, true, false, false, null); // объявили очередь QueueingConsumer queueingConsumer = new QueueingConsumer(channel); // создали слушателя channel.basicConsume(QUEUE_NAME, false, queueingConsumer); // запустили

Run method in Runnable

 @Override public void run() { try { while (true) { QueueingConsumer.Delivery delivery = queueingConsumer.nextDelivery(); if (delivery == null) { Thread.sleep(100); // сообщений нет - ждем } else { handleDelivery(delivery); // обработка } } } catch (IOException | InterruptedException e) { e.printStackTrace(); // цикл прервется если соединение разорвется - это на усмотрение уже } }

Processing received message:

 protected void handleDelivery(QueueingConsumer.Delivery delivery) throws IOException { try { processMsg(delivery.getBody()); // бизнес лоигка channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false); // подтверждение приема } catch (Exception e) { channel.basicReject(delivery.getEnvelope().getDeliveryTag(), true); // возвращение в очередь - на усмотрение } }

it turns out that you are creating a connection, then in the body of this anonymous class Consumer consumer = new DefaultConsumer (channel), on the basis of which the object is created, do you start the stream?
Does this one then work with a delay, "probing the channel" every 100 milliseconds?
If there is a message, it starts the cycle and processes it? I understand it correctly?
@Iga everything is so, except that there is no DefaultConsumer, in this example, consumer - QueueingConsumer
A few more questions. What is the method that you have commented on as “business logic?”? His idea is highlighted in red .. and more, tell me, where do you get information on methods and classes of rabbitmq? On their site, it is described as a little everything ... ((
Why did you use QueueingConsumer instead of DefaultConsumer?
@Iga 1. business logic is the processing of received bytes, implement it as you need in your program.
2. in java doc on classes it is described how they are best used 3. QueueingConsumer works in a separate stream, and DefaultConsumer in the same stream as the channel, as a result, the first does not block Rabbit message processing, and the second blocks

Answer 2 · 2016-03-09T12:32:07

Such a construct is called an anonymous class. This is a simultaneous declaration of the inheritor of the class DefaultConsumer , its implementation and a call to the constructor.

When you create a new object, you write:

 new Foo(<parameters>);

In the case of an anonymous class, you do this:

 new Foo() { // Тело нового класса }

In fact, you create a descendant of the class Foo and at the same time call its constructor, additionally implementing / redefining the necessary methods in the body. That is, such a one-time class heir.

You can do this in a separate file:

 class MyConsumer extends DefaultConsumer { @Override public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException { String message = new String(body, "UTF-8"); System.out.println(" [x] Received '" + message + "'"); } }

And then:

 channel.basicConsume(QUEUE_NAME, true, new MyConsumer());

it would be similar in meaning, but in the future you can still create MyConsumer objects, while using an anonymous class, you described it, created an object, and forgot about its description.

Why is it, specifically in your example? The consumer object in this case sets the logic that should work for different actions with messages. If you look at the DefaultConsumer source code , you will see that the handleDelivery method handleDelivery empty. Based on its name, it is a handler that is called if the message is delivered. In your example, for clarity, this method is redefined in a one-time inheritance class, so that the information that the message has been delivered and the message itself has been output to the console. In general, such constructions are useful, because you can make a flexible code. Today you only need to output information to the console about the delivery of the message, and tomorrow you will need to forward the message to someone, or send to grandma only after it was sent to the grandfather. Then you write some ForwardingConsumer or GrandmaConsumer , which will be the corresponding logic.

and I think what it is ... that is, using an anonymous class, we can not multiply the classes separately, but create an anonymous one and at the right time redefine its internal methods?
If you need to override / add 1-2 short methods and use such an object once, then use anonymous classes.
Anonymous, precisely because it is a class without a name, disposable.
@Iga, given that an anonymous class is also an object, it lives as long as there are links to it, like any other object.

Reception of messages in RabbitMQ, Java

2 answers 2

More articles: