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

This article mainly introduces the relevant knowledge of "the method of Springboot Asynchronous message processing". The editor shows you the operation process through the actual case. The operation method is simple, fast and practical. I hope this article "methods of Springboot Asynchronous message processing" can help you solve the problem.

In the work, we often encounter business scenarios that require asynchronous message processing, and there are completely different processing methods according to the nature of the message.

1. The message is not independent

Messages that are not independent usually have sequential dependencies, so the message processing mechanism will degenerate into a linear queue processing mode, and only one consumer can process the message in a single thread.

2. The message is completely independent.

Completely independent messages, which can be processed concurrently by multiple consumers (threads), can achieve maximum concurrent processing capacity.

3. The message is not completely independent.

Typically, homologous messages (from the same producer) are required to be ordered, regardless of the order of heterogeneous messages.

Message processing in this scenario is relatively complex, and in order to keep the messages of the same origin orderly, it is easy to think of a consumer thread that binds messages from the same source to a fixed thread, which is simple but problematic.

If the number of producers is large, the number of bound threads may not be enough, of course, you can reuse thread resources and bind multiple message sources to the same thread for processing, which will have another problem: the interaction between message sources.

Consider the following scenarios:

Producer P1 generates a large number of messages into the queue and is assigned to the consumer thread C1 for processing (C1 may take a long time to process), when producer P2 generates a message, which is unfortunately also assigned to the consumer thread C1 for processing.

Then the message processing of producer P2 will be blocked by a large number of messages of P1, resulting in the interaction between P1 and P2, and can not make full use of other consumer threads to cause imbalance.

Therefore, we must consider avoiding such a problem. Achieve the timeliness of consumption processing (as soon as possible), isolation (avoid mutual interference), balance (maximize concurrent processing)

In the implementation, there are two modes, and it is easier to think of the thread dispatch model (PUSH mode), which is usually as follows:

1. There is a global message dispatcher that polls the queue to retrieve the message.

two。 According to the source, dispatch to the appropriate consumer thread processing.

The algorithm mechanism of dispatch can be similar to Hash based on message source, and complex one can choose dispatch according to the current load of each consumer thread, the length of waiting queue and the complexity of messages.

Simple Hash will certainly encounter the problems described in the above scenario, but the complex dispatch calculation is obviously very troublesome and complex, the efficiency is not necessarily good, and it is difficult to achieve a good balance in terms of balance.

The second mode is PULL, which is pulled by threads on demand, as follows:

1. The message source puts the generated message directly into the temporary queue corresponding to the source (each session represents a different message source as shown below), and then places the session into a blocking queue to notify the thread processing.

two。 Multiple consumer threads poll the queue at the same time, scrambling for messages (ensure that only one thread gets the

3. Check whether the queue indicator is being processed by other threads (implementation requires detection synchronization based on homologous messages at the thread level)

4. If it is not processed by other threads, the status is indicated in the synchronization zone processing, and the messages in the temporary queue are processed after exiting the synchronization area.

5. After the processing is completed, the synchronous zone processing indicates that the status is idle again.

The following code describes the consumer thread processing process:

Public void run () {try {for (AbstractSession s = squeue.take (); s! = null; s = squeue.take ()) {/ / first check any worker is processing this session? / / if any other worker thread is processing this event with same session, just ignore it. Synchronized (s) {if (! s.isEventProcessing ()) {s.setEventProcessing (true);} else {continue }} / / fire events with same session fire (s) / / last reset processing flag and quit current thread processing s.setEventProcessing (false) / / if remaining events, so re-insert to session queue if (s.getEventQueue () .size () > 0 & &! s.isEventProcessing ()) {squeue.offer (s) } catch (InterruptedException e) {LOG.warn (e.getMessage (), e);}} this is the end of the introduction on "methods of Springboot Asynchronous message processing". Thank you for reading. If you want to know more about the industry, you can follow the industry information channel. The editor will update different knowledge points for you every day.

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.