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

This article introduces the relevant knowledge of "ActiveMQ architecture design and practice case analysis". In the operation of actual cases, many people will encounter such a dilemma, so let the editor lead you to learn how to deal with these situations. I hope you can read it carefully and be able to achieve something!

1. Outline of architecture design

ActiveMQ provides two architecture models that can be implemented: "Mmurs" and "network bridge"; where "Mmurs" is a HA scheme, and "network forwarding bridge" is used to implement "distributed queues".

1.1, Mmurs

Under the Master-Slave model, 2 + ActiveMQ instances are usually needed, and only one instance is Master at any time, providing "production" and "consumption" services to Client, and Slaves is used to do backup or wait for the role to take over when Failover.

Mmurs model is the most general architecture model, which provides a "high availability" feature. When Master fails, one of the Slaves is upgraded to master to continue to provide services, and messages can still be recovered after Failover. Depending on the underlying storage, there may be message loss.

There are two key points:

First, in the Mmurs framework, when it comes to elections, the first condition for elections is to have the support of "exclusive lock". Exclusive lock can be realized by sharing file lock, JDBC database exclusive lock, JDBC lock lease, zookeeper distributed lock and so on. It depends on the mechanism of your underlying storage.

Secondly, there are many mechanisms for message storage in Mmurs architecture, such as "shared file storage", "JDBC" storage, "non-shared storage" and so on. Different storage mechanisms have their own advantages and disadvantages. Be sure to weigh it when using it.

1.2.Network forwarding bridge (network bridge)

In any case, the number of messages and Client concurrency level that a group of Mmurs can carry is always limited. When our message size reaches the upper limit of a stand-alone machine, we should use a cluster-based approach to distribute and load balance messages and Client.

ActiveMQ provides a "network forwarding bridge" mode, and the core idea is:

1. Multiple broker in the cluster communicate with each other through "connection", and forward and store messages among multiple Broker, providing "load balancing" at the storage level.

2. According to the concurrency of Client, the Client is dynamically balanced to support large-scale producers and consumers.

This is similar to the core idea of Kafka.

2. Detailed explanation of Mmurs architecture design.

2.1. Non-shared storage mode

There are 2 + ActiveMQ instances in the cluster, each of which stores data separately. Master saves the message locally and forwards the message to Slaves asynchronously.

Master and slaves are deployed independently, each responsible for their own storage, and the connection between Master and slaves is established through a "network connector" connection, usually an one-way connection between Master and slaves. All messages received on master will be forwarded to slaves.

There are the following main points:

1) at any time, only Master provides services to Clients, and slaves serves as backup only. The old way of shadow nodes.

2) messages stored on slaves have a short delay.

3) master is always master. When master fails, we cannot switch roles at will. The best way to implement this is to restart master. Only when master fails physically will we consider upgrading slave to master. (this is really lame)

4) when slaves needs to be promoted to master, you should ensure that the messages for this slaves are up-to-date.

5) if slaves is offline, you should also manually synchronize the data of master to slaves before restarting slaves. Otherwise, the data during the offline period of slave will not be repeated on slaves.

6) failover protocol is not supported on Client, that is, Client will only establish a connection with master.

This architecture is the most primitive architecture, easy to real-time, but the problem is more serious, the lack of Failover mechanism, the reliability of messages can not be completely guaranteed, because there is no arbitrator or lack of distributed exclusive lock mechanism for the role switching between master and slaves. In Production environments, it is not recommended and can be used as an alternative if you can tolerate SLA degradation during failover.

2.2. Shared file storage

That is, using SAN (GFS) technology, network-based global shared file system model (really a ranking manufacturing machine), this architecture is easy to use, but the ability of architecture and design is weak, so it can be used in Production environment.

SAN storage, you can refer to GFS. The configurations of master and slaves are consistent, and each broker needs a unique brokerName;broker instance to acquire the exclusive lock of the file system through SAN at startup. The instance that acquires lock will become master, and other brokers will wait for lock and intermittently attempt to acquire the lock. Slaves does not provide Clients service. Because brokers writes data to GFS, after failover, the new master still gets the same view of the data as the original master, since GFS is a global shared file system.

We usually use kahaDB as the storage engine, even using log files; the storage efficiency of kahaDB is very high, and the TPS can be as high as 2W, which is an efficient and strong data recovery storage mechanism.

In this architecture mode, failover is supported. When master fails, Clients can reconnect with the new master through failover protocol, and the service interruption time is very short. Because it is based on GFS storage, the data is always stored in the remote shared storage area, so there is no problem of data loss.

The only problem is the stability of GFS (SAN). This needs to be determined that network communication between nodes in the SAN area must be stable and efficient. Build your own services such as NFS, or based on AWS EFS.

This is just transferring one problem to another component.

2.3.Based on JDBC shared storage

We can use the database that supports JDBC as the shared storage layer, that is, master writes the data to the database and does not save any data locally. During the failover period, after slave is upgraded to master, the new master can read data from the database, which also means that the data view of master and slaves is the same throughout the cycle (same as the SAN schema), so the resilience and consistency of the data can be guaranteed. There is also no data loss (in the storage layer).

However, the performance of JDBC storage mechanism is too low, and the performance is nearly 10 times lower than that of kahaDB, a log-based storage layer.

Oh my god . However, in some scenarios with low concurrency and pure decoupling, it is possible. To use the Internet with high concurrency is to kill yourself (the Internet does not necessarily mean it is highly concurrent).

If your business requirements show that data loss is intolerable and the level of SLA is very high, then JDBC may be your best choice.

Since the JDBC database is the final storage layer, we often need to pay attention to the availability of the database, for example, the database is based on Mmurs mode, etc. If the database fails, it will cause the ActiveMQ cluster to be unavailable.

The biggest problem facing JDBC storage is "TPS" (throughput capacity), which is indeed several times lower than kahaDB. If your business has a peak, the "peak clipping" strategy can first write the message to the local file (and then synchronize it to AcitveMQ Broker asynchronously).

I always have a question at this time. It is fine to use the database directly, why do you bother to use ActiveMQ so much? Maybe it's for the sake of design.

3. Network bridges schema architecture

This architecture model mainly deals with the scenario of large-scale Clients and high-density message increment; it will carry applications with a large amount of data in the mode of cluster.

It has the following requirements and characteristics.

1) A large number of Producers and Consumers clients are connected. Only so, perhaps because the message channel (Topic,Queue) in the direction of horizontal expansion, there is not much possibility of split. So they huddled together.

2) maybe the increment of messages is huge, especially some "non-persistent messages". We pin our hopes on building a "distributed queue" architecture. That is, other systems can not solve the problem, I hope the message queue can be buffered.

3) because of the large size of the cluster, we may allow some nodes of the cluster to be offline for a short time, but the data recovery mechanism still needs to be provided. generally speaking, the cluster still provides high availability.

4) the cluster supports load balancing of Clients. For example, when there are multiple producers, these producers will be dynamically balanced among multiple brokers. Otherwise, uneven distribution will pose a risk.

5) failover is supported, that is, when a broker fails, the Clients can reconnect with other brokers; when a new brokers in the cluster is added, the cluster topology can also be dynamically notified to the Clients. This is the favorite of the operation and maintenance staff, and no one wants to get up in the middle of the night and tamper with the machine.

The cluster is composed of multiple sub-Groups, each Group is Mmurs mode and shared storage; multiple Groups are connected based on "network Connector" (masterslave protocol), usually two-way connection, all the Groups are connected to each other, and the Groups forms a "subscription" relationship, for example, G2 is logically G1 subscriber (the subscription strategy is classified according to the Destination list of consumers on each Broker), the message forwarding principle is also based on this.

For Client, it still supports node addresses in the failover,failover protocol that can contain "majority pie" in the cluster.

Messages from Topic subscribers will be copied and stored in all Group; for Queue messages, they will be forwarded between brokers and eventually reach the node where the Consumer is located.

Producers and Consumers can establish connections and communicate messages with Master in any Group. When the topology of the Brokers cluster changes and the number of Producers or Consumers changes, the connection location of the Clients will be dynamically balanced. Brokers synchronizes the connection information of Clients through the "advisory" mechanism. For example, if a new Consumers joins, Broker will send advisory messages (internal channels) to notify other brokers.

The cluster mode provides good availability assurance, and some features may need to be weighed, such as the orderliness of Queue messages will be broken, because multiple Consumer of the same Queue may be on different Group, and if a Group implementation is implemented, the messages stored on it will not be visible to the Clients until it is restored.

4. Performance evaluation

To sum up, there are only three architectures that we can really adopt in a Production environment:

1) shared database mode based on JDBC: HA architecture, in which a single Group,Group contains a master and any number of slaves; Brokers to access data through a remote shared database. The Failover protocol is supported for the client.

2) build a distributed message cluster based on Network Bridge: Cluster architecture, there are multiple Group in the cluster, and each Group is based on shared storage. For Clients, load balancer and Failover; messages are supported to start from Producer and reach Broker nodes. According to "Consumers distribution in the cluster", Broker forwards messages to the Broker where Consumers resides to achieve message flow on demand.

3) simplified modification based on Network Bridge: similar to 2), but each "Group" has only one Broker node, and this Broker is based on kahaDB local file storage, that is, it lacks the HA feature relative to 2) Group; when the Broker node fails, the messages on it will not be visible until the Broker returns to normal. This simplified architecture model reduces the scale of data impact by increasing the number of machines and subdividing the distribution of messages, because it is based on kahaDB local log storage, so its performance is very high.

4.1. share JDBC test results

Production side configuration.

Producer side (pressure output machine): quantity: 4 hardware configurations: 16Core, 32G, CVM software configuration: JVMs 1.8JVMs 24G concurrency and threads: 32 concurrent threads, connection pool 128, send text messages, each message 128 character entity. Message: persistence, Queue, non-transaction

Broker side configuration.

Number of Broker (pressure bearer): 2 hardware configurations: 16Core, 32G, CVM software configuration: JDK 1.8 JVMs 24G architecture mode: Mmurs mode, enable asynchronous forwarding, disable FlowControl, database connection pool is 1024

Storage layer configuration.

Number of databases (storage layer): 2 sets of hardware configuration: 16Core, 32G Magi SSD (IOPS 3000), CVM architecture mode: Mmurs database: MySQL

Test results:

1. Message production efficiency: 1500 TPS

2. Broker load: CPU 30%, memory utilization 11%

3. MySQL load: CPU 46% IOP wait 25%

Conclusion:

1. Based on the shared JDBC storage architecture, the performance is indeed low.

2. The key point that affects the performance is the concurrent IO capability of the database. When the TPS is about 1800, the disk of the database (including slave synchronous IO) has already appeared a higher IO_WAIT.

3. By upgrading disks and adding IOPS, TPS metrics can be effectively improved. It is recommended to increase the number of CPU at the same time.

Students who are going to use the database to implement HA, have you seen this fucking TPS?

4.2. Test results based on non-shared file storage

Test a single ActiveMQ. Based on kahaDB storage, kahaDB is divided into two data flushing modes:

1) swipe the disk one by one

2) swipe every other second

The stress test environment is consistent with 1), except that the disk of the ActiveMQ machine is changed to: SSD (600IOPS).

Test results:

1) refresh the disk one by one

TPS:660

Broker IO_WAIT:19%

2) flush the disk every other second

TPS:9800

Broker IO_WAIT:1.6% (optimizing disk and IOPS in principle, etc., should still be able to improve)

Thus it can be seen that the storage performance based on log files is nearly 5 times higher than that of JDBC, in which the reliability of messages is the highest in one-by-one strategy, but the performance is lower than that of JDBC. If it is based on the "flush every other second" strategy, in extreme cases, it may result in the most recent second of data loss.

Not bad, but it's still a long way from a MQ like kafka.

4.3. Test results based on forwarding bridge

Based on the architecture of forwarding bridge, the implementation cost is high, the maintenance cost is high, and the architecture complexity is relatively high. Based on practical experience, I do not recommend this model. If you want to try, it doesn't matter, after all, it is the "distributed queue implementation mechanism" officially recommended by ActiveMQ, which in principle can support large-scale message storage.

But I have something lighter, so why use you?

4.4. Excellent practice

Therefore, this excellent practice is based on the above tests.

If we finally have to face the storage of "massive messages", after the queue is split according to the business, we still need to face the message volume of a simple business that can not be satisfied by the "single Mmurs architecture". However, we are not willing to bear the potential problems caused by the complexity of the Cluster schema, so we can adopt the more general "logical distribution" mechanism.

1) multiple Group groups are constructed, but there is no physical association between each Mmurs, that is, they do not communicate with each other and do not share storage.

2) on the client side of Producer, add a "router" layer, that is, develop a Client Wrapper, which provides the interfaces commonly used by Producer, and holds multiple ConnectionFactory of Mmurs groups. Before sending messages through the underlying channel, perform hash calculation based on a property in message or a specified KEY, and then select the corresponding connection (or the wrapper class of Spring), and then send the message. This is somewhat similar to the strategy of client-based database read-write separation.

3) for Consumers, you only need to configure multiple ConnectionFactory.

4) after the above practice, the message is sharding to multiple Mmurs groups, which solves the problem of message sending efficiency, and the logical cluster can be expanded on a large scale. And transparent to Client.

5) if you do not want to develop the shard-router layer, we can still achieve "logically distributed" message hash storage based on the failover protocol. At this time, we need to specify the list of all Groups brokers nodes in the failover protocol, and randomize=true. In this way, messages can be stored on multiple Group, and the only pity is that the lack of an "automatic load balancing strategy" may lead to uneven distribution of messages.

The configuration is as follows:

Failover: (tcp://G1.master,tcp://G1.slave,tcp://G2.master,tcp://G2.slave)? randomize=true / / randomize must be true "ActiveMQ Architecture Design and practice case study". Thank you for reading. If you want to know more about the industry, you can follow the website, the editor will output more high-quality practical articles for you!

Welcome to subscribe "Shulou Technology Information " to get latest news, interesting things and hot topics in the IT industry, and controls the hottest and latest Internet news, technology news and IT industry trends.