Springboot asynchronous message processing method

In work, we often encounter business scenarios that require asynchronous message processing. There are completely different processing methods depending on the nature of the message.

1. Messages are not independent

Independent messages usually have sequential dependencies. At this time, the message processing mechanism will degenerate into a linear queue processing mode, and only one consumer can go to a single thread. Process the message.

2. Messages are completely independent

Completely independent messages can be processed concurrently by multiple consumers (threads) at the same time, achieving maximum concurrent processing capabilities.

3. Messages are not completely independent

Usually this is the case where messages from the same source (from the same producer) are required to be ordered, and the order of messages from different sources has nothing to do with it.

Message processing in this scenario will be relatively complicated. In order to ensure the order of messages from the same source, it is easy to think of binding fixed consumer threads to messages from the same source. This is very simple but has big problems.

If the number of producers is large, the number of bound threads may not be enough. Of course, thread resources can be reused and the same thread can be bound to multiple message sources for processing. This will cause another problem: between message sources. interactions between.

Consider the following scenario:

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

Then the message processing of the producer P2 will be blocked by the large number of messages from P1, resulting in mutual influence between P1 and P2, and also Inability to fully utilize other consumer threads leads to imbalance.

So, we must consider avoiding such problems. Achieve the timeliness of consumption processing (as soon as possible), isolation (avoiding mutual interference), and balance (maximizing concurrent processing)

In implementation, there will be two modes, the easier to think of is thread dispatching Model (PUSH mode), the specific method is usually as follows:

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

2. According to the message source, dispatch it to the appropriate consumer thread for processing.

The distribution algorithm mechanism can be as simple as Hash based on the message source, or as complex as the current load of each consumer thread, the length of the waiting queue, and the complexity of the message, and can be selected for distribution based on a comprehensive analysis.

Simple Hash will definitely encounter the problems described in the above scenario, but complex distribution calculations are obviously very troublesome and complicated to implement, the efficiency is not necessarily good, and it is difficult to achieve a perfect balance in terms of balance.

The second mode uses the PULL method, and the thread pulls on demand. The specific method is as follows:

1. The message source directly puts the generated message into the temporary queue corresponding to the source (as follows) Each session shown represents a different message source), and then the session is placed in a blocking queue to notify the thread for processing

2. Multiple consumer threads poll the queue at the same time to compete for messages (guaranteing that only one thread takes the Go to

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

4. If it is not processed by other threads, then Indicate the status in the synchronization area setting processing, and process the messages in the temporary queue after exiting the synchronization area

5. After the processing is completed, finally enter the synchronization area setting processing again to indicate that the status is idle

The following is a piece of code to describe the consumption 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);
	}
}

The above is the detailed content of Springboot asynchronous message processing method. For more information, please follow other related articles on the PHP Chinese website!