The RocketMQ message model is mainly composed of Producer, Broker and Consumer. The Producer is responsible for producing messages, the Consumer is responsible for consuming messages, and the Broker is responsible for storing messages.
RocketMQ message model is mainly composed of Producer, Broker and Consumer. The producer is responsible for producing messages and the consumer is for consuming messages, while the broker stores messages.
The Broker corresponds to one server during actual deployment, and each Broker can store messages from multiple topics, and messages from each Topic can be stored in a different Broker by sharding strategy.
The broker is an independent server during actual deployment, and each broker can store messages from multiple topics. Even messages from the same topic can be stored in the different brokers by sharding strategy.
The Message Queue is used to store physical offsets of messages, and the Message addresses in multiple Topic are stored in multiple Message queues. The Consumer Group consists of multiple Consumer instances.
The message queue is used to store physical offsets of messages, and the message addresses are stored in seperate queues. The consumer group consists of multiple consumer instances.
## 2 Producer
## 2 Producer
The Producer is responsible for producing messages, typically by business systems. It sends messages generated by the business application systems to brokers. The RocketMQ provides multiple paradigms of sending: synchronous, asynchronous,sequential and one-way. Both synchronous and asynchronous methods require the Broker to return confirmation information, while one-way sending is not required.
The Producer is responsible for producing messages, typically by business systems. It sends messages generated by the systems to brokers. RocketMQ provides multiple paradigms of sending: synchronous, asynchronous, sequential and one-way. Both synchronous and asynchronous methods require the confirmation information return from the Broker, but one-way method does not require it.
## 3 Consumer
## 3 Consumer
The Consumer is responsible for consuming messages, typically the background system is responsible for asynchronous consumption. The Consumer pulls messages from brokers and feeds them into application. In perspective of user application, two types of consumers are provided:Pull Consumer and Push Consumer.
The Consumer is responsible for consuming messages, typically the background system is responsible for asynchronous consumption. The consumer pulls messages from brokers and feeds them into application. From the perspective of user, two types of consumers are provided: pull consumer and push consumer.
## 4 Topic
## 4 Topic
The Topic refers to a collection of one kind message. Each topic contains several messages and one message can only belong to one topic. The Topic is the basic unit of RocketMQ for message subscription.
The Topic refers to a collection of one kind of message. Each topic contains several messages and one message can only belong to one topic. The topic is the basic unit of RocketMQ for message subscription.
## 5 Broker Server
## 5 Broker Server
As the role of the transfer station, the Broker Server stores, and forwards messages. In the RocketMQ system, the Broker Server is responsible for receiving messages sent from producers, storing them and preparing to handle pull requests. It also stores message related meta data, including consumer groups, consuming progress, topics and queues info.
As the role of the transfer station, the Broker Server stores and forwards messages. In RocketMQ, the broker server is responsible for receiving messages sent from producers, storing them and preparing to handle pull requests. It also stores the related message meta data, including consumer groups, consuming progress, topics, queues info and so on.
## 6 Name Server
## 6 Name Server
The Name Server serves as the provider of routing service. The Producer or Consumer can find the list of Broker IP addresses corresponding each topic through name server. Multiple Name Servers can be deployed as a cluster, but are independent of each other and do not exchange information.
The Name Server serves as the provider of routing service. The producer or the consumer can find the list of broker IP addresses for each topic through name server. Multiple name servers can be deployed in one cluster, but they are independent of each other and do not exchange information.
## 7 Pull Consumer
## 7 Pull Consumer
A type of Consumer. Applications are usually pulls messages from brokers by actively calling the Consumer's pull message method, and the application has the advantages of controlling the timing and frequency of pulling messages. Once batches of messages are pulled, user application initiates consuming process.
A type of Consumer, the application pulls messages from brokers by actively invoking the consumer pull message method, and the application has the advantages of controlling the timing and frequency of pulling messages. Once the batch of messages is pulled, user application will initiate consuming process.
## 8 Push Consumer
## 8 Push Consumer
A type of Consumer. Under this mode, after the Broker receives the data, it will actively push it to the consumer, which is generally of high real-time performance.
A type of Consumer. Under this high real-time performance mode, it will push the message to the consumer actively when the Broker receives the data.
## 9 Producer Group
## 9 Producer Group
A collection of the same type of Producer, which sends the same type of messages with consistent logic. If a transaction message is sent and the original producer crashes after sending, the Broker Server will contacts other Producer in the same Producer group to commit or rollback the transactional message.
A collection of the same type of Producer, which sends the same type of messages with consistent logic. If a transaction message is sent and the original producer crashes after sending, the broker server will contact other producers in the same producer group to commit or rollback the transactional message.
## 10 Consumer Group
## 10 Consumer Group
A collection of the same type of Consumer, which sends the same type of messages with consistent logic. The Consumer Group makes load-balance and fault-tolerance super easy in terms of message consuming.
A collection of the same type of Consumer, which sends the same type of messages with consistent logic. The consumer group makes load-balance and fault-tolerance super easy in terms of message consuming.
Warning: consumer instances of one consumer group must have exactly the same topic subscription(s).
Warning: consumer instances of one consumer group must have exactly the same topic subscription(s).
RocketMQ supports two type of consumption mode:Clustering and Broadcasting.
RocketMQ supports two types of consumption mode:Clustering and Broadcasting.
## 11 Consumption Mode - Clustering
## 11 Consumption Mode - Clustering
Under the Clustering mode, all messages from one topic will be delivered to all consumer instances averagely as much as possible. That is, one message can be consumed by only one consumer instance.
Under the Clustering mode, all the messages from one topic will be delivered to all the consumers instances averagely as much as possible. That is, one message can be consumed by only one consumer instance.
## 12 Consumption Mode - Broadcasting
## 12 Consumption Mode - Broadcasting
Under the Broadcasting mode, each Consumer instance of the same Consumer Group receives every message that published to the corresponding topic.
Under the Broadcasting mode, each consumer instance of the same consumer group receives every message published to the corresponding topic.
## 13 Normal Ordered Message
## 13 Normal Ordered Message
Under the Normal Ordered Message mode, the messages received by consumers from the same ConsumeQueue are sequential, while the messages received from different message queues may be non-sequential.
Under the Normal Ordered Message mode, the messages received by consumers from the same ConsumeQueue are sequential, but the messages received from the different message queues may be non-sequential.
## 14 Strictly Ordered Message
## 14 Strictly Ordered Message
Under the Strictly Ordered Message mode, all messages received by the consumers from the same topic are sequential, as the order they were stored.
Under the Strictly Ordered Message mode, all messages received by the consumers from the same topic are sequential as the order they are stored.
## 15 Message
## 15 Message
The physical carrier of information transmitted by a messaging system, the smallest unit of production and consumption data, each message must belong to a topic.
The physical carrier of information transmitted by a messaging system, the smallest unit of production and consumption data, each message must belong to one topic.
Each Message in RocketMQ has a unique Message ID and can carry a key, generally used to store business-related value. The system provides the function to query messages by Message ID and Key.
Each Message in RocketMQ has a unique message id and can carry a key used to store business-related value. The system has the function to query messages by its id or key.
## 16 Tag
## 16 Tag
Flags set for messages to distinguish different types of messages under the same topic, functioning as a "sub-topic". Messages from the same business unit can set different tags under the same topic in terms of different business purposes. The Tag can effectively maintain the clarity and consistency of the code and optimize the query system provided by RocketMQ. The Consumer can realize different "sub-topic" by using Tag, so as to achieve better expansibility.
Flags set for messages to distinguish different types of messages under the same topic, functioning as a "sub-topic". Messages from the same business unit can set different tags under the same topic in terms of different business purposes. The tag can effectively maintain the clarity and consistency of the code and optimize the query system provided by RocketMQ. The consumer can realize different "sub-topic" by using tag in order to achieve better expansibility.
Relative to RocketMQ's Broker cluster, producers and consumers are client. In this section, it mainly describes the common behavior configuration of producers and consumers.
Relative to RocketMQ's Broker cluster, producers and consumers are client. In this section, it mainly describes the common behavior configuration of producers and consumers.
### Client Addressing mode
### 1 Client Addressing mode
```RocketMQ``` can let client find the ```Name Server```, and then find the ```Broker```by the ```Name Server```. Followings show a variety of configurations, and priority level from highly to lower, the highly priority configurations can override the lower priority configurations.
```RocketMQ``` can let client find the ```Name Server```, and then find the ```Broker```by the ```Name Server```. Followings show a variety of configurations, and priority level from highly to lower, the highly priority configurations can override the lower priority configurations.
...
@@ -36,11 +36,11 @@ By default, the client accesses the HTTP server every 2 minutes, and update the
...
@@ -36,11 +36,11 @@ By default, the client accesses the HTTP server every 2 minutes, and update the
```
```
HTTP static server addressing is recommended, because it is simple client deployment, and the Name Server cluster can be upgraded hot.
HTTP static server addressing is recommended, because it is simple client deployment, and the Name Server cluster can be upgraded hot.
### Client Configuration
### 2 Client Configuration
```DefaultMQProducer```,```TransactionMQProducer```,```DefaultMQPushConsumer```,```DefaultMQPullConsumer``` all extends the ```ClientConfig``` Class, ```ClientConfig``` as the client common configuration class. Client configuration style like getXXX,setXXX, each of the parameters can config by spring and also config their in the code. Such as the ```namesrvAddr``` parameter: ```producer.setNamesrvAddr("192.168.0.1:9876")```, same with the other parameters.
```DefaultMQProducer```,```TransactionMQProducer```,```DefaultMQPushConsumer```,```DefaultMQPullConsumer``` all extends the ```ClientConfig``` Class, ```ClientConfig``` as the client common configuration class. Client configuration style like getXXX,setXXX, each of the parameters can config by spring and also config their in the code. Such as the ```namesrvAddr``` parameter: ```producer.setNamesrvAddr("192.168.0.1:9876")```, same with the other parameters.
@@ -70,7 +70,7 @@ HTTP static server addressing is recommended, because it is simple client deploy
...
@@ -70,7 +70,7 @@ HTTP static server addressing is recommended, because it is simple client deploy
| checkRequestHoldMax | 2000 | Producer local buffer request queue size when Broker look back Producer transaction status |
| checkRequestHoldMax | 2000 | Producer local buffer request queue size when Broker look back Producer transaction status |
| RPCHook | null | This parameter is passed in when the Producer is creating, including the pre-processing before the message sending and the processing after the message response. The user can do some security control or other operations in the first interface.|
| RPCHook | null | This parameter is passed in when the Producer is creating, including the pre-processing before the message sending and the processing after the message response. The user can do some security control or other operations in the first interface.|
This section focuses on the configuration of the system (JVM/OS)
This section focuses on the configuration of the system (JVM/OS)
## **1 JVM Options** ##
## **1 JVM Options** ##
The latest released version of JDK 1.8 is recommended. Set the same Xms and Xmx value to prevent the JVM from resizing the heap for better performance. A simple JVM configurations looks like this:
The latest released version of JDK 1.8 is recommended. Set the same Xms and Xmx value to prevent the JVM from resizing the heap for better performance. A simple JVM configuration is as follows:
-server -Xms8g -Xmx8g -Xmn4g
-server -Xms8g -Xmx8g -Xmn4g
...
@@ -22,7 +22,7 @@ As for garbage collection, G1 collector with JDK 1.8 is recommended:
...
@@ -22,7 +22,7 @@ As for garbage collection, G1 collector with JDK 1.8 is recommended:
-XX:G1ReservePercent=25
-XX:G1ReservePercent=25
-XX:InitiatingHeapOccupancyPercent=30
-XX:InitiatingHeapOccupancyPercent=30
These GC options looks a little aggressive, but it’s proved to have good performance in our production environment
These GC options looks a little aggressive, but it’s proved to have good performance in our production environment.
Don’t set a too small value for -XX:MaxGCPauseMillis, otherwise JVM will use a small young generation to achieve this goal which will cause very frequent minor GC.So use rolling GC log file is recommended:
Don’t set a too small value for -XX:MaxGCPauseMillis, otherwise JVM will use a small young generation to achieve this goal which will cause very frequent minor GC.So use rolling GC log file is recommended:
...
@@ -30,13 +30,13 @@ Don’t set a too small value for -XX:MaxGCPauseMillis, otherwise JVM will use a
...
@@ -30,13 +30,13 @@ Don’t set a too small value for -XX:MaxGCPauseMillis, otherwise JVM will use a
-XX:NumberOfGCLogFiles=5
-XX:NumberOfGCLogFiles=5
-XX:GCLogFileSize=30m
-XX:GCLogFileSize=30m
If write GC file will increase latency of broker, consider redirect GC log file to a memory file system:
If writing a GC file increases the latency of broker, please consider redirecting it to memory file system:
-Xloggc:/dev/shm/mq_gc_%p.log123
-Xloggc:/dev/shm/mq_gc_%p.log123
## 2 Linux Kernel Parameters ##
## 2 Linux Kernel Parameters ##
There is a os.sh script that lists a lot of kernel parameters in folder bin which can be used for production use with minor changes. Below parameters need attention, and more details please refer to documentation for /proc/sys/vm/*.
There is an os.sh script that lists a lot of kernel parameters in folder bin which can be used for production enviroment with minor changes. Below parameters need attention, if need more details please refer to documentation for /proc/sys/vm/*.
...
@@ -51,11 +51,11 @@ There is a os.sh script that lists a lot of kernel parameters in folder bin whic
...
@@ -51,11 +51,11 @@ There is a os.sh script that lists a lot of kernel parameters in folder bin whic
-**vm.max_map_count**, limits the maximum number of memory map areas a process may have. RocketMQ will use mmap to load CommitLog and ConsumeQueue, so set a bigger value for this parameter is recommended.
-**vm.max_map_count**, limits the maximum number of memory map areas a process may have. RocketMQ will use mmap to load CommitLog and ConsumeQueue, so a bigger value is recommended.
-**vm.swappiness**, define how aggressive the kernel will swap memory pages. Higher values will increase agressiveness, lower values decrease the amount of swap. 10 is recommended for this value to avoid swap latency.
-**vm.swappiness**, defines how aggressive the kernel will swap memory pages. Higher values will increase agressiveness, lower values decrease the amount of swap. 10 is recommended for this value to avoid swap latency.
...
@@ -64,4 +64,4 @@ There is a os.sh script that lists a lot of kernel parameters in folder bin whic
...
@@ -64,4 +64,4 @@ There is a os.sh script that lists a lot of kernel parameters in folder bin whic
-**Disk scheduler**, the deadline I/O scheduler is recommended for RocketMQ, which attempts to provide a guaranteed latency for requests.
-**Disk scheduler**, the deadline I/O scheduler is recommended for RocketMQ, which attempts to provide a guaranteed latency for requests.
This is the simplest, but also the riskiest, mode that makes the entire service unavailable once the broker restarts or goes down. Production environments are not recommended, but can be used for local testing and development. Here are the steps to build.
This is the simplest but also the riskiest mode, that makes the entire service unavailable once the broker restarts or goes down. Production environments are not recommended, but can be used for local testing and development. Here are the steps to build.
The boot command shown above is used in the case of a single NameServer.For clusters of multiple NameServer, the address list after the -n argument in the broker boot command is separated by semicolons, for example, 192.168.1.1: 9876;192.161.2: 9876.
The boot command shown above is used in the case of a single NameServer.For clusters of multiple NameServer, the address list after the -n argument in the broker boot command is separated by semicolons, for example, 192.168.1.1: 9876;192.161.2: 9876.
### 3 Multiple Master And Multiple Slave Mode-Asynchronous replication
### 3 Multiple Master And Multiple Slave Mode-Asynchronous replication
RocketMQ message queue cluster mainly includes four roles: NameServer, Broker (Master/Slave), Producer and Consumer. The basic communication process is as follows:
RocketMQ message queue cluster mainly includes four roles: NameServer, Broker (Master/Slave), Producer and Consumer. The basic communication process is as follows:
(1) After Broker start-up, it needs to complete one operation: register itself to NameServer, and then report Topic routing information to NameServer at regular intervals of 30 seconds.
(1) After Broker start-up, it needs to complete one operation: register itself to NameServer, and then report Topic routing information to NameServer at regular intervals of 30 seconds.
(2) When message producer Producer sends a message as a client, it needs to obtain routing information from the local cache TopicPublishInfoTable according to the Topic of the message. If not, it will be retrieved from NameServer and update to local cache, at the same time, Producer will retrieve routing information from NameServer every 30 seconds by default.
(2) When message Producer sends a message as a client, it needs to obtain routing information from the local cache TopicPublishInfoTable according to the Topic of the message. If not, it will be retrieved from NameServer and update to local cache, at the same time, Producer will retrieve routing information from NameServer every 30 seconds by default.
(3) Message producer Producer chooses a queue to send the message according to the routing information obtained in 2); Broker receives the message and records it in disk as the receiver of the message.
(3) Message producer Producer chooses a queue to send the message according to the routing information obtained in 2); Broker receives the message and records it in disk as the receiver of the message.
(4) After message consumer Consumer get the routing information according to 2) and complete the load balancing of the client, then select one or several message queues to pull messages and consume them.
(4) After message consumer Consumer get the routing information according to 2) and complete the load balancing of the client, then select one or several message queues to pull messages and consume them.
Message storage is the most complicated and important part of RocketMQ. This section will describe the three aspects of RocketMQ: message storage architecture, PageCache and memory mapping, and RocketMQ's two different disk flushing methods.
Message storage is the most complicated and important part of RocketMQ. This section will describe the three aspects of RocketMQ:
* Message storage architecture
* PageCache and memory mapping
* RocketMQ's two different disk flushing methods.
## 1 Message Storage Architecture
## 1 Message Storage Architecture
...
@@ -18,7 +21,7 @@ The message storage architecture diagram consists of 3 files related to message
...
@@ -18,7 +21,7 @@ The message storage architecture diagram consists of 3 files related to message
From the above architecture of the RocketMQ message storage, we can see RocketMQ uses a hybrid storage structure, that is, all the queues in an instance of the broker share a single log file `CommitLog` to store messages. RocketMQ's hybrid storage structure(messages of multiple topics are stored in one CommitLog) uses a separate storage structure for the data and index parts for Producer and Consumer respectively. The Producer sends the message to the Broker, then the Broker persists the message to the CommitLog file synchronously or asynchronously. As long as the message is persisted to the CommitLog on the disk, the message sent by the Producer will not be lost. Because of this, Consumer will definitely have the opportunity to consume this message. When no message can be pulled, the consumer can wait for the next pull. And the server also supports the long polling mode: if a pull request pulls no messages, the Broker can wait for 30 seconds, as long as new message arrives in this interval, it will be returned directly to the consumer. Here, RocketMQ's specific approach is using Broker's background service thread `ReputMessageService` to continuously dispatch requests and asynchronously build ConsumeQueue (Logical Queue) and IndexFile data.
From the above architecture of the RocketMQ message storage, we can see RocketMQ uses a hybrid storage structure, that is, all the queues in an instance of the broker share a single log file `CommitLog` to store messages. RocketMQ's hybrid storage structure(messages of multiple topics are stored in one CommitLog) uses a separate storage structure for the data and index parts for Producer and Consumer respectively. The Producer sends the message to the Broker, then the Broker persists the message to the CommitLog file synchronously or asynchronously. As long as the message is persisted to the CommitLog on the disk, the message sent by the Producer will not be lost. Because of this, Consumer will definitely have the opportunity to consume this message. When no message can be pulled, the consumer can wait for the next pull. And the server also supports the long polling mode: if a pull request pulls no messages, the Broker can wait for 30 seconds, as long as new message arrives in this interval, it will be returned directly to the consumer. Here, RocketMQ's specific approach is using Broker's background service thread `ReputMessageService` to continuously dispatch requests and asynchronously build ConsumeQueue (Logical Queue) and IndexFile data.
## 2 PageCache and memory map
## 2 PageCache and Memory Map
PageCache is a cache of files by the operating system to speed up the reading and writing of files. In general, the speed of sequential read and write files is almost the same as the speed of read and write memory. The main reason is that the OS uses a portion of the memory as PageCache to optimize the performance of the read and write operations. For data writing, the OS will first write to the Cache, and then the `pdflush` kernel thread asynchronously flush the data in the Cache to the physical disk. For data reading, if it can not hit the page cache when reading a file at a time, the OS will read the file from the physical disk and prefetch the data files of other neighboring blocks sequentially.
PageCache is a cache of files by the operating system to speed up the reading and writing of files. In general, the speed of sequential read and write files is almost the same as the speed of read and write memory. The main reason is that the OS uses a portion of the memory as PageCache to optimize the performance of the read and write operations. For data writing, the OS will first write to the Cache, and then the `pdflush` kernel thread asynchronously flush the data in the Cache to the physical disk. For data reading, if it can not hit the page cache when reading a file at a time, the OS will read the file from the physical disk and prefetch the data files of other neighboring blocks sequentially.
@@ -64,15 +64,15 @@ public class ScheduledMessageProducer {
...
@@ -64,15 +64,15 @@ public class ScheduledMessageProducer {
}
}
```
```
### 3.Verification
### 3Verification
You should see messages are consumed about 10 seconds later than their storing time.
You should see messages are consumed about 10 seconds later than their storing time.
### 4.Use scenarios for scheduled messages
### 4Use scenarios for scheduled messages
For example, in e-commerce, if an order is submitted, a delay message can be sent, and the status of the order can be checked after 1 hour. If the order is still unpaid, the order can be cancelled and the inventory released.
For example, in e-commerce, if an order is submitted, a delay message can be sent, and the status of the order can be checked after 1 hour. If the order is still unpaid, the order can be cancelled and the inventory released.
### 5.Restrictions on the use of scheduled messages
### 5Restrictions on the use of scheduled messages
Nowadays RocketMq does not support any time delay. It needs to set several fixed delay levels, which correspond to level 1 to 18 from 1s to 2h. Message consumption failure will enter the delay message queue. Message sending time is related to the set delay level and the number of retries.
Nowadays RocketMq does not support any time delay. It needs to set several fixed delay levels, which correspond to level 1 to 18 from 1s to 2h. Message consumption failure will enter the delay message queue. Message sending time is related to the set delay level and the number of retries.
The following questions are frequently asked with regard to the RocketMQ project in general.
The following questions are frequently asked with regard to the RocketMQ project in general.
## General
## 1 General
### 1. Why did we create rocketmq project instead of selecting other products?
1. Why did we create rocketmq project instead of selecting other products?
Please refer to [Why RocketMQ](http://rocketmq.apache.org/docs/motivation)
Please refer to [Why RocketMQ](http://rocketmq.apache.org/docs/motivation)
### 2. Do I have to install other softeware, such as zookeeper, to use RocketMQ?
2. Do I have to install other softeware, such as zookeeper, to use RocketMQ?
No. RocketMQ can run independently.
No. RocketMQ can run independently.
## Usage
## 2 Usage
### 1. Where does the newly created Consumer ID start consuming messages?
### 1. Where does the newly created Consumer ID start consuming messages?
1. If the topic sends a message within three days, then the consumer start consuming messages from the first message saved in the server.
 1) If the topic sends a message within three days, then the consumer start consuming messages from the first message saved in the server.
2. If the topic sends a message three days ago, the consumer starts to consume messages from the latest message in the server, in other words, starting from the tail of message queue.
3. If such consumer is rebooted, then it starts to consume messages from the last consumption location.
 2) If the topic sends a message three days ago, the consumer starts to consume messages from the latest message in the server, in other words, starting from the tail of message queue.
 3) If such consumer is rebooted, then it starts to consume messages from the last consumption location.
### 2. How to reconsume message when consumption fails?
### 2. How to reconsume message when consumption fails?
1. Cluster consumption pattern, The consumer business logic code returns Action.ReconsumerLater, NULL, or throws an exception, if a message failed to be consumed, it will retry for up to 16 times, after that, the message would be descarded.
 1) Cluster consumption pattern, The consumer business logic code returns Action.ReconsumerLater, NULL, or throws an exception, if a message failed to be consumed, it will retry for up to 16 times, after that, the message would be descarded.
2. Broadcast consumption patternThe broadcaset consumption still ensures that a message is consumered at least once, but no resend option is provided.
 2) Broadcast consumption patternThe broadcaset consumption still ensures that a message is consumered at least once, but no resend option is provided.
### 3. How to query the failed message if there is a consumption failure?
### 3. How to query the failed message if there is a consumption failure?
1. Using topic query by time, you can query messages within a period of time.
 1) Using topic query by time, you can query messages within a period of time.
2. Using Topic and Message Id to accurately query the message.
3. Using Topic and Message Key accurately query a class of messages with the same Message Key.
 2) Using Topic and Message Id to accurately query the message.
 3) Using Topic and Message Key accurately query a class of messages with the same Message Key.
### 4. Are messages delivered exactly once?
### 4. Are messages delivered exactly once?
...
@@ -37,10 +42,11 @@ RocketMQ ensures that all messages are delivered at least once. In most cases, t
...
@@ -37,10 +42,11 @@ RocketMQ ensures that all messages are delivered at least once. In most cases, t
### 5. How to add a new broker?
### 5. How to add a new broker?
1. Start up a new broker and register it to the same list of name servers.
 1) Start up a new broker and register it to the same list of name servers.
2. By default, only internal system topics and consumer groups are created automatically. If you would like to have your business topic and consumer groups on the new node, please replicate them from the existing broker. Admin tool and command lines are provided to handle this.
 2) By default, only internal system topics and consumer groups are created automatically. If you would like to have your business topic and consumer groups on the new node, please replicate them from the existing broker. Admin tool and command lines are provided to handle this.
## Configuration related
## 3 Configuration related
The following answers are all default values and can be modified by configuration.
The following answers are all default values and can be modified by configuration.
### 1. If you start a producer or consumer failed and the error message is producer group or consumer repeat.
### 1. If you start a producer or consumer failed and the error message is producer group or consumer repeat.
...
@@ -76,15 +82,15 @@ Solution: Fastjson version has to be upgraded to rocketmq client dependent versi
...
@@ -76,15 +82,15 @@ Solution: Fastjson version has to be upgraded to rocketmq client dependent versi
### 3. What is the impact of a broker crash.
### 3. What is the impact of a broker crash.
1. Master crashes
 1) Master crashes
Messages can no longer be sent to this broker set, but if you have another broker set available, messages can still be sent given the topic is present. Messages can still be consumed from slaves.
Messages can no longer be sent to this broker set, but if you have another broker set available, messages can still be sent given the topic is present. Messages can still be consumed from slaves.
2. Some slave crash
 2) Some slave crash
As long as there is another working slave, there will be no impact on sending messages. There will also be no impact on consuming messages except when the consumer group is set to consume from this slave preferably. By default, comsumer group consumes from master.
As long as there is another working slave, there will be no impact on sending messages. There will also be no impact on consuming messages except when the consumer group is set to consume from this slave preferably. By default, comsumer group consumes from master.
3. All slaves crash
 3) All slaves crash
There will be no impact on sending messages to master, but, if the master is SYNC_MASTER, producer will get a SLAVE_NOT_AVAILABLE indicating that the message is not sent to any slaves. There will also be no impact on consuming messages except that if the consumer group is set to consume from slave preferably. By default, comsumer group consumes from master.
There will be no impact on sending messages to master, but, if the master is SYNC_MASTER, producer will get a SLAVE_NOT_AVAILABLE indicating that the message is not sent to any slaves. There will also be no impact on consuming messages except that if the consumer group is set to consume from slave preferably. By default, comsumer group consumes from master.
...
@@ -92,9 +98,12 @@ There will be no impact on sending messages to master, but, if the master is SYN
...
@@ -92,9 +98,12 @@ There will be no impact on sending messages to master, but, if the master is SYN
This happens when you are trying to send messages to a topic whose routing info is not available to the producer.
This happens when you are trying to send messages to a topic whose routing info is not available to the producer.
1. Make sure that the producer can connect to a name server and is capable of fetching routing meta info from it.
 1) Make sure that the producer can connect to a name server and is capable of fetching routing meta info from it.
2. Make sure that name servers do contain routing meta info of the topic. You may query the routing meta info from name server through topicRoute using admin tools or web console.
3. Make sure that your brokers are sending heartbeats to the same list of name servers your producer is connecting to.
 2) Make sure that name servers do contain routing meta info of the topic. You may query the routing meta info from name server through topicRoute using admin tools or web console.
4. Make sure that the topic’s permssion is 6(rw-), or at least 2(-w-).
 3) Make sure that your brokers are sending heartbeats to the same list of name servers your producer is connecting to.
 4) Make sure that the topic’s permssion is 6(rw-), or at least 2(-w-).
If you can’t find this topic, create it on a broker via admin tools command updateTopic or web console.
If you can’t find this topic, create it on a broker via admin tools command updateTopic or web console.
Broker Role is ASYNC_MASTER, SYNC_MASTER or SLAVE. If you cannot tolerate message missing, we suggest you deploy SYNC_MASTER and attach a SLAVE to it. If you feel ok about missing, but you want the Broker to be always available, you may deploy ASYNC_MASTER with SLAVE. If you just want to make it easy, you may only need a ASYNC_MASTER without SLAVE.
## 3.2 FlushDiskType
ASYNC_FLUSH is recommended, for SYNC_FLUSH is expensive and will cause too much performance loss. If you want reliability, we recommend you use SYNC_MASTER with SLAVE.
| brokerIP1 | InetAddress for network interface | Should be configured if having multiple addresses |
| brokerIP2 | InetAddress for network interface | If configured for the Master broker in the Master/Slave cluster, slave broker will connect to this port for data synchronization |
| brokerName | null | broker name |
| brokerClusterName | DefaultCluster | this broker belongs to which cluster |
| brokerId | 0 | broker id, 0 means master, positive integers mean slave |
| flushDiskType | ASYNC_FLUSH | {SYNC_FLUSH/ASYNC_FLUSH}. Broker of SYNC_FLUSH mode flushes each message onto disk before acknowledging producer. Broker of ASYNC_FLUSH mode, on the other hand, takes advantage of group-committing, achieving better performance. |
RocketMQ cannot avoid Exactly-Once, so if the business is very sensitive to consumption repetition, it is important to perform deduplication at the business level. Deduplication can be done with a relational database. First, you need to determine the unique key of the message, which can be either msgId or a unique identifier field in the message content, such as the order Id. Determine if a unique key exists in the relational database before consumption. If it does not exist, insert it and consume it, otherwise skip it. (The actual process should consider the atomic problem, determine whether there is an attempt to insert, if the primary key conflicts, the insertion fails, skip directly)
RocketMQ cannot avoid Exactly-Once, so if the business is very sensitive to consumption repetition, it is important to perform deduplication at the business level. Deduplication can be done with a relational database. First, you need to determine the unique key of the message, which can be either msgId or a unique identifier field in the message content, such as the order Id. Determine if a unique key exists in the relational database before consumption. If it does not exist, insert it and consume it, otherwise skip it. (The actual process should consider the atomic problem, determine whether there is an attempt to insert, if the primary key conflicts, the insertion fails, skip directly)
### 2.2 Slow message processing
### 2 Slow message processing
#### 2.2.1 Increase consumption parallelism
#### 2.1 Increase consumption parallelism
Most messages consumption behaviors are IO-intensive, That is, it may be to operate the database, or call RPC. The consumption speed of such consumption behavior lies in the throughput of the back-end database or the external system. By increasing the consumption parallelism, the total consumption throughput can be increased, but the degree of parallelism is increased to a certain extent. Instead it will fall.Therefore, the application must set a reasonable degree of parallelism. There are several ways to modify the degree of parallelism of consumption as follows:
Most messages consumption behaviors are IO-intensive, That is, it may be to operate the database, or call RPC. The consumption speed of such consumption behavior lies in the throughput of the back-end database or the external system. By increasing the consumption parallelism, the total consumption throughput can be increased, but the degree of parallelism is increased to a certain extent. Instead it will fall.Therefore, the application must set a reasonable degree of parallelism. There are several ways to modify the degree of parallelism of consumption as follows:
* Under the same ConsumerGroup, increase the degree of parallelism by increasing the number of Consumer instances (note that the Consumer instance that exceeds the number of subscription queues is invalid). Can be done by adding machines, or by starting multiple processes on an existing machine.
* Under the same ConsumerGroup, increase the degree of parallelism by increasing the number of Consumer instances (note that the Consumer instance that exceeds the number of subscription queues is invalid). Can be done by adding machines, or by starting multiple processes on an existing machine.
* Improve the consumption parallel thread of a single Consumer by modifying the parameters consumeThreadMin and consumeThreadMax.
* Improve the consumption parallel thread of a single Consumer by modifying the parameters consumeThreadMin and consumeThreadMax.
#### 2.2.2 Batch mode consumption
#### 2.2 Batch mode consumption
Some business processes can increase consumption throughput to a large extent if they support batch mode consumption. For example, order deduction application, it takes 1s to process one order at a time, and it takes only 2s to process 10 orders at a time. In this way, the throughput of consumption can be greatly improved. By setting the consumer's consumeMessageBatchMaxSize to return a parameter, the default is 1, that is, only one message is consumed at a time, for example, set to N, then the number of messages consumed each time is less than or equal to N.
Some business processes can increase consumption throughput to a large extent if they support batch mode consumption. For example, order deduction application, it takes 1s to process one order at a time, and it takes only 2s to process 10 orders at a time. In this way, the throughput of consumption can be greatly improved. By setting the consumer's consumeMessageBatchMaxSize to return a parameter, the default is 1, that is, only one message is consumed at a time, for example, set to N, then the number of messages consumed each time is less than or equal to N.
#### 2.2.3 Skip non-critical messages
#### 2.3 Skip non-critical messages
When a message is accumulated, if the consumption speed cannot keep up with the transmission speed, if the service does not require high data, you can choose to discard the unimportant message. For example, when the number of messages in a queue is more than 100,000 , try to discard some or all of the messages, so that you can quickly catch up with the speed of sending messages. The sample code is as follows:
When a message is accumulated, if the consumption speed cannot keep up with the transmission speed, if the service does not require high data, you can choose to discard the unimportant message. For example, when the number of messages in a queue is more than 100,000 , try to discard some or all of the messages, so that you can quickly catch up with the speed of sending messages. The sample code is as follows:
...
@@ -40,7 +40,7 @@ public ConsumeConcurrentlyStatus consumeMessage(
...
@@ -40,7 +40,7 @@ public ConsumeConcurrentlyStatus consumeMessage(
}
}
```
```
#### 2.2.4 Optimize each message consumption process
#### 2.4 Optimize each message consumption process
For example, the consumption process of a message is as follows:
For example, the consumption process of a message is as follows:
...
@@ -52,7 +52,7 @@ For example, the consumption process of a message is as follows:
...
@@ -52,7 +52,7 @@ For example, the consumption process of a message is as follows:
There are 4 interactions with the DB in the consumption process of this message. If it is calculated by 5ms each time, it takes a total of 20ms. If the business calculation takes 5ms, then the total time is 25ms, So if you can optimize 4 DB interactions to 2 times, the total time can be optimized to 15ms, which means the overall performance is increased by 40%. Therefore, if the application is sensitive to delay, the DB can be deployed on the SSD hard disk. Compared with the SCSI disk, the former RT will be much smaller.
There are 4 interactions with the DB in the consumption process of this message. If it is calculated by 5ms each time, it takes a total of 20ms. If the business calculation takes 5ms, then the total time is 25ms, So if you can optimize 4 DB interactions to 2 times, the total time can be optimized to 15ms, which means the overall performance is increased by 40%. Therefore, if the application is sensitive to delay, the DB can be deployed on the SSD hard disk. Compared with the SCSI disk, the former RT will be much smaller.
### 2.3 Print Log
### 3 Print Log
If the amount of messages is small, it is recommended to print the message in the consumption entry method, consume time, etc., to facilitate subsequent troubleshooting.
If the amount of messages is small, it is recommended to print the message in the consumption entry method, consume time, etc., to facilitate subsequent troubleshooting.
...
@@ -68,33 +68,33 @@ public ConsumeConcurrentlyStatus consumeMessage(
...
@@ -68,33 +68,33 @@ public ConsumeConcurrentlyStatus consumeMessage(
If you can print the time spent on each message, it will be more convenient when troubleshooting online problems such as slow consumption.
If you can print the time spent on each message, it will be more convenient when troubleshooting online problems such as slow consumption.
### 2.4 Other consumption suggestions
### 4 Other consumption suggestions
#### 2.4.1、Consumer Group and Subscriptions
#### 4.1、Consumer Group and Subscriptions
The first thing you should be aware of is that different Consumer Group can consume the same topic independently, and each of them will have their own consuming offsets. Please make sure each Consumer within the same Group to subscribe the same topics.
The first thing you should be aware of is that different Consumer Group can consume the same topic independently, and each of them will have their own consuming offsets. Please make sure each Consumer within the same Group to subscribe the same topics.
#### 2.4.2、Orderly
#### 4.2、Orderly
The Consumer will lock each MessageQueue to make sure it is consumed one by one in order. This will cause a performance loss, but it is useful when you care about the order of the messages. It is not recommended to throw exceptions, you can return ConsumeOrderlyStatus.SUSPEND_CURRENT_QUEUE_A_MOMENT instead.
The Consumer will lock each MessageQueue to make sure it is consumed one by one in order. This will cause a performance loss, but it is useful when you care about the order of the messages. It is not recommended to throw exceptions, you can return ConsumeOrderlyStatus.SUSPEND_CURRENT_QUEUE_A_MOMENT instead.
#### 2.4.3、Concurrently
#### 4.3、Concurrently
As the name tells, the Consumer will consume the messages concurrently. It is recommended to use this for good performance. It is not recommended to throw exceptions, you can return ConsumeConcurrentlyStatus.RECONSUME_LATER instead.
As the name tells, the Consumer will consume the messages concurrently. It is recommended to use this for good performance. It is not recommended to throw exceptions, you can return ConsumeConcurrentlyStatus.RECONSUME_LATER instead.
#### 2.4.4、Consume Status
#### 4.4、Consume Status
For MessageListenerConcurrently, you can return RECONSUME_LATER to tell the consumer that you can not consume it right now and want to reconsume it later. Then you can continue to consume other messages. For MessageListenerOrderly, because you care about the order, you can not jump over the message, but you can return SUSPEND_CURRENT_QUEUE_A_MOMENT to tell the consumer to wait for a moment.
For MessageListenerConcurrently, you can return RECONSUME_LATER to tell the consumer that you can not consume it right now and want to reconsume it later. Then you can continue to consume other messages. For MessageListenerOrderly, because you care about the order, you can not jump over the message, but you can return SUSPEND_CURRENT_QUEUE_A_MOMENT to tell the consumer to wait for a moment.
#### 2.4.5、Blocking
#### 4.5、Blocking
It is not recommend to block the Listener, because it will block the thread pool, and eventually may stop the consuming process.
It is not recommend to block the Listener, because it will block the thread pool, and eventually may stop the consuming process.
#### 2.4.6、Thread Number
#### 4.6、Thread Number
The consumer use a ThreadPoolExecutor to process consuming internally, so you can change it by setting setConsumeThreadMin or setConsumeThreadMax.
The consumer use a ThreadPoolExecutor to process consuming internally, so you can change it by setting setConsumeThreadMin or setConsumeThreadMax.
#### 2.4.7、ConsumeFromWhere
#### 4.7、ConsumeFromWhere
When a new Consumer Group is established, it will need to decide whether it needs to consume the historical messages which had already existed in the Broker. CONSUME_FROM_LAST_OFFSET will ignore the historical messages, and consume anything produced after that. CONSUME_FROM_FIRST_OFFSET will consume every message existed in the Broker. You can also use CONSUME_FROM_TIMESTAMP to consume messages produced after the specified timestamp.
When a new Consumer Group is established, it will need to decide whether it needs to consume the historical messages which had already existed in the Broker. CONSUME_FROM_LAST_OFFSET will ignore the historical messages, and consume anything produced after that. CONSUME_FROM_FIRST_OFFSET will consume every message existed in the Broker. You can also use CONSUME_FROM_TIMESTAMP to consume messages produced after the specified timestamp.
One application instance should use one topic as much as possible and the subtype of messages can be marked by tags. Tag provides extra flexibility to users. In the consume subscribing process, the messages filtering can only be handled by using tags when the tags are specified in the message sending process: `message.setTags("TagA")`
One application instance should use one topic as much as possible and the subtype of messages can be marked by tags. Tag provides extra flexibility to users. In the consume subscribing process, the messages filtering can only be handled by using tags when the tags are specified in the message sending process: `message.setTags("TagA")`.
###### 1.2 The Use of Keys
###### 1.2 The Use of Keys
A business key can be set in one message and it will be easier to look up the message on a broker server to diagnose issues during development. Each message will be created index(hash index) by server, instance can query the content of this message by topic and key and who consumes the message.Because of the hash index, make sure that the key should be unique in order to avoid potential hash index conflict.
A business key can be set in one message and it will be easier to look up the message on a broker server to diagnose issues during development. Each message will be created index(hash index) by server, instance can query the content of this message by topic and key and who consumes the message.Because of the hash index, make sure that the key should be unique in order to avoid potential hash index conflict.
``` java
``` java
...
@@ -41,4 +41,4 @@ The message sending is usually a process like below:
...
@@ -41,4 +41,4 @@ The message sending is usually a process like below:
* Sever handles request
* Sever handles request
* Sever returns response to client
* Sever returns response to client
The total costing time of sending one message is the sum of costing time of three steps above. Some situations demand that total costing time must be in a quite low level, however, do not take reliable performance into consideration, such as log collection. This kind of application could be called in one-way mode, which means client sends request but not wait for response. In this kind of mode, the cost of sending request is only a call of system operation which means one operation writing data to client socket buffer. Generally, the time cost of this process will be controlled n microseconds level.
The total costing time of sending one message is the sum of costing time of three steps above. Some situations demand that total costing time must be in a quite low level, however, do not take reliable performance into consideration, such as log collection. This kind of application could be called in one-way mode, which means client sends request but not wait for response. In this kind of mode, the cost of sending request is only a call of system operation which means one operation writing data to client socket buffer. Generally, the time cost of this process will be controlled n microseconds level.
| whether the message was sent successfully | Whether the message was successfully consumed | The Key of the message |
| whether the message was sent successfully | Whether the message was successfully consumed | The Key of the message |
| Time spent sending | Time spent consuming | Tag of the message |
| Time spent sending | Time spent consuming | Tag of the message |
## 2. Support for Message Trace Cluster Deployment
## 2 Support for Message Trace Cluster Deployment
### 2.1 Broker Configuration Fille
### 2.1 Broker Configuration Fille
...
@@ -44,7 +42,7 @@ For scenarios with large amount of trace message data , one of the Broker nodes
...
@@ -44,7 +42,7 @@ For scenarios with large amount of trace message data , one of the Broker nodes
### 2.4 Start the Broker that Starts the MessageTrace
### 2.4 Start the Broker that Starts the MessageTrace
`nohup sh mqbroker -c ../conf/2m-noslave/broker-a.properties &`
`nohup sh mqbroker -c ../conf/2m-noslave/broker-a.properties &`
## 3. Save the Topic Definition of Message Trace
## 3 Save the Topic Definition of Message Trace
RocketMQ's message trace feature supports two ways to store trace data:
RocketMQ's message trace feature supports two ways to store trace data:
### 3.1 System-level TraceTopic
### 3.1 System-level TraceTopic
...
@@ -53,7 +51,7 @@ By default, message track data is stored in the system-level TraceTopic(names:
...
@@ -53,7 +51,7 @@ By default, message track data is stored in the system-level TraceTopic(names:
### 3.2 Custom TraceTopic
### 3.2 Custom TraceTopic
If the user is not prepared to store the message track data in the system-level default TraceTopic, you can also define and create a user-level Topic to save the track (that is, to create a regular Topic to save the message track data)。The following section introduces how the Client interface supports the user-defined TraceTopic.
If the user is not prepared to store the message track data in the system-level default TraceTopic, you can also define and create a user-level Topic to save the track (that is, to create a regular Topic to save the message track data)。The following section introduces how the Client interface supports the user-defined TraceTopic.
## 4. Client Practices that Support Message Trace
## 4 Client Practices that Support Message Trace
In order to reduce as much as possible the transformation work of RocketMQ message trace feature used in the user service system, the author added a switch parameter (**enableMsgTrace**) to the original interface in the design to realize whether the message trace is opened or not.
In order to reduce as much as possible the transformation work of RocketMQ message trace feature used in the user service system, the author added a switch parameter (**enableMsgTrace**) to the original interface in the design to realize whether the message trace is opened or not.
### 4.1 Opening the Message Trace when Sending the Message
### 4.1 Opening the Message Trace when Sending the Message
> Problem: Sometimes after deploying the RocketMQ cluster, when you try to execute some commands of "mqadmin", the following exception will appear:
> Problem: Sometimes after deploying the RocketMQ cluster, when you try to execute some commands of "mqadmin", the following exception will appear:
>
>
...
@@ -10,7 +10,7 @@
...
@@ -10,7 +10,7 @@
Solution: Execute `export NAMESRV_ADDR=ip:9876` (ip refers to the address of NameServer deployed in the cluster) on the VM that deploys the RocketMQ cluster.Then you will execute commands of "mqadmin" successfully.
Solution: Execute `export NAMESRV_ADDR=ip:9876` (ip refers to the address of NameServer deployed in the cluster) on the VM that deploys the RocketMQ cluster.Then you will execute commands of "mqadmin" successfully.
## 2. The inconsistent version of RocketMQ between the producer and consumer leads to the problem that message can't be consumed normally.
## 2 The inconsistent version of RocketMQ between the producer and consumer leads to the problem that message can't be consumed normally.
> Problem: The same producer sends a message, consumer A can consume, but consumer B can't consume, and the RocketMQ Console appears:
> Problem: The same producer sends a message, consumer A can consume, but consumer B can't consume, and the RocketMQ Console appears:
>
>
...
@@ -20,7 +20,7 @@ Solution: Execute `export NAMESRV_ADDR=ip:9876` (ip refers to the address of Na
...
@@ -20,7 +20,7 @@ Solution: Execute `export NAMESRV_ADDR=ip:9876` (ip refers to the address of Na
Solution: The jar package of RocketMQ, such as rocketmq-client, should be the same version on the consumer and producer.
Solution: The jar package of RocketMQ, such as rocketmq-client, should be the same version on the consumer and producer.
## 3. When adding a new topic consumer group, historical messages can't be consumed.
## 3 When adding a new topic consumer group, historical messages can't be consumed.
> Problem: When a new consumer group of the same topic is started, the consumed message is the current offset message, and the historical message is not obtained.
> Problem: When a new consumer group of the same topic is started, the consumed message is the current offset message, and the historical message is not obtained.
In some cases, the Consumer needs to reset the consume position to 1-2 days ago. At this time, on the Master Broker with limited memory, the CommitLog will carry a relatively heavy IO pressure, affecting the reading and writing of other messages on that Broker. You can enable `slaveReadEnable=true`. When Master Broker finds that the difference between the Consumer's consume position and the latest value of CommitLog exceeds the percentage of machine's memory (`accessMessageInMemoryMaxRatio=40%`), it will recommend Consumer to read from Slave Broker and relieve Master Broker's IO.
In some cases, the Consumer needs to reset the consume position to 1-2 days ago. At this time, on the Master Broker with limited memory, the CommitLog will carry a relatively heavy IO pressure, affecting the reading and writing of other messages on that Broker. You can enable `slaveReadEnable=true`. When Master Broker finds that the difference between the Consumer's consume position and the latest value of CommitLog exceeds the percentage of machine's memory (`accessMessageInMemoryMaxRatio=40%`), it will recommend Consumer to read from Slave Broker and relieve Master Broker's IO.
## 5. Performance
## 5 Performance
Asynchronous flush disk is recommended to use spin lock.
Asynchronous flush disk is recommended to use spin lock.
...
@@ -64,7 +64,7 @@ Asynchronous flush disk is recommended to open `TransientStorePoolEnable` and cl
...
@@ -64,7 +64,7 @@ Asynchronous flush disk is recommended to open `TransientStorePoolEnable` and cl
Synchronous flush disk is recommended to increase the `sendMessageThreadPoolNums` appropriately. The specific configuration needs to be tested.
Synchronous flush disk is recommended to increase the `sendMessageThreadPoolNums` appropriately. The specific configuration needs to be tested.
## 6. The meaning and difference between msgId and offsetMsgId in RocketMQ
## 6 The meaning and difference between msgId and offsetMsgId in RocketMQ
After sending message with RocketMQ, you will usually see the following log:
After sending message with RocketMQ, you will usually see the following log:
* 1. Finish dispatching the messages fall behind, then to start other services.
* 2. DLedger committedPos may be missing, so here just require dispatchBehindBytes <= 0
*/
while(true){
if(dispatchBehindBytes()<=0){
break;
}
Thread.sleep(1000);
log.info("Try to finish doing reput the messages fall behind during the starting, reputOffset={} maxOffset={} behind={}",this.reputMessageService.getReputFromOffset(),this.getMaxPhyOffset(),this.dispatchBehindBytes());
log.warn("The reputFromOffset={} is smaller than minPyOffset={}, this usually indicate that the dispatch behind too much and the commitlog has expired.",