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

This article mainly introduces the relevant knowledge of "what is the principle of Java distributed architecture". The editor shows you the operation process through an actual case, and the operation method is simple, fast and practical. I hope this article "what is the principle of Java distributed architecture" can help you solve the problem.

1. The distributed term 1.1. Abnormal server downtime

Memory errors, server power outages and so on will lead to server downtime, when the node does not work properly, which is called unavailable.

Server downtime may cause nodes to lose all memory information, so memory information needs to be saved to persistent media.

Network anomaly

There is a special network exception called network partition, that is, all nodes in the cluster are divided into multiple areas, and each area can communicate within, but not between regions.

Disk failure

Disk failure is an exception with a high probability of occurrence.

Use a redundancy mechanism to store data on multiple servers.

1.2. Timeout

In a distributed system, a request not only has success and failure states, but also has a timeout state.

The operation of the server can be designed to be idempotent, that is, the result of multiple execution is the same as that of one execution. If you use this approach, when a timeout occurs, you can keep re-requesting until it succeeds.

1.3. Measure indicator performance

Common performance indicators are: throughput, response time.

Among them, throughput refers to the total number of requests that the system can process in a certain period of time, usually the number of read operations or write operations per second, and the response time refers to the time it takes from the time a request is sent to receiving the return result.

These two indicators are often contradictory. For systems that pursue high throughput, it is often difficult to achieve low response time, as explained as follows:

In systems without concurrency, throughput is the reciprocal of response time, for example, if the response time is 10 ms, then the throughput is 100 req/s, so high throughput means low response time.

However, in concurrent systems, a request needs to wait when it invokes the Imax O resource. The server side generally uses asynchronous waiting mode, that is, after the waiting request is blocked, it does not need to occupy CPU resources all the time. This approach can greatly improve the utilization of CPU resources. For example, in the above example, the response time of a single request is 10 ms in a non-concurrent system, but in a concurrent system, the throughput will be greater than 100 req/s. Therefore, in order to pursue high throughput, the degree of concurrency is usually improved. However, the increase of the degree of concurrency will lead to an increase in the average response time of the request, because the request cannot be processed immediately and needs to be processed concurrently with other requests, the response time will naturally increase.

Usability

Availability refers to the ability of the system to provide normal services in the face of various exceptions. It can be measured by the ratio of system available time to total time, and the availability of 4 9s means that 99.99% of the system's time is available.

Consistency

Consistency can be understood from two perspectives: from the client's point of view, whether read and write operations satisfy certain characteristics; and from the server's point of view, whether multiple copies of data are consistent.

Expandability

Refers to the ability of a system to improve performance by expanding the size of cluster servers. An ideal distributed system needs to achieve "linear scalability", that is, with the increase of cluster size, the overall performance of the system will increase linearly.

two。 Data distribution

The data of distributed storage system is distributed in multiple nodes, and the commonly used data distribution methods are hash distribution and sequential distribution.

The horizontal Sharding of the database is also a distributed storage method, and the following data distribution method is also applicable to Sharding.

2.1. Hash distribution

Hash distribution is that after calculating the hash value of the data, it is assigned to different nodes according to the hash value. For example, if there are N nodes and the primary key of the data is key, the node sequence number assigned to the data is: hash (key)% N.

There is a problem in the traditional hash distribution algorithm: when the number of nodes changes, that is, the N value changes, then almost all the data need to be redistributed, which will lead to a large number of data migration.

Consistent hash

Distributed Hash Table (DHT): for the hash space [0, 2n-1], treat the hash space as a hash ring, and configure each node on the hash ring. After each data object obtains the hash value through the hash module, it is stored on the first node in the hash ring that is greater than or equal to the hash value clockwise.

The advantage of consistent hashing is that adding or deleting nodes will only affect the adjacent nodes in the hash ring. For example, if you add node X in the following figure, you only need to restore the data object C on node X. it has no effect on nodes A, B, and D.

2.2. Sequential distribution

Hash distribution destroys the ordering of data, while sequential distribution does not.

The sequentially distributed data is divided into several consecutive parts, which are distributed to different nodes according to the ID or time of the data. For example, in the following figure, the ID range of the User table is 1 ~ 7000, and it can be divided into multiple child tables using sequential distribution, and the corresponding primary key range is 1 ~ 1000, 1001 ~ 2000.

The advantage of sequential distribution is that it can make full use of the space of each node, while hash distribution is difficult to control how much data a node stores.

But sequential distribution requires the use of a mapping table to store the mapping of data to nodes, which is usually stored using separate nodes. When the amount of data is very large, the mapping table becomes larger, so one node may not be able to store the entire mapping table. And it is very expensive for a single node to maintain the entire mapping table, and the search speed will be slower. In order to solve the above problems, a middle tier, that is, the Meta table, is introduced to share the maintenance of the mapping table.

2.3. Load balancing

There are many factors to measure the load, such as CPU, memory, disk and other resources usage, the number of read and write requests, and so on.

Distributed system storage should be able to automatically load balance, when the load of a node is high, migrate part of its data to other nodes.

Each cluster has a master control node, and the other nodes are working nodes, which are dispatched by the master node according to the global load information. The work node regularly sends heartbeat packets (Heartbeat) to send the load-related information to the master node.

A new online work node, because of its low load, if not controlled, the master control node will migrate a large amount of data to the node at the same time, resulting in the node can not work for a period of time. Therefore, the load balancing operation needs to be carried out smoothly, and the newly added nodes need a long period of time to achieve a more balanced state.

3. Distributed Theory 3.1. CAP

It is impossible for a distributed system to satisfy C:Consistency, A:Availability and P:Partition Tolerance at the same time, but only two at most.

Consistency

Consistency refers to the characteristic of whether multiple copies of data can be consistent.

Under the condition of consistency, the system can transfer from the consistency state to another consistency state after performing the data update operation.

After a data update to the system is successful, the system is considered to be highly consistent if all users can read the latest values.

Usability

Availability refers to the ability of a distributed system to provide normal services in the face of various exceptions, which can be measured by the ratio of the available time of the system to the total time. The availability of 4 9s means that 99.99% of the time is available.

Under the condition of availability, the service provided by the system is always available, and the result can always be returned in a limited time for each operation request of the user.

Partition tolerance

Network partitioning means that nodes in a distributed system are divided into multiple areas, and each area can communicate within, but not between regions.

Under the condition of partition tolerance, when a distributed system encounters any network partition failure, it still needs services that can provide consistency and availability, unless the whole network environment fails.

Tradeoff

In distributed systems, partition tolerance is essential because it is always assumed that the network is unreliable. Therefore, CAP theory is actually about making a tradeoff between usability and consistency.

Usability and consistency are often conflicting, and it is difficult to satisfy both at the same time. When synchronizing data between multiple nodes

In order to ensure consistency (CP), it is necessary to make all nodes offline unavailable and wait for synchronization to complete.

To ensure AP, data from all nodes is allowed to be read during synchronization, but the data may be inconsistent.

3.2. BASE

BASE is an acronym for basic availability (Basically Available), soft state (Soft State), and final consistency (Eventually Consistent).

BASE theory is the result of the tradeoff between consistency and usability in CAP. The core idea of the theory is that even if strong consistency can not be achieved, each application can adopt appropriate ways to achieve the final consistency according to its own business characteristics.

Basic availability

Refers to the distributed system in the event of failure, to ensure the availability of the core, allowing the loss of partial availability.

For example, when e-commerce is doing sales promotion, in order to ensure the stability of the shopping system, some consumers may be led to a downgraded page.

Soft state

It means that the data in the system is allowed to have an intermediate state, and it is considered that the intermediate state will not affect the overall availability of the system, that is, there is a delay in the process of allowing synchronization between copies of data from different nodes of the system.

Final consistency

The ultimate consistency emphasizes that all the copies of data in the system can reach a consistent state after a period of synchronization.

ACID requires strong consistency and is usually used in traditional database systems. BASE requires ultimate consistency and achieves availability at the expense of strong consistency, which is usually used in large-scale distributed systems.

In the actual distributed scenario, different business units and components have different requirements for consistency, so ACID and BASE are often used together.

4. Distributed transaction problem 4.1. Two-phase commit (2PC)

Two-phase commit (Two-phase Commit,2PC)

It is mainly used to implement distributed transactions, which means that transaction operations span multiple nodes and are required to meet the ACID characteristics of transactions.

Through the introduction of a coordinator (Coordinator) to schedule the behavior of the participants, and finally decide whether these participants actually execute the transaction.

Running process

Preparation stage

The coordinator asks the participant whether the transaction was executed successfully, and the participant sends back the transaction execution result.

Submission stage

If the transaction executes successfully on each participant, the transaction coordinator sends a notification to the participant to commit the transaction; otherwise, the coordinator sends a notification to the participant to roll back the transaction.

It is important to note that during the preparation phase, the participant executed the transaction, but has not yet committed. Commit or roll back only after receiving a notification from the coordinator during the submission phase.

problem

Synchronous blocking

All transaction participants are in a synchronous blocking state while waiting for a response from other participants and are unable to perform other operations.

Single point problem

The coordinator plays a very important role in the 2PC, and a failure will have a great impact, especially in phase 2, where all participants will wait for the state and cannot complete other operations.

Data inconsistency

In phase 2, if the coordinator sends only part of the Commit message and the network is abnormal, then only some participants receive the Commit message, that is to say, only some participants commit the transaction, which makes the system data inconsistent.

Too conservative

The failure of any node will lead to the failure of the whole transaction, and there is no perfect fault-tolerant mechanism.

Advantages and disadvantages of 2PC

Advantages: try to ensure the strong consistency of data, which is suitable for key areas with high requirements for strong consistency of data. (in fact, strong consistency cannot be guaranteed 100%) disadvantages: complex implementation, sacrificing availability, which has a great impact on performance, and is not suitable for high-concurrency and high-performance scenarios.

4.2. Compensation transaction (TCC)

The core idea of compensation transaction (TCC) is to register a corresponding confirmation and compensation (revocation) operation for each operation. It is divided into three stages:

The main purpose of the Try phase is to test the business system and reserve resources.

The main purpose of the Confirm phase is to confirm the submission of the business system. When the Try phase is executed successfully and the Confirm phase is started, the default Confirm phase will not go wrong. That is, as long as Try succeeds, Confirm will succeed.

The Cancel phase mainly cancels the business executed in the case of business execution error and needs to be rolled back, and reserves resources to be released.

For example, suppose Bob wants to transfer money to Smith, the idea would be:

First of all, in the Try phase, you need to call the remote interface to freeze the money of Smith and Bob.

In the Confirm phase, the transfer operation of the remote call is performed and the transfer is thawed successfully.

If step 2 is successful, the transfer is successful, and if the second step fails, the thaw method (Cancel) corresponding to the remote freeze API is called.

Advantages and disadvantages of TCC

Pros: compared with 2PC, the implementation and process are relatively simple, but the data consistency is also worse than 2PC.

Disadvantages: the shortcomings are still quite obvious, and it is possible to fail in the 3 steps of 2Jing. TCC is a compensation method in the application layer, so programmers need to write a lot of compensation code when implementing it. In some scenarios, some business processes may not be easy to define and handle with TCC.

4.3. Local message table (asynchronously guaranteed)

The local message table and the business data table are located in the same database, so that local transactions can be utilized to ensure that the operations on these two tables satisfy the transaction characteristics.

After one side of the distributed transaction operation completes the operation of writing business data, a message is sent to the local message table, and the local transaction can guarantee that the message will be written to the local message table.

The message in the local message table is then forwarded to a message queue (MQ) such as Kafka, and if the forwarding is successful, the message is deleted from the local message table, otherwise it will continue to be re-forwarded.

On the other side of the distributed transaction operation, a message is read from the message queue and the operation in the message is performed.

This scheme follows the BASE theory and adopts the ultimate consistency.

The local message table uses local transactions to implement distributed transactions, and message queues are used to ensure final consistency.

Advantages and disadvantages of local message table

Advantages: a very classic implementation that avoids distributed transactions and achieves ultimate consistency.

Disadvantages: message tables are coupled to the business system, and if there is no encapsulated solution, there will be a lot of chores to deal with.

4.4. MQ transaction message

There are some third-party MQ that support transactional messages, such as RocketMQ, and they support transactional messages in a way similar to the two-phase commit adopted. However, some mainstream MQ in the market do not support transaction messages, such as RabbitMQ and Kafka.

Take Ali's RocketMQ middleware as an example, the idea is roughly as follows:

If you Prepared the message, you will get the address of the message.

Perform local transactions.

Access the message through the address obtained in the first stage and modify the status.

That is, within the business method, you have to submit two requests to the message queue, one to send a message and one to confirm. If the delivery of the confirmation message fails, RocketMQ will periodically scan the transaction messages in the message cluster, and when the Prepared message is found, it will confirm to the message sender, so the producer needs to implement a check interface, and RocketMQ will decide whether to roll back or continue to send the confirmation message according to the policy set by the sender. This ensures that message delivery succeeds or fails at the same time as the local transaction.

Advantages and disadvantages of MQ transaction messages

Advantages: ultimate consistency is achieved without relying on local database transactions.

Disadvantages: difficult to implement, mainstream MQ does not support.

5. Consensus question 5.1. Paxos

Used to achieve consensus problems, that is, for the values generated by multiple nodes, the algorithm can ensure that only one value is selected.

There are three main types of nodes:

Proponent (Proposer): propose a value

Recipient (Acceptor): vote on each proposal

Learner: be informed of the result of the vote and do not participate in the voting process.

The algorithm needs to meet both safety and liveness constraints (in fact, these two basic attributes should be considered by most distributed algorithms):

Safety: make sure that the result of the resolution is correct, unambiguous, and that there will be no errors.

The resolution (value) can be finally approved only if the proposal proposed by proposers

In an implementation example, only one final decision is approved (chosen), which means that the result of the majority acceptance (accept) can become the resolution.

Liveness: ensure that the resolution process can be completed in a limited time.

Resolutions always occur, and learners can be approved (chosen) resolutions.

The basic process includes proposer putting forward a proposal, first winning the support of the majority of acceptor, and when more than half of them support it, sending the closing result to everyone for confirmation. One potential problem is that proposer fails during this process, which can be resolved by the timeout mechanism. Coincidentally, the proposer of each new round of proposals happens to fail, and the system will never be able to agree (with little chance).

Paxos ensures that the system can reach a consensus when a normal node of more than $1amp 2$ exists.

Single sponsor + multiple recipients

If only a particular node in the system is the sponsor, then consistency must be achieved (only one solution, either achieved or failed). Sponsors can be considered adopted as long as they receive votes from a majority of recipients, as there are no other proposals in the system.

However, once the sponsor fails, the system will not work.

Multiple sponsors + single recipient

Qualify a node as the recipient. In this case, it is also easy to reach a consensus. The recipient receives multiple proposals, chooses the first proposal as the resolution, and rejects the subsequent proposal.

Defects are also prone to single point of failure, including receiver failure or first sponsor node failure.

The above two situations are actually similar to the master-slave mode, which is not so reliable, but is widely used because of its simple principle.

When both sponsors and recipients are extended to multiple situations, there will be some challenges.

Multiple sponsors + multiple recipients

Since both the single sponsor and the single receiver are subject to failure, multiple sponsors and multiple receivers must be allowed. The problem suddenly became complicated.

One situation is that there is only one sponsor in the same period of time (such as a proposal cycle), which can be reduced to the case of a single sponsor. A mechanism needs to be designed to ensure the correct generation of sponsors, such as according to time, sequence, or fist guessing (come up with a number to compare). Considering the heavy workload of the distributed system, the process should be as efficient as possible, and the mechanism to meet this condition is very difficult to design.

Another situation is to allow multiple sponsors to appear in the same time frame. Then the same node may receive multiple proposals, how to distinguish them? At this time, the method of accepting only the first proposal and rejecting the subsequent proposal is also not applicable. Naturally, the proposal needs to be numbered differently. The node needs to decide which to accept according to the serial number of the proposal. For example, to accept a proposal with a larger serial number (which often means accepting a new proposal because the old sponsor is more likely to fail).

How do you assign a serial number to a proposal? One possible solution is that the proposal number intervals of each node are isolated from each other and do not conflict with each other. In order to meet the increasing demand, you can use the timestamp as the prefix field.

In addition, even if the sponsors received the votes of the majority of the recipients, they did not dare to say that they would definitely pass it. Because the system may have other proposals in the process.

5.2. Raft

Raft algorithm is a simplified implementation of Paxos algorithm.

There are three roles: leader, candidate, and follower, the basic process of which is:

Leader election-every candidate randomly puts forward an election plan after a certain period of time, and the one who gets the most votes in the most recent stage is chosen as leader.

Synchronizing log-leader will find the latest log record in the system and force all follower to refresh to this record

Note: here log does not refer to log messages, but the occurrence of various events.

A single Candidate campaign.

There are three kinds of nodes: Follower, Candidate and Leader. Leader periodically sends heartbeats to Follower. Each Follower sets a random campaign timeout, usually 150ms~300ms. If you do not receive Leader's heartbeat within this time, it will become Candidate and enter the campaign stage.

The following figure shows the initial phase of a distributed system with only Follower and no Leader. After waiting for a random campaign timeout, Follower A did not receive a heartbeat from Leader, so he entered the campaign phase.

At this point A sends a voting request to all other nodes.

Other nodes will reply to the request, and if more than half of the nodes reply, the Candidate will become Leader.

After that, Leader will periodically send heartbeats to Follower,Follower to receive heartbeats and restart the timing.

Multiple Candidate campaigns

If more than one Follower becomes Candidate and gets the same number of votes, you need to restart voting, for example, Candidate B and Candidate D both get two votes in the following figure, so you need to restart voting.

When voting starts again, because the random election timeout set by each node is different, the probability of reappearing multiple Candidate and getting the same number of votes next time is very low.

Synchronization log

Changes from the client are passed into Leader. Note that the change has not been committed, it is just written to the log.

Leader copies the changes to all Follower.

Leader waits for most of the Follower to be modified before committing the changes.

At this point, all Follower notified by Leader have them also submit changes, and the values of all nodes are agreed.

6. Distributed cache problem 6.1. Cache avalanche

Cache avalanche means that in high concurrency scenarios, due to the expiration of the original cache, the new cache does not arrive (for example, we set the cache with the same expiration time, and there is a large area of cache expiration at the same time), all the requests that should access the cache query the database, which causes great pressure on the database CPU and memory, which will seriously cause database downtime. Thus forming a series of chain reactions, resulting in the collapse of the whole system.

Solution:

Locking or queuing is used to ensure that there will not be a large number of threads reading and writing to the database at one time, so as to avoid a large number of concurrent requests falling on the underlying storage system in case of failure.

Another simple solution is to spread the cache expiration time. Do not set all cache expiration times to 5 minutes or 10 minutes. For example, we can add a random value to the original expiration time, such as 1-5 minutes randomly. In this way, the repetition rate of each cache expiration time will be reduced, and it will be difficult to trigger collective failure events.

Due to the avalanche effect caused by cache failure, all requests are placed on the database, so it is easy to reach the bottleneck of the database, resulting in the normal provision of services. Try to avoid this kind of scene.

6.2. Cache penetration

Cache traversal means that the data queried by the user does not exist in the database, and naturally there will not be in the cache. As a result, when the user makes a query, he can't find it in the cache, so he has to go to the database and query it again each time, and then return null (equivalent to two useless queries). In this way, the request bypasses the cache and looks up the database directly, which is also a frequently asked issue of cache hit rate.

When the traffic is large, when this happens, keep requesting DB, it is easy to cause the service to hang up.

Solution:

Add a step to the encapsulated cache SET and GET. If you query a KEY that does not exist, set an identification KEY; with the KEY prefix and then query the KEY later, query the identity KEY first, if the identity KEY exists, return an agreed non-false or NULL value, and then the APP does the corresponding processing, so that the cache layer will not be penetrated. Of course, the expiration time of this verification KEY should not be too long.

If the data returned by a query is empty (whether the data does not exist or a system failure), we still cache the empty result, but its expiration time is very short, usually only a few minutes.

The Bloom filter is used to hash all possible data into a large enough bitmap, and a certain non-existent data will be intercepted by this bitmap, thus avoiding the query pressure on the underlying storage system.

6.3. Cache warm-up

Cache preheating should be a common concept. I believe many partners can easily understand it. Cache preheating means loading relevant cache data directly into the cache system after the system is online. In this way, you can avoid the problem of querying the database and then caching the data when the user requests it. Users directly query pre-warmed cache data!

Solution:

Write a cache to refresh the page directly, and do it manually when you launch.

The amount of data is small and can be loaded automatically when the project starts.

Regularly refresh the cache

6.4. Cache update

In addition to the cache invalidation policy provided by the cache server (Redis has 6 policies to choose from by default), we can also customize cache elimination according to specific business needs. There are two common strategies:

Clean the expired cache regularly.

When a user request comes, determine whether the cache used in the request has expired, and if it expires, go to the underlying system to get new data and update the cache.

Both have their own advantages and disadvantages. The first disadvantage is that it is troublesome to maintain a large number of cached key, and the second disadvantage is to judge the cache invalidation every time the user requests it. The logic is relatively complex! You can weigh which scheme to use according to your own application scenario.

6.5. Cache degradation

When there is a sharp increase in traffic, when there is a problem with the service (such as slow response time or non-response), or when non-core services affect the performance of the core process, it is still necessary to ensure that the service is still available, even if it is damaging. The system can be degraded automatically according to some key data, or the switch can be configured to achieve manual degradation.

The ultimate goal of the downgrade is to ensure that core services are available, even if they are damaging. And some services cannot be downgraded (such as adding shopping carts, clearing).

This is the end of the introduction to "what is the principle of Java distributed architecture". Thank you for reading. If you want to know more about the industry, you can follow the industry information channel. The editor will update different knowledge points for you every day.

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.