Interviewers often ask: how to set up a "message queue" and "delayed message queue"?

background

There are many scenarios for using message queues , The most common usage scenarios are as follows .

1. Commodity seckill

such as , When we were doing second kill , There will be explosive user requests in a short time , If relevant measures are not taken , It will cause the server to be busy , The problem of response timeout , Light will lead to false death of the service , More importantly, the server will be shut down directly , The experience brought to users is also very bad . If the message queue is added at this time , After the server receives all the user's requests , First write all these requests to the message queue, and then queue them for processing , This will not result in multiple requests being processed at the same time ; If the message queue length exceeds the maximum number that can be carried , Then we can discard the current user's request , Notify the foreground user “ Page error , Please refresh again ” Etc , This will have a better interactive experience .

2. system decoupling

After using the message queue , We can modularize the business functions of the system , Realize the decoupling of the system . for example , Before using message queuing , After the current user has improved his personal information , First, we need to update the user's information , Add another user information modification log . But suddenly one day, the product manager put forward a demand , After the foreground user information is updated , You need to give this user some bonus points , Then a few days later, the product manager raised another demand , After the foreground user information is updated , Not only to increase points and rewards , Also increase the experience value of users , But a few days later, the needs of the product manager changed again , He asked to improve the data without increasing users' points , This is repeated over and over again 、 Back and forth , I think the students of R & D can't stand it , But this is the norm for Internet companies , Do we have a way to do it once and for all ？

you 're right , At this time, we thought of using message queue to decouple the system , The realization of each function is independent , Just need a switch to subscribe or unsubscribe , When you need to add functions , Just open the subscription “ Perfect user information ” Just a queue , If you don't need it in two days , Just turn off the subscription switch again , So we don't have to change the business code back and forth , It is easy to realize the decoupling between system modules .

3. logging

In fact, most of our logging behaviors are not directly related to the main business operated by foreground users , Only our operators and operators need to get the log information of these user operations , To conduct user behavior analysis or behavior monitoring . Before we use message queuing , The general approach is when there is a user request , First process the user's request and then log , These two operations are put together , The foreground users also need to wait for the log to be added before they can get the response information from the background , This actually wastes part of the time of the front desk users . At this point we can use message queuing , After responding to the user's request , Just put the operation information into the message queue , You can directly return the result to the foreground user , Wait out of order for log processing and log addition to complete , Thus, the waiting time of foreground users is shortened .

We can go through  JDK  Provided  Queue  To implement a custom message queue , Use  DelayQueue  Implement delayed message queuing .

The key analysis

The investigation of message queue focuses more on the core idea of message queue , Because only when you understand what message queuing is ？ And when to use message queuing ？ To solve the problems we encounter in our daily work , The specific implementation of message queue , You only need to master the use of a message oriented middleware , Because the core implementation idea of message queue middleware is the same , Not only that , The use of Message Queuing Middleware is similar , As long as you master one, you can make good use of other messaging middleware by analogy .

The following two questions are common ：

• A message that you are familiar with ？

• How to implement message queuing manually ？

Knowledge expansion

1. Common message oriented middleware  RabbitMQ

At present, it is commonly used in the market  MQ（Message Queue, Message queue ） Middleware has  RabbitMQ、Kafka、RocketMQ, If it is a lightweight message queue, you can use  Redis  Message queue provided , Let's introduce this article first  RabbitMQ.

RabbitMQ  Is an old open source message middleware , It implements the standard  AMQP（Advanced Message Queuing Protocol, Advanced message queue protocol ） Message middleware , Use  Erlang  Language development , Support cluster deployment , Mixed calls with multiple client languages , It supports the following mainstream development languages ：

• Java and Spring

• .NET

• Ruby

• Python

• PHP

• JavaScript and Node

• Objective-C and Swift

• Rust

• Scala

• Go

RabbitMQ  There is  3  Important concepts ： producer 、 Consumers and agents .

• producer ： The creator of the message , Responsible for creating and pushing data to the message server .

• consumer ： The receiver of the message , Used to process data and confirm messages .

• agent ： That is to say  RabbitMQ

Service itself , Used to play it “ Courier ” Role , Because it doesn't produce news itself , Just playing “ Courier ” Role , Temporarily store and deliver messages .

Their operation process , As shown in the figure below ：

RabbitMQ  It has the following advantages ：

• Support persistence ,RabbitMQ  Support disk persistence function , It ensures that messages are not lost ;

• High concurrency ,RabbitMQ  Used  Erlang  development language ,Erlang  It's a language developed for telephone switches , It is naturally equipped with high and luminous ring and high availability features ;

• Support distributed cluster , Precisely because  Erlang  The realization of language , therefore  RabbitMQ  Cluster deployment is also very simple , Just start each node and use  --link Add the node to the cluster , also  RabbitMQ  Support automatic primary selection and automatic disaster recovery ;

• Support for multiple languages , such as  Java、.NET、PHP、Python、JavaScript、Ruby、Go  etc. ;

• Support message confirmation , Support message consumption confirmation （ack） It ensures that each message can be consumed normally ;

• It supports many plug-ins , For example, web console message management plug-in 、 Message delay plug-ins, etc ,RabbitMQ  There are many plug-ins and they are very convenient to use .

RabbitMQ  The message type of , There are four kinds of ：

• direct（ Default type ） Pattern , This mode is one-to-one , That is, a message will only be sent to one consumer ;

• headers  Pattern , Allows you to match messages  header  Instead of routing keys （RoutingKey）, besides  headers  and  direct

• The use of is exactly the same , But because  headers  The matching performance is very poor , Hardly used ;

• fanout  Pattern , For multicast , A message will be distributed to all subscribers ;

• topic Pattern , Subscribe to the theme mode , Wildcards are allowed （#、*） Match one or more messages , I can use “cn.mq.#” Matching to multiple prefixes is “cn.mq.xxx” The news of , For example, you can match “cn.mq.rabbit”、“cn.mq.kafka” Wait for news .

2. Custom message queue

We can use  Queue  To implement message queuing ,Queue  It can be roughly divided into the following three categories ：

• deque （Deque） yes  Queue  So are subclasses of  Queue  Supplementary category of , Both the head and tail support element insertion and acquisition ;

• Blocking queue refers to the operation of elements （ Add or remove ）, If it doesn't work , Will block waiting for execution , For example, when adding elements , If the queue element is full , The queue will block and wait until there is an empty space before inserting ;

• Non blocking queues , As opposed to blocking queues , It will directly return the result of the operation , Instead of blocking wait operations , Double ended queues are also non blocking queues .

The implementation code of custom message queue is as follows ：

import java.util.LinkedList;import java.util.Queue;public class CustomQueue { //  Define message queues  private static Queue<String> queue = new LinkedList<>(); public static void main(String[] args) { producer(); //  Call producer  consumer(); //  Call consumer  } //  producer  public static void producer() { //  Add message  queue.add(&quot;first message.&quot;); queue.add(&quot;second message.&quot;); queue.add(&quot;third message.&quot;); } //  consumer  public static void consumer() { while (!queue.isEmpty()) { //  News consumption  System.out.println(queue.poll()); } }}

The execution result of the above procedure is ：

first message.second message.third message.

It can be seen that messages are consumed in first in first out order .

To implement a custom delay queue, you need to implement  Delayed  Interface , rewrite  getDelay()  Method , The complete implementation code of delay queue is as follows ：

import lombok.Getter;import lombok.Setter;import java.text.DateFormat;import java.util.Date;import java.util.concurrent.DelayQueue;import java.util.concurrent.Delayed;import java.util.concurrent.TimeUnit;/** *  Custom delay queue  */public class CustomDelayQueue { //  Delayed message queuing  private static DelayQueue delayQueue = new DelayQueue(); public static void main(String[] args) throws InterruptedException { producer(); //  Call producer  consumer(); //  Call consumer  } //  producer  public static void producer() { //  Add message  delayQueue.put(new MyDelay(1000, &quot; news 1&quot;)); delayQueue.put(new MyDelay(3000, &quot; news 2&quot;)); } //  consumer  public static void consumer() throws InterruptedException { System.out.println(&quot; Start execution time ：&quot; + DateFormat.getDateTimeInstance().format(new Date())); while (!delayQueue.isEmpty()) { System.out.println(delayQueue.take()); } System.out.println(&quot; End execution time ：&quot; + DateFormat.getDateTimeInstance().format(new Date())); } /** *  Custom delay queue  */ static class MyDelay implements Delayed { //  Delay deadline （ Company ： millisecond ） long delayTime = System.currentTimeMillis(); //  With the help of  lombok  Realization  @Getter @Setter private String msg; /** *  initialization  * @param delayTime  Set the delay execution time  * @param msg  Messages executed  */ public MyDelay(long delayTime, String msg) { this.delayTime = (this.delayTime + delayTime); this.msg = msg; } //  Get the time left  @Override public long getDelay(TimeUnit unit) { return unit.convert(delayTime - System.currentTimeMillis(), TimeUnit.MILLISECONDS); } //  The elements in the queue are sorted by  @Override public int compareTo(Delayed o) { if (this.getDelay(TimeUnit.MILLISECONDS) > o.getDelay(TimeUnit.MILLISECONDS)) { return 1; } else if (this.getDelay(TimeUnit.MILLISECONDS) < o.getDelay(TimeUnit.MILLISECONDS)) { return -1; } else { return 0; } } @Override public String toString() { return this.msg; } }}

The execution result of the above procedure is ：

Start execution time ：2020-4-2 16:17:28 news 1 news 2 End execution time ：2020-4-2 16:17:31

It can be seen that , news  1 news  2  All realize the function of delaying execution .

summary

This article describes the usage scenario of message queue ： Commodity seckill 、 System decoupling and logging , We also introduced  RabbitMQ  And its message type and its characteristics , It also uses  Queue  Subclasses of  LinkedList  A custom message queue is implemented , Use  DelayQueue  Implemented a custom delayed message queue .