Shulou(Shulou.com)06/03 Report--

With the increasing popularization of the application of message queue, the use of message queue in distributed systems can greatly reduce the degree of coupling between components, thus improving the processing efficiency of components. Because of the existence of message queue, we can process the task asynchronously, which can reduce the request response time and decoupling. At the same time, due to the use of message queues, as long as the message format remains unchanged, the sender and receiver of the message do not need to contact each other, nor need to be affected by each other, that is, decoupling and.

The so-called decoupling means that system A generates a piece of data and sends it to MQ. Which system needs data to be consumed in MQ. If the new system needs data, you can consume it directly from MQ; if a system no longer needs this data, you can cancel the consumption of MQ messages. In this way, system A does not need to think about who to send data to, maintain the code, or consider whether the call is successful, failure timeout, and so on.

The so-called asynchronous, then system A continuously sends 3 messages to the MQ queue. If it takes 5ms for system A to accept a request and return the response to the user, the total time is 3 + 5 = 8ms. For the user, it actually feels like a click of a button, and 8ms will return directly later. Cool! The website is so good, so fast!

In addition, the use of message queues has the advantage of peaking. The so-called peak clipping means that at certain times, users will make a large number of requests to our service, and our database sometimes needs to write to these requests. However, the throughput of mysql reaches a peak of 5000, and the rest will have to wait slowly. And when the concurrency is too high, all kinds of exceptions in the database will also drive people crazy, but, uh, we use message queues. All the user's requests are crammed into the message queue, and then the processing result is returned by the message queue, while the requests are stored in the queue and consumed sequentially one by one, so that there is no peak and trough in request writing.

Based on these points, our development team recently carried out the access transformation of the message queue due to the needs of the project in the development process of spring boot.

I encountered such a problem in the process of transformation. At first, I wrote the comments on class. An exception will occur during the operation. Here are the details of the exception:

2019-05-31 17 s.a.r.l.ConditionalRejectingErrorHandler 42 WARN 36.798 30544-[cTaskExecutor-1] s.a.r.l.ConditionalRejectingErrorHandler: Execution of Rabbit message listener failed.

Org.springframework.amqp.rabbit.listener.exception.ListenerExecutionFailedException: Listener threw exception

At org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.wrapToListenerExecutionFailedExceptionIfNeeded (AbstractMessageListenerContainer.java:1506)

At org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.doInvokeListener (AbstractMessageListenerContainer.java:1417)

At org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.actualInvokeListener (AbstractMessageListenerContainer.java:1337)

At org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.invokeListener (AbstractMessageListenerContainer.java:1324)

At org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.executeListener (AbstractMessageListenerContainer.java:1303)

At org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer.doReceiveAndExecute (SimpleMessageListenerContainer.java:817)

At org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer.receiveAndExecute (SimpleMessageListenerContainer.java:801)

At org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer.access$700 (SimpleMessageListenerContainer.java:77)

At org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer$AsyncMessageProcessingConsumer.run (SimpleMessageListenerContainer.java:1042)

At java.lang.Thread.run (Thread.java:748)

Caused by: org.springframework.amqp.AmqpException: No method found for class java.util.LinkedHashMap

At org.springframework.amqp.rabbit.listener.adapter.DelegatingInvocableHandler.getHandlerForPayload (DelegatingInvocableHandler.java:147)

At org.springframework.amqp.rabbit.listener.adapter.DelegatingInvocableHandler.getMethodNameFor (DelegatingInvocableHandler.java:250)

At org.springframework.amqp.rabbit.listener.adapter.HandlerAdapter.getMethodAsString (HandlerAdapter.java:70)

At org.springframework.amqp.rabbit.listener.adapter.MessagingMessageListenerAdapter.invokeHandler (MessagingMessageListenerAdapter.java:190)

At org.springframework.amqp.rabbit.listener.adapter.MessagingMessageListenerAdapter.onMessage (MessagingMessageListenerAdapter.java:120)

At org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.doInvokeListener (AbstractMessageListenerContainer.java:1414)

... 8 common frames omitted

As for the source code of the development, this is what I wrote. I annotate here in class. At this time, I guess that the annotations are not in the right position.

@ Component@RabbitListener (queues = "xx.yy.zz") public class Receiver {@ RabbitHandler public void process (MSGSTO message) {System.out.println ("consumption message"); System.out.println (message.toString ());}}

In fact, it is the wrong location, but a more professional solution is that this listener annotation is at the method level and cannot be used on class. We might as well take a look at the source code of RabbitListener to fundamentally understand the use of this method.

Package org.springframework.amqp.rabbit.annotation;import java.lang.annotation.Documented;import java.lang.annotation.ElementType;import java.lang.annotation.Repeatable;import java.lang.annotation.Retention;import java.lang.annotation.RetentionPolicy;import java.lang.annotation.Target;import org.springframework.messaging.handler.annotation.MessageMapping;@Target ({ElementType.TYPE, ElementType.METHOD, ElementType.ANNOTATION_TYPE}) @ Retention (RetentionPolicy.RUNTIME) @ MessageMapping@Documented@Repeatable (RabbitListeners.class) public @ interface RabbitListener {String id () default "" String containerFactory () default ""; String [] queues () default {}; Queue [] queuesToDeclare () default {}; boolean exclusive () default false; String priority () default ""; String admin () default ""; QueueBinding [] bindings () default {}; String group () default ""; String returnExceptions () default "; String errorHandler () default"; String concurrency () default"; String autoStartup () default";}

Due to business needs, we do need to process messages asynchronously, and the easiest way to receive messages asynchronously is to use an annotated listening endpoint infrastructure. In short, it allows methods that host bean to be exposed as endpoints of Rabbit listener. Here, when using the queues attribute, you can specify that the associated container can listen on multiple queues. You can use the @ Header annotation to create the name of the queue where the POJO method can receive messages. Br/ > here, when using the queues attribute, you can specify that the associated container can listen on multiple queues. You can use the @ Header annotation to create the name of the queue where the POJO method can receive messages.

@ Componentpublic class Receiver {@ RabbitListener (queues = "xx.yy.zz") @ RabbitHandler public void process (MSGSTO message) {System.out.println ("consumption message"); System.out.println (message.toString ())

As for the configuration method, I do the access configuration in the form of application.yml, for example

Rabbitmq: addresses: 127.0.0.1 port: 5672 username: guest password: guest publisher-confirms: true publisher-returns: true virtual-host: dev listener: simple: concurrency: 10 max-concurrency: 20

These properties are injected into the RabbitProperties property, as shown in

@ ConfigurationProperties (prefix = "spring.rabbitmq") public class RabbitProperties {… }

It's interesting, right:)

Reference:

Why use message queuing, https://www.javazhiyin.com/22897.html

Welcome to subscribe "Shulou Technology Information " to get latest news, interesting things and hot topics in the IT industry, and controls the hottest and latest Internet news, technology news and IT industry trends.