Why didn’t I know RocketMQ producer had so many uses?

Small knowledge, big challenge! This paper is participating in theEssentials for programmers”Creative activities

preface

RocketMQ version of message queue is a distributed message middleware with low latency, high concurrency, high availability and high reliability built by Ali Cloud based on Apache RocketMQ.

Those of you who have read my previous articles probably have an idea of what a message queue is, but where does this message come from?

The so-called water of the Yellow River comes from heaven, and everything in nature does not come for nothing, right? 🐒🐒 how come, 🐒🐒 its mother produced; Chanel 💄💄 how to, machine and raw material production; Even the rice we eat at ordinary times has its origin; How did we get here? We are of course the great mother who gave birth

By the way, thank your great mother, and remember to call her on Sunday

Now let’s move on to the theme. This is the dividing line

Message queue RocketMQ not only provides asynchronous decoupling and peak-filling capabilities for distributed applications, but also features massive message stacking, high throughput, and reliable retry for Internet applications. Some of the features are listed below

• ** Message query: ** Message queue RocketMQ provides three ways to query messages by Message ID, Message Key, and Topic

• ** Query message track: ** Through message track, it can clearly locate the complete link that the message is sent from the producer and delivered to the message consumer through the RocketMQ version server of message queue, which is convenient to locate and troubleshoot problems

• ** Cluster consumption and broadcast consumption: ** When using the cluster consumption mode, RocketMQ version of message queue assumes that any message needs to be processed by any consumer in the consumer cluster; When using broadcast consumption mode, message queue RocketMQ pushes each message to all registered consumers in the consumer cluster, ensuring that the message is consumed at least once on each machine

• Reset consumption point: Resets consumption progress based on time or site, allowing users to backtrack or discard accumulated messages

• ** Dead-letter queue: ** Stores messages that cannot be consumed normally in a special dead-letter queue for subsequent processing

• ** Global message routing: ** Used for message synchronization between different regions of the world to ensure data consistency between regions

Client, in fact, it is easy to understand, we can put the RocketMQ understanding into a message service, since this is a service, we need to call the service, then invoke the service, where the news comes from, this is to be set according to the business scenario, so, news producers Producer belongs to a client; When messages are generated, they can’t stay there forever. Someone has to dispose of them. This is a business decision, so the consumer of messages is also a client.

Below ah, big fish will take you to see the use of this client

Producers Producer

As the name implies, producers are responsible for producing messages. There should be a lot of questions in your mind. For example, where does a Producer send messages, what is the flow, and what types of messages he sends

Fish fish teach you a small skill, learn a thing, first understand the general process, and then split and attack it, and finally overall understanding, so the effect will be very good, the exclusive secret recipe

Next, I will introduce Producer from three aspects: how the messages are sent (load balancing, fault tolerance mechanism), who the messages are sent to, where the messages are stored, and the types of messages

1. How is the message sent?

First of all, the message can’t be generated anywhere, there’s no point in generating the message, so the message has to be sent somewhere, relayed. See the picture below

The Producer first retrieves the Topic from the local cache, and if the Producer finds the Topic, he directly sends the generated message based on the Topic. Caching is known to optimize speed and reduce network traffic.

** Off topic: you can also go to see the Redis cache article written by this fish, love you, blah blah **

If not, go to NameServer to get the latest Topic list (which is registered to NameServer when the Broker is started), select a MessageQueue queue, and get the address of the Broker, again from the local cache. If not, request NameServer (which also registers the mapping between Broker address and Topic) to send the message

If sending fails, a retry mechanism is implemented. The default value is three retries

In fact, saving so much can not only reduce the network transmission between NameServer and NameServer, but also reduce the pressure of NameServer. NameServer itself is a lightweight design, which also helps to reduce the pressure of NameServer. I will also write a separate article to introduce NameServer

Load balancing

As we know, a corresponding Topic will be selected first when messages are sent, and each Topic will correspond to multiple MessageQueue. As a result, there is a problem. If the message is not evenly distributed, some queues may be too many and some queues are too few, which will cause a waste of resources

RocketMQ takes a frugal approach — yes, polling — and high-end ingredients often require only the most frugal cooking

Producers poll all MessageQueue under a Topic to achieve load balancing of the sender, which simply means that everyone has a share, as shown in the figure below:

In this way, messages from a Topic can be spread across multiple MessageQueue and, in turn, across multiple brokers.

Error tolerance mechanism for sending messages:

As the party sending messages, Producer has three fault-tolerant mechanisms:

• ** Local cache: ** Caches information retrieved from NameSever locally in case NameSever goes down

• ** Unavailable Broker collection: **Producer has a fault tolerance mechanism for brokers. The sendLatencyFaultEnable switch can be turned on. RocketMq maintains a HashMap of failed brokers internally, putting brokers of a certain latency level into this map. The next time you select a Broker, you will avoid unavailable brokers.

• ** Retries: ** When Producer sends messages, there is a retry mechanism, with three retries by default. Dead letter queue Consumer The number of Consumer retries exceeds the specified number

In this way, messages from a Topic can be spread across multiple MessageQueue and, in turn, across multiple brokers.

2. Who is the message sent to and where is it stored?

Producer connection NameSever

Producer uses NameSever to obtain the routing information of the specific Topic’s brokers and stores a cache of data locally, such as which MessageQueue a Topic has and which brokers the MessageQueue is on. Broker IP. Port, etc. The Producer sends messages only to the Master Broker, and the Slave obtains data through Master/Slave synchronization.

So how does Produce connect to NameSever

• ** Connection: ** A single producer keeps a long connection with a Nameserver, periodically queries topic configuration information, and if the Nameserver fails, the producer will automatically connect to the next Nameserver until a connection is available and can automatically reconnect.

• ** Polling time: ** By default, producers get the latest queue status for all topics from Nameserver every 30 seconds, which means that if a broker goes down, it takes up to 30 seconds for producers to sense it, during which time messages to that broker fail to be sent. This time is determined by the pollNameServerInteval parameter of DefaultMQProducer and can be manually configured.

• ** Heartbeat: ** has no heartbeat with Nameserver

**Producer connects brokers **

• ** Connections: ** Producers maintain long connections to all brokers involved in a Topic.

• ** Heartbeat: ** By default, producers send heartbeats to all brokers every 30 seconds. The broker scans all surviving connections every 10 seconds. If a connection has not received heartbeat data within 2 minutes (the difference between the current time and the last update time is more than 2 minutes, which cannot be changed), it closes the connection

After the Producer connects to the Broker, messages are sent to the Broker through polling and stored in the CommitLog of the Broker, which stores the original messages and a ConsumeQueue, which stores the location of messages delivered to a queue. Of course, message queues are persisted to disk without affecting memory, and of course messages are cleaned up periodically.

So where does the message go? When do you clean up the physical message files? And what are the benefits of this design?

We will leave all of this in the next article, Broker, to give you an insight into how the Broker brain helps RocketMQ support such high throughput

Anyway, this is a question worth digging into, and if you can explain not only what RocketMQ does, but how it stores and addresses, then the interviewer will love you. Then you have the ability to solve all kinds of practical problems, such as how to handle duplicate messages, how to ensure that messages are sequential, how to ensure that distributed transactions in a distributed system

How much money do you expect to work for us?

3. Types of messages

RocketMQ messages can be roughly divided into four types: normal messages, timed and delayed messages, sequential messages, and transactional messages ****, and that’s the point!

Let me briefly introduce the four types

• Normal messages: Non-featured messages in the RocketMQ version of the message queue, distinguished from featured timed and delayed messages, sequential messages, and transaction messages.

• Timed and delayed messages: Allows message producers to deliver timed (delayed) messages for up to 40 days.

• Sequential messaging: Allows message consumers to consume messages in the order in which they were sent.

• Transaction messages: Implement distributed transaction functionality similar to X or Open XA to achieve the final consistency state of the transaction.

Message queue RocketMQ provides four message types corresponding to the Topic can not be mixed. For example, the Topic created for ordinary messages can only be used to send and receive ordinary messages, not other types of messages. Similarly, transaction message topics can only send and receive transaction messages, not other types of messages, and so on

Ordinary message

Normal messages: Messages in message queue RocketMQ that are featureless, distinguished from featureless timed and delayed messages, sequential messages, and transaction messages

Common messages can be sent in Sync, Async, and Oneway

Synchronization is when we send a message and then we have to wait for the server to respond before we send the next one; Asynchrony applies to time-sensitive service scenarios. Asynchrony can continuously send messages without waiting for a response from the server. One-way takes less time than asynchronous, typically at the microsecond level, but reliability is reduced because you just send without waiting for the server to respond and no callback function is triggered

The synchronous

Synchronization: A communication mode in which a sender sends a message and sends the next message after receiving a response from the server

Asynchronous send

Asynchronous sending refers to the communication mode in which the sender sends a message and then sends the next message without waiting for the server to return a response

Message queue RocketMQ version of the asynchronous send, need to implement asynchronous SendCallback interface (SendCallback). After sending one message, the message sender can send a second message without waiting for a response from the server. The sender receives the server response through the callback interface and processes the response result

This parameter is applicable to time-sensitive service scenarios

One way to send

The sender only sends the message without waiting for a response from the server and no callback function is triggered, that is, only sends the request without waiting for a reply. The process of sending messages in this way is very short, usually in the microsecond level

Apply to scenarios that do not require high reliability, such as log collection

Timing and delay messages

Timed and delayed messages: Allows message producers to deliver timed (delayed) messages for up to 40 days

Delay message is used to specify the message is sent to a message queue RocketMQ version of the service side, after a delay time to be delivered to the client for consumption (3 seconds before consumption, for example), is applicable to solve some messages have time window for production and consumption of scene, or through the message trigger delay task scenario, similar to the delays in the queue.

Scheduled messages can be consumed by consumers after a specified timestamp, which is applicable to scenarios where there is a time window for message production and consumption, or when scheduled tasks are triggered by messages.

** Application scenarios **

Use messages to trigger timed tasks, when this timed message is useful, such as a reminder message sent to the user at a certain time; Some messages have a time window between their production and consumption, such as a typical e-commerce scenario where an order is closed without payment due to a timeout. This is where a delayed message comes in, and the order is closed without payment due due to a timeout

The accuracy of timing message will have 1s~2s delay error

In fact, there are some differences between timing message and delay message in use, used should all know, to mention, timing message needs to specify a certain point in time after the message sending time point as the message delivery time point; Delay message you may need to set a delay time length, length is fixed, but the time point is not fixed, is based on the timing of the message, the message will be sent from the current time point after fixed time delay to delivery, this we should be very clear, taobao under a single, give you 30 minutes time to pay, unpaid overtime, shut down the order

The order message

Sequential messaging: Allows message consumers to send messages in the order in which they were sent

Sequential messages fall into two categories:

• ** Global order: ** For a given Topic, all messages are published and consumed In a strict First In First Out (FIFO) order.

• ** Partitioning order: ** For a given Topic, all messages are partitioned according to a Sharding Key. Messages within the same partition are published and consumed in a strict FIFO order. Sharding Key is a Key field used to distinguish different partitions in sequential messages, which is completely different from the Key of ordinary messages.

In fact, this is a more classic question, the interview is also more often asked, is how to ensure the sequence? Fish fish will answer anyway, can you?

Will not come out to be beaten ~~

If you encounter this problem, you first need to divide the case description into global order and partition order:

1. Global order applies to scenarios where the performance requirements are not high and all messages are published and consumed in a first-in, first-out order. I’ve never seen this before, and I don’t usually use global order

2, partition order is suitable for high performance requirements, Sharding Key as a partition field, in a block in strict accordance with the order of first-in, first-out release and consumption. For example, the verification code of the user when registering and the user ID is used as Sharding Key, the message sent by the same user will be consumed in accordance with the order of release. Another example is the order process in e-commerce

The internal e-commerce systems of Alibaba Group all use the partition order message, which not only ensures the business order, but also ensures the high performance of business. Don’t ask me how I know, ali cloud official website

Sequential message FAQ

Why is global sequential message performance mediocre?

Global sequential messages follow the MESSAGE blocking principle of FIFO strictly, that is, if the previous message was not successfully consumed, the next message will be stored in the Topic queue. If you want to improve TPS for global sequential messages, you can upgrade the instance configuration while minimizing the time spent processing local business logic with the messaging client application.

Which message sending modes are supported for sequential messages?Whether cluster consumption and broadcast consumption are supported?

Sequential messages can only be sent in synchronous mode. Asynchronous mode is not supported. Otherwise, the sequence cannot be strictly guaranteed. For the time being, sequential messages only support cluster consumption mode, not broadcast consumption mode.

Transaction message

Transaction messages: Implement distributed transaction functionality similar to X or Open XA to achieve ultimate consistency

RocketMQ provides distributed transaction functions similar to X or Open XA. The final consistency of distributed transactions can be achieved through the transaction messages of RocketMQ.

** Semi-transactional message: ** Temporarily undeliverable message. The sender has successfully sent the message to the RocketMQ version of the message queue, but the server has not received a second acknowledgement of the message from the producer. At this point, the message is marked as “temporarily undeliverable”.

** Message back check: ** The secondary confirmation of a transaction message is lost due to intermittent network disconnection or producer application restart. When the RocketMQ server of the message queue finds that a message has been in the “semi-transaction message” for a long time through scanning, it needs to ask the message producer for the final status of the message (Commit or Rollback). The query process is called message back lookup.

Follow the fairy to see the transaction message sending steps:

1. The sender sends a semi-transaction message to the server Broker, which persists the message and returns an ACK to confirm that the message has been successfully sent

2. The sender starts executing the logic of the local transaction

3. According to the execution result of the local transaction, the sender will submit a second confirmation to the server to decide whether to Commit or Rollback. After receiving the Commit, the server will mark the message as deliverable and send it to the consumer. After the server receives Rollback, the semi-transaction message is deleted. If the server does not send the semi-transaction message, the consumer does not receive the semi-transaction message

For example, if the network is disconnected or the application is restarted, the secondary confirmation information of the above steps cannot reach the server. What should I do?

Here actually has a back to check mechanism, the sender sends the message, need local executing a transaction, if the transaction execution process has the stuck situation, or transaction execution results because the network problems, can’t transfer the transaction result to the server, the server will perform a check mechanism, back to confirm the final submit half a transaction message

conclusion

The consumer and producer client objects for the RocketMQ version of the message queue are thread-safe and can be shared across multiple threads. Multiple producer and consumer instances can be deployed on a server (or multiple servers), and messages can be sent or received by multithreading within the same producer or consumer instance to improve message sending or receiving TPS. Avoid creating one client instance per thread.

Ok, let’s review the content of this article

1. Load balancing and fault tolerance mechanism for message sending

2. Message sending process and storage (how to store them will be explained in the Broker article, as they are stored in the CommitLog and ConsumerQueue of the Broker)

3. Message types: ordinary message (synchronous message, asynchronous message, one-way message), timed message and delayed message, sequential message (global order and partial order), transaction message

O praise

Well, that’s all. I’m xiaoyuxian, your learning partner

I hope that one day I can support myself by writing, and I am still honing my skills. This time may be many years. Thank you for being my original reader and communicator. Please believe that as long as you give me a love, I will eventually return you a page of love.

Thank you again for reading here, I will continue to update the technical articles and some records of the soul of life articles, if you feel good, think [big fish students] something, please like, pay attention to, share three

Oh, right! The subsequent update of the article I will be put here in time, welcome to click to watch, are dry articles ah, suggest collection, at any time to check

Github.com/DayuMM2021/…

Everyone who liked it found a date. Everyone who liked it got on board

Recommended reading

● The interviewer asked me: Are you sure that using BigDecimal is accurate?

● This GitHub address smells good

● Getting started with message queues

● What is RocketMQ