TWI735519B - Distributed environment coordinated consumption queue method and device - Google Patents

Abstract

The embodiment of the present invention provides a method and device for a decentralized environment coordinated consumption (consume) queue, which relates to the field of decentralized technologies. The method includes: for a queue segment to be consumed, the status data of the queue segment obtained based on the lease agreement; judging whether there are other clients consuming the queue segment according to the status data; if it is determined that there is no When other clients are consuming the queue segment, they update the status data of the queue segment and obtain the current consumption progress of the queue segment; continue to consume the queue segment according to the current consumption progress, and add the Queue and record the new consumption progress of shards. In the present invention, when the current client A preempts the queue segment that is being consumed by other clients under load balancing, a certain client crash, etc., the consumption progress of the queue segment can be seamlessly transferred to the current client Avoid repeated consumption of part of the data, and make the consumption result more accurate.

Description

Distributed environment coordinated consumption queue method and device

The present invention relates to the field of decentralized technology, in particular to a decentralized environment-coordinated consumption (consume) queue method and a decentralized environment-coordinated consumption queue device.

With the advent of the era of cloud computing and big data, the sources of data are becoming wider and wider, faster and faster, and the quantity is increasing. For example, on the web server, various clients, sensors, etc., a large amount of data is generated in real time, recording various user access requests, monitoring data, program running status and other information. In order to better tap the value of data, there are often multiple systems that consume the data, such as user behavior analysis, monitoring alarms, and program intrusion detection performed by various real-time and offline systems on the data.

In order to reduce the degree of coupling between production and consumption of data, a queue system is usually used to cache and aggregate data produced by multiple producers, and then multiple clients as computing nodes consume data from the queue. In order to support a large amount of data, a general queue system uses multiple queue shards (such as shards or partitions) to support the level expansion of a large amount of data. In order to increase the throughput of writing, you only need to increase the number of shards (or partitions).

When a client as a computing node needs to consume data from multiple shards (or partitions), it needs to run the same program on multiple clients to collaboratively consume these data. When the client consumes data, it also needs to specify from which shard (or partition) to grab the data to complete the consumption. Usually, for a data queue consisting of N shards (or partitions), M(M The value range of is 1~N) consumer clients to collaboratively consume these shards (or partitions), and each client consumes N/M shards (or partitions) on average. In order to correctly consume the data in all shards (or partitions), these M clients need to cooperate.

In the prior art, for each client in a decentralized environment, there is the following solution to cooperate with each client to consume each queue fragment: for example, using Kinesis Client Library.

Kinesis is a real-time data queuing service provided by AWS, and Kinesis Client Library is a collaborative lib for consuming Kinesis data. Kinesis Client Library relies on DynamoDB to complete the collaboration between the clients. Although Kinesis Client Library supports new client consumption data, when performing operations such as load balancing, the preemption queue fragments are directly re-consumed on a certain client side, which will lead to repeated consumption of some data and lead to consumption results. Not precise.

In view of the above problems, the embodiments of the present invention are proposed to provide a decentralized environment coordinated consumption queue method and a corresponding distributed environment coordinated consumption queue device that overcome the above problems or at least partially solve the above problems.

In order to solve the above problems, the present invention discloses a method for coordinating consumption queues in a distributed environment, including: for a queue segment to be consumed, the status data of the queue segment is obtained based on a lease agreement; and according to the status data To determine whether there are other clients consuming the queue segment; if it is determined that no other clients are consuming the queue segment, update the status data of the queue segment and obtain the current consumption of the queue segment Progress; continue to consume the queue segment according to the current consumption progress, and record the new consumption progress of the queue segment.

The present invention also discloses a device for coordinating consumption queues in a distributed environment, including: a status data acquisition module for acquiring status data of a queue segment to be consumed based on a lease agreement; The consumption judgment module is used to judge whether other clients are consuming the queue segment based on the status data; the progress acquisition module is used to update the queue segment if it is determined that no other client is consuming the queue segment The status data of the queue segment and obtain the current consumption progress of the queue segment; the consumption module is used to continue to consume the queue segment according to the current consumption progress, and the new consumption progress of the queue segment Make a record.

The embodiments of the present invention include the following advantages: in a distributed environment, for a client, after it preempts the queue segment, the queue segment is the queue segment to be consumed for the client. The client needs to first obtain the status data of the queue segment based on the lease agreement, and then determine whether there are other clients consuming the queue segment based on the status data, and after confirming that no other client is consuming the queue segment After the queue is fragmented, the client will obtain the current consumption progress of the queue fragment and update the status data of the queue fragment, and then the client will continue to consume the queue fragment with the current consumption progress Film and record new consumption progress. Through the above-mentioned process, in the embodiment of the present invention, when the current client A preempts the queue segment that is being consumed by other clients, the consumption progress of the queue segment can be seamlessly transferred to the current client, so that the queue segment is in progress. When the slice load is balanced, or a client that is consuming the queue fragment crashes, after the queue fragment of a certain client is preempted by the current client, after the current client A preempts the above queue fragment, you can follow The consumption progress of the queue segment has been consumed, continue to consume the queue segment, avoid repeated consumption of part of the data, and make the consumption result more accurate.

110‧‧‧Step

120‧‧‧Step

130‧‧‧Step

140‧‧‧Step

310‧‧‧Status data acquisition module

320‧‧‧Consumption Judgment Module

330‧‧‧Progress Obtaining Module

340‧‧‧Consumer Module

Fig. 1 is a step flow diagram of an embodiment of a method for coordinated consumption queuing in a distributed environment according to the present invention; Fig. 2 is a structural block diagram of an embodiment of a device for coordinated consumption queuing in a distributed environment according to the present invention; Fig. 2A is an embodiment of the present invention An example of the architecture of a decentralized environment.

In order to make the above-mentioned objects, features and advantages of the present invention more obvious and easy to understand, the present invention will be further described in detail below in conjunction with the drawings and specific embodiments.

One of the core concepts of the embodiments of the present invention is that for a distributed environment, when a generator writes data, the data can be written into different queue shards (shards or partitions) through hashing or round-robin scheduling. To complete load balancing. As the client of a computing node in a distributed environment, when consuming data, it also needs to specify which queue shard to fetch data from to complete the consumption, usually for a queue composed of N queue shards In the data queue, M (the value range of M is 1~N) consumer clients will coordinately consume these queue fragments, and each client consumes N/M queue fragments on average. In the embodiment of the present invention, for the convenience of description, the queue segment is represented by a shard.

In order to correctly consume the data in all shards, the M clients need to cooperate and cooperate, and it needs to consider the following aspects: First, how to correctly choose the shard to be consumed by each client to ensure that any shard at any time has and only A consumer client; second, how to deal with the situation that one or some clients are down at the same time, and when the load balance is performed, all the data on the shard can still be processed correctly and will not be repeatedly consumed; third, when When processing pressure increases, users that need to be added for processing, how to achieve automatic load balancing, and to ensure that any data is not repeatedly consumed.

Based on the consideration of ensuring that data is not repeatedly consumed, in the embodiment of the present invention, in a distributed environment, for a client A, after it preempts the queue segment, the queue segment is to be consumed for the client A Queue segment, then the client A needs to first obtain the status data of the queue segment based on the lease agreement, and then determine whether other clients are consuming the queue segment based on the status data. After other clients consume the queue segment, the client A will obtain the current consumption progress of the queue segment and update the status data of the queue segment, and then the client A will use the current consumption The progress continues to consume the queue segment and records the new consumption progress. In this loop, when another client B preempts the queue segment of the client, it is executed according to the same logic. Through the above-mentioned process, in the embodiment of the present invention, when the current client A preempts the queue segment that is being consumed by other clients, the consumption progress of the queue segment can be seamlessly transferred to the current client, so that the queue segment is in progress. When the slice load is balanced, or a client that is consuming the queue fragment crashes, after the queue fragment of a certain client is preempted by the current client, after the current client A preempts the above queue fragment, you can follow The consumption progress of the queue segment has been consumed, continue to consume the queue segment, avoid repeated consumption of part of the data, and make the consumption result more accurate.

In order to describe the embodiments of the present invention more clearly, the present invention defines the following terms:

Shard: Queue fragment, which is the container that actually stores data in the data queue. A data queue is composed of multiple shards. Each worker needs to select a specific shard when consuming data. Of course, it can also be represented by partition, which is not limited by the present invention.

Worker: It can be understood as a client or a work process, corresponding to a computing node, and a worker can consume data in one or more shards.

Worker_name: The name of each worker, used to distinguish between different clients.

Lease: Each shard has a lock (lease), and only workers who seize the lease can consume the data in the shard.

Check Point: Consumption progress, which records the current location information of a shard that has been consumed, indicating which data has been consumed.

Example one

1, there is shown a step flow chart of an embodiment of a method for coordinating consumption queues in a distributed environment of the present invention, which may specifically include the following steps: Step 110, for a queue segment to be consumed, obtained based on a lease agreement The status data of the queue fragments; the embodiment of the present invention provides a set of lease agreements for each shard in a multi-consumer collaborative consumption queue in a distributed environment to realize the process of multi-computing nodes collaboratively consuming data in multiple shards.

According to the lease agreement, when a client consumes a shard, its consumption progress will be recorded in a persistent storage space, and the status data of the shard will be updated. The status data of the shard is used to determine whether the shard is being consumed by clients.

So for the current client worker A, if it preempts a shard, the shard is a shard to be consumed for the client, then the current client worker A needs to obtain the queue shards based on the lease agreement For the status information, go to step 120.

In a preferred embodiment of the present invention, before step 110, the method further includes:

Step 100: Determine the queue fragments to be consumed that are currently required by the client based on the lease agreement.

In practical applications, the current client first needs to determine how many shards it can consume according to the lease agreement, and then where to seize the above shards as the shards to be consumed by the current client worker A.

In a preferred embodiment of the present invention, the step 100 includes sub-step M1: sub-step M1, based on the lease agreement, determines the queue fragments to be consumed that are currently required by the user at a first time period.

In the embodiment of the present invention, the lease agreement stipulates that the current user-side worker A determines how many shards it can consume every first time period, and then where to seize the above-mentioned shards as the current user-side shard to be consumed. In this way, load balancing can be performed on the shards in a distributed environment in real time.

In a preferred embodiment of the present invention, this step 100 includes sub-steps 101-102: Sub-step 101, obtaining the total number of active clients U, the total number of queued fragments P, and the total number of queued fragments consumed by the current client Q, to calculate the number N of queue fragments that the current client needs to preempt.

In practical applications, an active client means that the client is normal and can share the processing of shards. In order to calculate the number N of shards that the current client A needs to preempt, the number N needs to ensure the load balance of the distributed environment, then it is necessary to obtain the total number of active clients U, the total number of queued shards P, and the queues that the current client has consumed The total number of shards Q, and then the number N is calculated.

In another preferred embodiment of the present invention, the sub-step 101 includes: sub-steps A11-A14: sub-step A11, obtain active client from the client instance table and queue lease table stored in the persistent storage space The total number U.

In the embodiment of the present invention, the client instance table and the queue lease period table can be constructed in the persistent storage space in advance. The persistent storage space may be a database, or, for example, a distributed cache, of course, it may also be other types of persistent storage space, which is not limited in the embodiment of the present invention. In the embodiment of the present invention, the preferred persistent storage space is a database. Of course, the database can be a mysql database or other types of database, which is not limited in the embodiment of the present invention.

Among them, the queue lease period table client_shard_lease can take the form of Table 1:

In Table 1: consume_group: consumer group field, the data type is Char(64), which is one of the primary keys of the table, and represents the consumer group name for a certain queue. In practical applications, shards are divided by consumption groups, and user terminals are also created by consumption groups. Therefore, the present invention can use consume_group to distinguish different consumption groups.

shard_id: queue identification field, the data type is Char(64), which is one of the primary keys of the table, and the identification of each shard in the queue; lease_id: lease identification field, the data type is int(20), the worker rents the shard for use Id. In the embodiment of the present invention, the atomic test and set operation is used to ensure that at any time, only one owner can modify the value of the lease, that is, the shard can be grabbed. Among them, the "test" and "set" operations are completed in an inseparable atomic operation, so that only one client can preempt the shard at the same time to ensure the correctness of data operations.

lease_owner: The renter field, the data type is Char(64), the owner of the lease that preempts the shard, that is, a worker_name; consumer_owner: the consumer field, the data type is Char(64), which is currently consuming all of the shard , That is, a certain worker_name; check_point: progress field, the data type is Text, which records the current consumption progress of the shard.

update_time: Update time field, the data type is DateTime, used to record the update time for monitoring.

Of course, in practical applications, the client_shard_lease table of the embodiment of the present invention may include shard_id, lease_owner, consumer_owner, and check_point to transfer the preempted queue fragments between the preempted client and the preempted client. The consumption progress of the robbed client. Several other fields can be preferred fields.

In the embodiment of the present invention, when each client consumes a shard, the status data of the shard in Table 1 is updated.

Among them, the user-side instance table client_worker_instance can take the form of Table 2:

consume_group: is the consumer field, the data type is Char(128), which is one of the primary keys of the table, and represents the consumer group name for a certain queue; worker_name: is the client name, and the data type is Char(64), yes One of the primary keys of the table.

create_time: The data type is DateTime, which is the time when the client is started.

Table 2 can record each client created for a consumer group.

In practical applications, after the client for a consumer group is started, the worker_name of the client will be written into the instance table of the client, and the start time will be recorded.

Specifically, in the embodiment of the present invention, the client lib and data processing logic are included in the client terminal. The lient lib executes the logic of preemption and lease renewal of the present invention, and the data processing logic is the logic for normally consuming data in the shard. Among them, the above consumption logic, such as user behavior analysis, monitoring alarm, program intrusion detection and other specific analysis and processing operations. Among them, client lib can be understood as a thread on the user side, and the execution process of client lib is roughly as follows:

1. When the worker starts, the client lib writes the worker name worker_name and the current system time into the client_worker_instance table shown in Table 2.

2. The client lib obtains the names of all shards in the queue, that is, shard_id.

3. Client lib obtains all shard_id information from client_shard_lease, including the lease_id and lease_owner corresponding to each shard;

4. Client lib judges whether the shard_id obtained in 2 appears in the client_shard_lease table; if it does not appear, add the record of the shard_id in the client_shard_lease table, set the lease_id to 0, and set the lease_owner, consumer_owner, check_point, and update_time to empty.

Of course, in actual applications, the client will create preemptive threads and leased threads. The preemption thread is used to preempt the shard, and the leased thread is used to rent the shard for consumption.

Due to practical applications, the embodiment of the present invention can obtain active client terminals from Table 1 and Table 2, so as to calculate the total number U of active client terminals.

Further, this sub-step A11 includes: sub-steps A111-A113:

Sub-step A111, obtain the number live1 of the client started in the last first time period recorded in the client instance table; in actual applications, if the execution time of a client is short after it is started, it may not be preempted by the client For any shard, there is no record of the client in the tenant field in Table 1.

Then, as in the aforementioned client_worker_instance table, since each client is started, its worker_name and start time will be written into the client_worker_instance table based on the consumer group name of the queue it processes.

In the embodiment of the present invention, t obtained by subtracting the startup time from the system time of the user terminal can be used to determine whether the t is less than the first time period T. If it is less than, the user terminal is active, and the number of active users of this type is obtained by statistics. Get live1.

Sub-step A112, obtain the number of clients live2 corresponding to the queue segment without timeout in the queue lease table; the client without timeout is the client recorded under the tenant field of the queue segment; this The embodiment of the invention also obtains the records of all shards from the client_shard_lease table, saves records of all shards whose leases have not timed out, and then extracts the worker_name under lease_owner from it. The corresponding client is active, and the statistics show that this type of activity is active The number of clients gets live2.

In sub-step A113, add live1 to live2 to obtain the total number of active clients U.

U=live1+live2, this U is the total number of active clients.

Sub-step A12: Obtain the total number of shards P from the queue system; in the embodiment of the present invention, for a consumer group, the queue system acquires the total number P of shards under the consumer group.

Sub-step A13, obtain the total number Q of the queue fragments consumed by the current client from the queue lease period table stored in the persistent storage space.

For any shard record in the client_shard_lease table, when the worker_name of lease_owner and consumer_owner are the same, and the consumer_owner is the consumer_owner of the current preempted worker process, it means that the worker_name occupies the shard. If the worker_names of lease_owner and consumer_owner are not the same, it means that the worker_name in consumer_owner is no longer allowed to occupy the shard, and the shard is preempted by other workers.

Therefore, in the embodiment of the present invention, for a worker_name, the client_shard_lease table is counted. When the lease_owner and consumer_owner are both the worker_name, the number of shards P is the number of shards occupied by the worker.

Assume that the records in Table 1 are in the form of Table 3 as follows:

For example, for the current client worker A, in the client_shard_lease table, the lease_owner and consumer_owner of shard_1 and shard_2 are all worker A, while the lease_owner of shard_3 is worker B, and the consumer_owner is worker B. Then worker A occupies shard_1 and shard_2, but not shard_3. Therefore, the number of shards occupied by worker A is 2.

Of course, the sequence between the sub-steps A11-A13 is not limited in the embodiment of the present invention.

Sub-step A14, through N=[P/U]-Q, calculate the number N of queue fragments that the current client needs to preempt.

In the embodiment of the present invention, N=[P/U]-Q is used to calculate the number of queue fragments N that the current user terminal needs to preempt.

Among them [P/U], calculate the number of shards M that the worker should occupy. Among them, [P/U] means rounding up the calculation result of P/U.

For example, before worker A performs specific preemption, worker F occupies 11 shards, worker B occupies 10 shards, worker C occupies 10 shards, worker D occupies 10 shards, and worker E occupies 11 shards. The worker A is a new worker and does not occupy a shard. And the total number of shards is 54, then the number of shards that worker A should occupy is 54/6=9, and the number of shards that worker A needs to occupy is 9-0=9. Other situations can be deduced by analogy.

Sub-step 101, from the timeout queue fragments and/or the queue fragments being consumed by other clients, preempt the N queue fragments as the queue fragments to be consumed by the current client.

In the embodiment of the present invention, because for all shards, there may be the following situations:

1. There are queue fragments that have timed out, and the number of timed out shards L is greater than or equal to N.

2. There are queued shards that have timed out, and the number of timed out shards L is less than N.

3. There is no timeout queue fragment.

For the first case, worker A can directly preempt N queue shards from the timeout shard.

For the second case, worker A can preempt the queue shards that are not larger than L from the timed out shards, and the remaining N-L shards can be preempted from other workers.

For the third case, N shards are directly preempted from other workers.

In another preferred embodiment of the present invention, the sub-step 101 includes sub-steps A21-A24:

Sub-step A21, judge whether the N is greater than 0; it is understandable that for the N obtained in the previous step, if N=0, then the worker does not need to be preempted. If N>0, then preempt.

In sub-step A22, if the N is greater than 0, determine whether the number of timeout queue fragments L is less than N;

N，則從L個超時佇列分片中隨機搶佔N個。 Sub-step A23, if the number of timeout queue fragments L is greater than or equal to N, then N queue fragments are preempted from the timeout queue fragments; in the embodiment of the present invention, if L

N, then randomly preempt N from the L timeout queue fragments.

Sub-step A24, if it is determined that the number of timed out queue fragments L is less than N, then preempt the L queue fragments from the timed out queue fragments, and from the queue fragments that are being consumed by other clients, Preempt NL queue fragments.

If L<N, the L queue fragments are preempted, and then the remaining N-L shards are preempted from the shards being consumed by other workers.

It is understandable that if L=0, worker A needs to preempt N shards from the shards that other workers are consuming.

Further, in another preferred embodiment of the present invention, the step of preempting NL queue fragments from the queue fragments being consumed by other clients in sub-step A24 includes: sub-step A241: sub-step A241, the current client seizes each NL queue fragment from the queue fragments occupied by other client terminals, and the difference between the number of the most queue fragments occupied by each client and the number of the least queue fragments occupied by each client does not exceed Specify the quantity.

In the embodiment of the present invention, the preemption thread of each worker, during preemption, ensures that the number of shards occupied by each worker meets the following conditions: the number of shards occupying the most queue and the number of shards occupying the least queue The difference does not exceed the specified amount. The specified number is, for example, 1.

Then when the current client worker A preempts the shard from other workers, the difference between the number of shards occupying the largest queue and the number of shards occupying the smallest queue does not exceed the specified number.

For example, before worker A performs specific preemption, worker B occupies 11 shards, worker C occupies 10 shards, worker D occupies 10 shards, worker E occupies 10 shards, and worker F occupies 11 shards. The total number of shards is 54 and the number of timeout queue fragments is 2. Worker A is the newly started worker.

Then you need to preempt 2 shards from worker B for worker A, 1 shard from worker C, 1 shard from worker D, 1 shard from worker E, 2 shards from worker F, and shards that have timed out. Seize 2 of them. Ensure that the number of shards occupied by each worker does not exceed 1. Other situations can be deduced by analogy.

In another preferred embodiment of the present invention, the sub-step A241 includes: sub-step A2411-sub-step A2412:

Sub-step A2411, based on the queue lease time table, each client is sorted according to the number of queue fragments occupied by it, from most to least; in the embodiment of the present invention, before worker A preempts the shard, it is first based on client_shard_lease Table, count the number of shard_id occupied by each worker_name.

For example, in the previous example, sorted by quantity, the results are shown in Table 4:

Among them, there are 2 shards that have timed out, and worker A needs to preempt 9 shards.

Sub-step A2412, each time one or more preemption queue fragments are preempted from the first K client terminals, the average of the remaining queue fragments of the first K client terminals after the first K client terminals are preempted by J The number is greater than the number of queue fragments currently occupied by the K+1 client, and the remaining number of queue fragments of each client among the first K client terminals does not differ from the average number by more than the specified number, until the preemption is successful To NL queue fragments.

It is understandable that when worker A preempts the shards of other workers, the number of shards of the next worker does not exceed the number of the previous shard.

Taking the specified number as 1, preempt the aforementioned Table 4, because it is worker A, it can preempt 2 timeout queues. Then worker A needs to grab 7 shards from other workers.

For example, if you first set K=1 and preempt from the first one, it does not satisfy that the number of remaining shards is greater than the number of shards of the second 11.

Then set K=2, because the difference between the number of remaining queue fragments of each of the first K work processes and the average number does not exceed the specified number, and the average number of remaining queue fragments after preemption is greater than the K+1 The number of shards in the queue currently occupied by a worker process is limited by two conditions, and it is not possible to preempt more than 2 shards from the first 2 jobs. Then one can be preempted from the first one, and the second one is not preempted. At this time worker F still needs to seize 6 shards, and needs to continue seizing.

At this time, Table 4 is changed to Table 5:

Then set K=3, because if the average number of remaining queue fragments is greater than the condition limit of the number of queue fragments currently occupied by the K+1 working process after preemption, you cannot preempt from the first 3 A shard. K=4 is similar.

Then set K equal to 5, then preempt 2 shards from worker C, and 1 shard from worker B, worker D, worker E, and worker F before satisfying the foregoing two conditions. Other preemption methods do not satisfy the foregoing two condition.

Based on the aforementioned queue lease schedule, in another preferred embodiment of the present invention, the sub-step 101 includes: sub-step 1011, when a queue segment is preempted, the queue is queued in the lease table, and the preempted queue The consumer field of the column fragment is modified to the current client.

For example, in Table 3, when worker A preempts shard_3 from worker B, the lease_owner is changed to worker A, and table 3 is changed to the following table 6:

It can also be understood that after the client has preempted a queue segment, it will also update the status data of the shard according to the lease agreement.

In this way, through the above steps, N shards are preempted for the client worker A. For worker A, the N shards are queue fragments to be consumed.

In practical applications, it is necessary to determine the timeout queue fragments before sub-step 101. Preferably, before sub-step 101, it further includes: sub-steps A31-A34:

Sub-step A31, every first time period, obtain the queue lease period table, and determine whether the lease ID of the tenant field under each queue segment in the queue lease period table is consistent with the map table of the local storage area The lease ID of the corresponding queue segment recorded last time is the same; in the embodiment of the present invention, the client lib of each worker creates a map table, which includes the queue segment identification field shard_id and the lease identification field Bit lease_id, last monitoring time field last_update_time.

Among them, the map table is in the memory of the computing server where the worker is located.

After the map table is created, first obtain all shard information from the client_shard_lease table and write it into the map table. For example, the related information is based on shard_id, and for each shard, shard_id, lease_id and current system time are recorded one by one. Then you can enter step B11 to perform the timeout monitoring process.

In the embodiment of the present invention, the client lib of a worker obtains each shard_id and the lease_id of the shard_id from the client_shard_lease table of the database every first time period T.

Then, match the shard_id+lease_id in the client_shard_lease table with the shard_id+lease_id in the map table.

If the shard_id does not match, it means that the shard is new, and there is no corresponding record in the map. Add the shard_id+lease_id to the map table and record the current system time in the corresponding last_update_time field.

In addition, the present invention is for newly emerging shards. If the lease_id of the shard_id is not 0, it means that the shard is preempted or occupied by a worker. You can record the shard_id and lease_id in the map table, and record the current time in last_update_time. If the lease_id of the shard_id is 0, it means that no worker preempts or occupies the shard. Record the lease_id=0 of the shard_id in the map table and set the last_update_time to 0. It can be directly considered that the shard has timed out.

If the shard_id matches but the lease_id does not match, go to step A32.

If shard_id+lease_id matches, go to step A33.

Sub-step A32, if the lease ID of a queue segment in the queue lease table is different from the lease ID of the corresponding queue segment recorded last time in the map table, update the lease of the queue segment in the map table ID is the lease ID in the queue lease period table, and the last monitoring time field of the queue segment in the updated map table is the current system time; in the embodiment of the present invention, due to the renewal thread of each worker Renew the lease for the worker with the same lease_owner and consumer_owner shards every second time period. The second time period is less than the first time period, and the second time period is, for example, T1/2. When renewing a shard, the client_shard_lease table lease_id will be changed to lease_id+1. Then if the worker renews the lease of the shard, the lease_id will change, indicating that the worker is consuming the shard normally. If lease_owner and consumer_owner are the same and different, the worker is not allowed to renew the shard.

Then, the current user-side workerA in the embodiment of the present invention, for each shard, determines that the map table shard_id matches the shard_id of the queue lease table, but the lease_id does not match, then it is determined that the responding shard has not timed out, in the map table Under the last_update_time field after the shard_id, update the system time recorded in its last_update_time to the current system time.

Sub-step A33, if the lease ID of a queue segment in the queue lease period table is the same as the lease ID of the corresponding queue segment recorded last time in the map table, then the system time under the last monitoring time field is maintained , And determine whether the current system time minus the last monitored system time is greater than the first time period; as mentioned above, if there is no shard for renewing the worker's lease, the lease_id does not change, indicating that the worker does not consume the shard normally.

Then, worker A maintains the system time under the last_update_time column in the map table unchanged, and then determines whether the current system time sys_time minus last_update_time is greater than the first time period T.

In sub-step A34, if the current system time minus the last monitored system time is greater than the first time period, it is determined that the corresponding queue segment has timed out.

If the current system time sys_time minus last_update_time is greater than the first time period T, it means that the shard has timed out. The shard will be regarded as the preempted object. You can go to step A11.

Then after the current client has preempted each queue segment, the step 110 includes sub-step 111: sub-step 111, for a queue segment to be consumed, from the queue lease table stored in the persistent storage space Get the status data of the queue segment.

In the embodiment of the present invention, since each queue segment is consumed by any client, or when the queue segment is newly generated, its status data is written into the aforementioned queue lease table, then the current user After the end worker A has preempted the N queue shards, it can obtain the status data of the queue shard from the queue lease table. Go to step 120.

Step 120: According to the status data, determine whether there are other clients consuming the queue segment; in the embodiment of the present invention, for each of the N shards preempted by the current client worker A, the result is obtained by the foregoing steps. With the status data of each shard, you can analyze whether these shards have other workers using the shard based on the status data. If the shard is not used, go to step 130. If it is still in use, continue to obtain the new status data of the shard and make the above judgment.

Preferably, based on the aforementioned queue rental period table, the status data includes the value of the consumer field under the queue segment in the queue rental period table, the value of the renter field, and the consumption time of the update time field; The consumption time is obtained by modifying the value of the update time field after the current client preempts the queue segment.

Based on the aforementioned queue lease table, for each shard preempted by worker A, the status data of the shard, such as lease_owner, consumer_owner, and update_time, are obtained from the queue lease table according to the shard_id.

The value of the update_time is that when worker A preempts the shard, the consumption time value is sent to the database based on the shard_id generation time update request, which is obtained from the update_time value of the shard_id in the database update queue lease table.

Of course, in practical applications, it can also be combined with the consumer group to obtain the status data of the shard, and send an update request to the database to update the status data in the queue lease.

Further, this step 120 includes sub-steps 121-123: sub-step 121, judging whether the value of the consumer field of the queue segment is the current client; in practical applications, the current client worker A is for The shard to be consumed, according to the corresponding shard_id, sends a status data acquisition request to the database to obtain the corresponding status data, and then the value of lease_owner can be extracted from the status data. If the value is worker A, it means the user The terminal successfully seizes the shard. Proceed to sub-step 121.

Sub-step 122, if the value of the consumer field of the queue segment is the current client, determine whether the preemption time of the current client is greater than the consumption time; sub-step 123, if the preemption time of the current client is greater than the consumption time, It is determined that no other client is consuming the queue segment.

In actual use, after the current client worker A preempts the shard, it will update the queue lease table in the database, and the update_time value of the corresponding shard_id is the consumption time.

The preemption time can be recorded locally on worker A on the current user side.

In this way, it can be judged whether the preemption time of the current user terminal is greater than the consumption time. If the preemption time of the current client is greater than the consumption time, it means that no other client is consuming the queue segment, and the consumption progress of the queue segment is not updated by other client terminals. If the preemption time of the current client is not greater than the consumption time, it means that there may be other client consuming the queue segment, and the consumption progress of other clients may not have been updated to the queue lease period table.

Preferably, the preemption time is the system time of the current user end, and the consumption time is the sum of the system time when the user end preempts the queue segment and the first cycle time.

In the embodiment of the present invention, for the convenience of calculation, when the current client worker A preempts the shard, when the lease_owner of the queue lease table is updated to worker A, the current system time of worker A is added to the first time period to obtain The consumption time is updated to the update_time under the shard_id in the queue rental period table.

Then after obtaining the status data of shard_id in step 110, sub-step 122 extracts the consumption time from update_time, and then directly compares the current system time of worker A with the consumption time. If the current system time exceeds the consumption time, it is determined that there are no other users The end is consuming the queue fragment.

In practical applications, because each worker is in a different computing server, and because the computing server has a different system, the system time of each worker may also be different. Then, the present invention records its own system time for each worker, and unifies the first time period T. Taking the aforementioned worker A as an example, through the aforementioned steps, it is easy to obtain the system time that the system should reach after the preemption of worker A begins and after T has passed, and the system time that should arrive is the aforementioned consumption time. When the system time of worker A exceeds the consumption time of shard_3, it is determined that no other users are consuming the queue segment. For example, the system time when worker A preempts shard_3 is 12:00:00:000, and the first time period T is 50ms, then the consumption time is 12:00:00:050. At this time, worker A can start every T/2 from The database obtains the status data of shard_3 of the queued lease period table, and the status data includes update_time. Then, if the system time reaches 12:00:00:051, it is greater than the consumption time, which means that other users write the latest consumption progress into the check_point of shard_3 in the first time period T, and you can go to step 130 . If the system time is not greater than the consumption time, it means that there may be other clients consuming the queue segment. In this way, the difference of the system time of each client is avoided, which causes the current client to make an error in determining that no other client is consuming the queue segment.

Step 130: If it is determined that no other client is consuming the queue segment, update the status data of the queue segment, and obtain the current consumption progress of the queue segment; in the embodiment of the present invention, if time is preempted If it is longer than the consumption time, it is determined that no other client is consuming the queue segment, and can send an update request to the database, update the status data of the queue segment in the queue lease table, and rent from the queue of the database at the same time In the timetable, obtain the consumption progress of the progress field of the queue segment.

As mentioned above, if the preemption time is not greater than the consumption time, there may be other clients consuming the shard, and step 110 may be continued to obtain status data for judgment.

Based on the aforementioned queue lease schedule, in another preferred embodiment of the present invention, step 130 includes sub-steps 131-132: sub-step 131, lease the queue segment, and store the queue lease in the persistent storage space The consumer field under the queue shard in the schedule is updated to the current client; sub-step 132, from the queue lease table stored in the persistent storage space, obtain the progress field of the queue shard Consumption progress.

In practical applications, for any shard to be consumed preempted by worker A, when it is judged that the preemption time is greater than the consumption time, it means that the consumption progress of the shard recorded in the queue lease table is the latest shard, and there are no other workers. The shard is being consumed.

Therefore, the renting thread of worker A can rent the shard, and then send a request to the database to update the consumer_owner of the corresponding shard_id of the queued lease table in the database to worker A. Of course, in actual applications, the lease_id under the shard_id in the queue lease table is also updated to lease_id+1, which means that worker A occupies the queue shard.

At the same time, the record consumption progress can be obtained from the update_time of the corresponding shard_id in the queue lease period table.

If the preemption time is greater than the consumption time, you can only update the lease_id under the shard_id in the queue lease table to lease_id+1, and prevent other clients from preempting the shard.

In this way, in this way, for shards that have been consumed by other users, the final consumption progress can be seamlessly transferred to the current user, and there is no problem of repeated consumption.

In another preferred embodiment of the present invention, after sub-step 132, it further includes sub-steps 133-134: sub-step 133, judging in the queue lease table, the preemptor field under the queue segment and Whether the consumer field is the current client; sub-step 134, if the preemptor field and the consumer field under the queue segment are the current client, then the queue segment is renewed for the current client.

Understandably, the lease renewal thread of each worker determines whether the worker occupies each shard, and whether the consumer_onwer and lease_owner under each shard_id in the client_shard_lease table are the same. If they are the same, the shard is renewed for the worker. If they are different, the shard will not be renewed for the worker.

In practical applications, each worker's lease renewal thread executes the process of steps A41 to A42 every T1/2.

During the consumption process of each work process, it also includes: step A31, for each work process, when consuming the queue fragments occupied by the work process, every first time period, in the memory where the work process is located , Generate a new lease ID based on the last lease ID; in the embodiment of the present invention, during the process of consuming any shard for each worker, every first time period T.

Step A32, determine whether the last lease ID recorded in the memory of the work process is the same as the lease ID of the corresponding queue segment in the queue lease period table; Step A33, if they are the same, in the update queue lease period table The lease identification field is the new lease ID; step A34, if they are different, refuse to update the lease identification field in the queue lease period table to the new lease ID.

In the embodiment of the present invention, the clientlib corresponding to each worker uses the lease_id of each shard in the current memory to update the lease_id of the shard in the database. The client lib adds the lease_id of the shard to the memory every T time, and compares the lease_id of the last time with the lease_id of the shard in the client_shard_lease table. If they are the same, the lease_id of the client_shard_lease table is allowed to be updated to lease_id+1, if different The above update is not allowed. Due to the atomicity of the operation, it can be guaranteed that only one worker can update successfully at any time, that is, only one worker can grab the lease of the shard.

Step 140: Continue to consume the queue segment according to the current consumption progress, and record the new consumption progress of the queue segment.

When the consumption progress of the above-mentioned shard is obtained, worker A can continue to consume the shard from the consumption progress of the shard. At the same time, in the process of consuming the shard, the new consumption progress is updated to the queue lease period table of the database.

Preferably, the step of recording the new consumption progress of the queue fragments in step 140 includes: sub-steps A41-A42: sub-step A41, every second time period, determine whether the queue is in the queue Whether the tenant field of the queue segment is the current client; sub-step A42, if the tenant field of the queue segment is the current client, update the consumption progress of the queue segment by the current client Go to the progress column of the queue segment in the queue lease table.

For example, for the aforementioned Table 6, after the current user-side worker A determines that it can consume shard_3, it reads the consumption progress 2/5 from the check_point column under shard_3 in Table 6, and then starts reading from 2/5 of shard_3 Take data for consumption.

It should be noted that if the check_point information of the shard in the database is empty, the data will be read from the beginning or end of the shard according to the pre-configuration. It is preferable to configure to read data from begin to avoid data omission.

At the same time, in the process of consuming shard_3 by worker A, the new consumption progress is updated to the check_point field of shard_3 in the queue lease table of the database.

It should be noted that the client lib of the embodiment of the present invention provides a check point interface, through which check point information is saved, ensuring that the worker consumes correctly when the worker fails over or the shard is preempted by different workers Data in the shard. The use of check point consists of the following two parts:

(1) The part used for worker initialization check point:

(1.1) When a shard is determined to be consumed, the client automatically loads the check point information from the database. That is, the relevant method of step 120 is executed.

If the check_point information of the shard in the database is empty, the data will be read from the beginning or end of the shard according to the pre-configuration.

(2) The part used for worker's persistent check point, that is, the related method of executing step 140:

(2.1) Clientlib provides an interface to perform check point operations saveCheckPoint (Bool persistent). Among them, the parameter persistent is used to control whether it is necessary to understand the persistence to an external database. If persistent is true, it will be persisted to the database immediately, otherwise, it will be persisted at regular intervals. In the embodiment of the present invention, the value of persistent is controlled by the client lib.

(2.11) If persistent is true, the consumption progress will be persisted to the check_point column of the corresponding shard in the database. In the embodiment of the present invention, after the renter field in the queue lease period table of the queue fragment is modified to a preemptive work process, at the end of the current first time period T, the persistent Modify it to true to immediately persist the consumption progress to the check_point column of the corresponding shard in the database.

(2.12) If persistent is not true, the consumption progress will be saved in the memory. At the same time, the background will periodically persist the consumption progress that has not been persisted for a long time longer than the specified time and persist to the check_point field of the corresponding shard in the database Down.

In the embodiment of the present invention, when the lease_owner and consumer_owner of the queue shards are the same, persistent can be kept not true, so that the consumption progress that has not been persisted for a long time longer than a specified time period can be persisted to the corresponding database. Under the check_point column of the shard. It should be regularly as 2T.

(2.2) For a shard, under normal circumstances, only when consumer_onwer and worker_name in the database are the same, clientlib will persist the consumption progress to the check_point column of the corresponding shard in the database.

(2.3) When a shard is preempted or crashed by other workers, clientlib calls the shutdown interface to notify the upper application and persist the consumption progress to the check_point field of the corresponding shard in the database.

After the worker's shard is preempted, since the consumer_owner information of the shard in the database has not been updated in the first time period T, the preempted worker can correctly persist the check point information of each shard it consumes.

In the embodiment of the present invention, in a distributed environment, for a client, after it preempts the queue segment, the queue segment is the queue segment to be consumed for the client, and the client needs to first The status data of the queue segment obtained based on the lease agreement, and then based on the status data, determine whether other clients are consuming the queue segment. After confirming that no other client is consuming the queue segment, the The client will obtain the current consumption progress of the queue segment, and update the status data of the queue segment, and then the client will continue to consume the queue segment with the current consumption progress and record the new consumption progress. Through the above process, in the embodiment of the present invention, when the current client seizes the queue fragment that is being consumed by other clients, the consumption progress of the queue fragment can be seamlessly transferred to the current client, so that the queue fragmentation is in progress. During load balancing, or a client that is consuming a queue fragment crashes, after the queue fragment of a certain client is preempted by the current client, the current client can continue to consume the queue according to the consumption progress. To avoid repeated consumption of part of the data, make the consumption result more accurate. In addition, consider the aforementioned three aspects: first, how each computing server correctly selects the shard to consume to ensure that any shard has and only one consuming computing server at any time; second, how to deal with one or some of them at the same time When the computing server is down, the data on all shards can still be processed correctly when re-load balancing is performed, and will not be repeatedly consumed; third, when the processing pressure increases, a new computing server needs to be added. How to achieve automatic load balancing, and to ensure that any data is not repeatedly consumed. In the prior art, there is also a solution using Kafka Client. Kafka is a distributed message queue of specially processed logs that provides a publish-subscribe function of messages. Kakfa provides a set of advanced consumption API (Application Programming Interface, application programming development interface) to complete the synchronization between multiple work processes. The API relies on the kafka backend zookeeper system. Due to the reliance on the zookeeper system, after the worker process A updates the data of the key of a certain queue segment A, it cannot be transmitted back to the zookeeper in time. Then if the load balance is performed, the other worker processes get the key of the queue segment A Still update the previous information. Therefore, in the Kafka system, the zookeeper data seen by different worker processes may be of different versions, causing the load balancing operation to fail during failover.

In the above process of the embodiment of the present invention, the work process of the present invention directly faces the queue shards and processes the queue shards, which also avoids the problem that different work processes see different data versions when the Kafka system re-balances the load. .

Furthermore, in the prior art, Kinesis Client Library can only load balance one queue segment at a time during load balancing. For example, worker process A is consuming 100 queue segments, and for worker process B, When load balancing, it is necessary to preempt the queue shards from the work process A. Each time the load is balanced, only 1 queue shard can be preempted from the work process A. If in fact, A wants to preempt 30 queue shards from B , To trigger 30 load balancing operations, resulting in long load balancing time-consuming and unable to reach a stable state quickly

However, in the present invention, in a load balancing process, the N queue segments that should be preempted by the work process can be preempted at one time. The embodiment of the present invention can quickly select the method of preempting shards in batches, so that each worker can quickly reach stability. State, and balance the number of shards consumed by each worker.

Especially based on the client instance table and the queue lease period table stored in the persistent storage space, the embodiment of the present invention queues the lease period table, and then lists the consumer_owner, consumer_owner, and consumer_owner fields in the queue lease period table. The renter field lease_owner and progress field check_point respectively record the consumption progress parameters of the preemptor, occupant, and occupant of each queue segment. Then when a worker process preempts a queue segment occupied by another worker process, the consumption will be consumed through the consumer_owner field, lease_owner field and progress field check_point of the queue segment in the queue lease table. The progress parameter is passed to the preempted work process, so that when a new work process is added, the computing server where a work process is located is down, or a work process fails over, the original queue fragments occupied when the load balance is restarted When preempted by other work processes, the preempted queue segment can continue to consume according to its consumed position, avoiding repeated consumption of part of the data, and making the consumption result more accurate.

In order to describe the present invention more clearly, this embodiment is based on the queue lease period table and the client instance table, and describes the situation where a certain client worker A is newly started:

Step 201: Create a client instance table and a queue lease period table in the database in advance.

The client_worker_instance table and the client_shard_lease table are created in a principle similar to the first embodiment.

Then steps 202-227 are executed for each newly started client terminal.

Step 202: Start the client, and write the client name into the client instance table.

When worker A moves, the client lib thread will be started, and the client lib thread on the client side will write worker_name and the current system time into the client_worker_instance table.

Step 203: Obtain the queue fragment identifiers of all the queue fragments from the queue system.

Get the shard_id of all shards in the queue system by the client lib thread.

Step 204: Obtain the lease ID field and the renter field of all the queue segments from the queue lease period table in the database.

According to the shard_id, the client lib thread obtains the lease_id and lease_owner of all shards from the client_shard_lease table

Step 205: Determine whether all the queue fragments in the queue system do not appear in the queue lease period table; if there are queue fragments that do not appear in the queue lease period table, then the queue fragments are listed Relevant information is written into the queue lease period table of the database.

When the client lib thread determines that a new shard_id does not appear in the client_shard_lease table, it writes the record corresponding to the shard_id to the client_shard_lease table, sets the corresponding lease_id to 0, and sets other fields to be empty.

Then Client lib threads create preemption threads and lease renewal threads. The preemption thread executes the preemption process of the queue segment, and the renewal thread executes the lease process of the queue segment.

Step 206, obtain the number live1 of the client started in the last first time period recorded in the client instance table.

Step 207: Obtain the number live2 of the client corresponding to the queue segment without timeout in the queue lease period table; the client without timeout is the client recorded under the tenant field of the queue segment.

Step 208: Add live1 to live2 to obtain the total number of active users U.

Step 209: Obtain the total number of queue fragments P from the queue system;

Step 210: Obtain the total number Q of queue fragments consumed by the current client from the queue lease period table;

Step 211, by N=[P/U]-Q, calculate the number N of queue fragments that the current client needs to preempt.

Step 212: Obtain the queue lease period table every first time period, and determine whether the lease ID of the tenant field under each queue segment in the queue lease period table is consistent with the map table of the local storage area The lease ID of the corresponding queue shards recorded last time is the same; the preemption thread on the client side obtains the lease_id of each shard in the client_shard_lease table in the database every first time period, and compares it with the corresponding shard last recorded in the map table. The lease_id is compared to determine whether they are the same.

Step 213: If the lease ID of a queue segment in the queue lease period table is different from the lease ID of the corresponding queue segment recorded last time in the map table, update the lease ID of the queue segment in the map table. To list the lease ID in the lease period table, and update the last monitoring time field of the queue segment in the map table to the current system time; if the lease_id of a shard in the client_shard_lease table is the same as the last record in the map table If the lease_id of the corresponding shard is different, update the lease_id of the shard in the map table to the lease_id in the client_shard_lease table, and update the monitoring field of the shard in the map table to the current system time

Step 214: If the lease ID of a queue segment in the queue lease period table is the same as the lease ID of the corresponding queue segment recorded last time in the map table, the system time under the last monitoring time field is maintained. And determine whether the current system time minus the last monitored system time is greater than the first time period; if the lease_id of a shard in the client_shard_lease table is the same as the lease_id of the corresponding shard recorded last time in the map table, the last monitoring field will be maintained And determine whether the current system time minus the last monitored system time is greater than the first time period T.

Step 215: If the current system time minus the last monitored system time is greater than the first time period, it is determined that the corresponding queue segment has timed out.

Step 216: Determine whether the N is greater than 0;

Step 217: If the N is greater than 0, determine whether the number of timed out queue fragments L is less than N;

Step 218: If the number of timeout queue fragments L is greater than or equal to N, preempt N queue fragments from the timeout queue fragments;

Step 219: If it is determined that the number of timed out queue fragments L is less than N, then preempt the L queue fragments from the timed out queue fragments, and preempt the queue fragments that are being consumed by other clients NL queue fragments.

Among them, for the preempted queue segment, when a queue segment is preempted, the consumer field of the preempted queue segment in the queue lease table of the database is modified to the current client, And modify the update time field in the queue lease period table to consumption time; the consumption time is the sum of the system time and the first cycle time when the client seizes the queue segment.

For a preempted shard, modify the lease_owner of the preempted shard in the client_shard_lease table to the worker A, and record the consumption time under update_time. And put the preempted shard into the lease renewal thread.

For shards that are preempted from other clients, because the looted client judges that the lease ID in the local map is different from that in the queue lease table, at this time the consumer field is the same as the robbed client, it can be The robbed client time updates the consumption progress, and since the renewal period is T/2, the robbed client cannot renew the shard, and the robbed client cannot continue to process the shard if the lease cannot be renewed. This ensures that the consumption progress of the queue fragments is seamlessly transmitted to each client.

The above steps 212 to 219 are the aforementioned preemptive thread execution, and the preemptive thread is executed once every 2*T. When the user-side worker A newly joins, all its queue fragments are obtained by the above steps. In each subsequent execution, if it is determined through the above steps that the client terminal wants to load balance the queue fragments of other client terminals, the preemption process is executed.

Step 220: For the preempted queue segment, obtain the value of the consumption field of the queue segment from the queue lease table, and determine whether the value of the consumer field of the queue segment is The current client.

Step 221: If the value of the consumer field of the queue segment is the current client, it is determined whether the preemption time of the current client is greater than the consumption time.

In this step, it is judged whether the system time of the preempted client is greater than the consumption time under update_time.

Step 222: If the current client's preemption time is greater than the consumption time, the queue segment is leased, and the consumer field under the queue segment in the storage queue lease table in the persistent storage space is updated to the current user terminal.

Determine whether the system time of the preempted client is greater than the consumption time under update_time; if the system time of the client exceeds the consumption time, modify the consumer_owner of the preempted shard to the client of worker A, and change the preempted shard’s The lease_id is modified to lease_id+1, so that worker A leases the shard, and it means that worker A successfully preempts the shard.

Step 223: For the queue segment occupied by the client, obtain the preemptor field value and the tenant field value from the queue lease period table of the database, and determine the preemptor field and the tenant field value under the queue segment Whether the consumer field is the current client;

Step 224: If the preemptor field and the consumer field under the queue segment are the current client, then the queue segment is renewed for the current client.

The above steps 220-224 are the execution steps of the lease renewal thread. The lease renewal is now every T/2. For the shard occupied by the current client, when it is judged that the lease_owner and the consumer_owner in the client_shard_lease table are the same, the lease renewal is performed.

Step 225: For the queue segment that is successfully preempted by the client, the consumption progress is obtained from the progress field under the queue segment in the queue lease table of the database.

The client extracts the consumption progress from the check_point column of the client_shard_lease table. When the consumption progress is empty, the consumption progress of the shard is to start consumption from the head.

In step 226, the client continues to consume the queue segment from the consumption progress position of the queue segment.

Step 227: During the process of consuming the queue segment, the user terminal updates the consumption progress to the progress field under the queue segment in the database.

When the worker consumes the shard, the client lib thread calls the checkpoint interface and updates the consumption progress to the check_point in the client_shard_lease table every T time; or after the shard lease_owner is preempted, the checkpoint is called when the lease renewal period arrives. The shutdown interface of, notifies the upper application, and persists the check point information of the current successful consumption to the check_point in the client_shard_lease table; or, when worker A is down, calls the shutdown interface of checkpoint to notify the upper application to check the current successful consumption check The point information is persisted under check_point in the client_shard_lease table.

Of course, after the user terminal performs the first preemption queue segmentation, it can perform step 203 and subsequent steps.

It should be noted that for the method embodiments, for the sake of simple description, they are all expressed as a series of action combinations, but those with ordinary knowledge in the technical field should know that the embodiments of the present invention are not subject to the described sequence of actions. The limitation is because according to the embodiment of the present invention, some steps can be performed in other order or simultaneously. Secondly, those with ordinary knowledge in the technical field should also know that the embodiments described in the specification are all preferred embodiments, and the actions involved are not necessarily required by the embodiments of the present invention.

Example two

Referring to FIG. 2, there is shown a structural block diagram of an embodiment of a system for coordinating consumption queues in a distributed environment of the present invention, which may specifically include the following modules: a status data acquisition module 310 for a queue to be consumed The sharding is based on the status data of the queue shard obtained based on the lease agreement; the consumption judgment module 320 is used to determine whether other clients are consuming the queue shard according to the status data; the progress obtaining module 330, If it is determined that no other client is consuming the queue segment, update the status data of the queue segment, and obtain the current consumption progress of the queue segment; the consumption module 340 is used to according to the current consumption The progress continues to consume the queue segment, and the new consumption progress of the queue segment is recorded.

In a preferred embodiment of the present invention, before the status data acquisition module 310, it further includes: a queue segment determination module, which is used to determine the queue segment to be consumed based on the current client demand based on the lease agreement.

In a preferred embodiment of the present invention, the queue segment determination module includes: a demand quantity determination sub-module for obtaining the total number of active clients U, the total number of queue segments P, and the current client has consumed The total number of queue shards Q is used to calculate the number of shards N that the current client needs to preempt; the preemption sub-module is used for queue shards that have timed out and/or queue shards that other clients are consuming In, preempt the N queue fragments as the queue fragments to be consumed on the current user end.

In a preferred embodiment of the present invention, the demand quantity determination sub-module includes: a client total number obtaining sub-module, which is used to obtain active data from the client instance table and the queue lease period table stored in the persistent storage space. The total number of client-sides U; the total number of shards in the overall queue obtains the sub-module, which is used to obtain the total number of shards P from the queue system; the total number of single-queue shards obtains the sub-module, which is used from Get the total number Q of queue shards consumed by the current client from the queue lease table stored in the persistent storage space; the demand quantity calculation sub-module is used to calculate the current user through N=[P/U]-Q The number of shards in the queue that the end needs to preempt is N.

In a preferred embodiment of the present invention, the client total number acquisition sub-module includes: a newly created client number acquisition sub-module, which is used to obtain the latest client information recorded in the client instance table during the first time period. Quantity live1; the sub-module for obtaining the number of clients that have not timed out is used to obtain the number of clients that do not have a timeout queued segment in the queue lease table, live2; the client without a timeout is the queued segment The client recorded under the tenant field; the client quantity accumulation sub-module is used to add live1 to live2 to get the total active client quantity U.

In a preferred embodiment of the present invention, the preemption sub-module includes: a first judging sub-module for judging whether the N is greater than 0; a second judging sub-module for judging if the N is greater than 0 Whether the number of timeout queue fragments L is less than N; full timeout preemption sub-module, used to preempt N from the timeout queue fragments if the number of timeout queue fragments L is greater than or equal to N Queue shards; mixed preemption sub-module, used to determine that the number of timed out queue shards L is less than N, then preempt L queue shards from the timed out queue shards, and correct them from other clients Among the queue fragments for consumption, NL queue fragments are preempted.

In a preferred embodiment of the present invention, the hybrid preemption module includes: a first hybrid preemption sub-module for the current client to preempt each queue segment of NL from the queue segments occupied by other client terminals, and The difference between the number of shards that each client occupies the most and the number of shards that occupies the least does not exceed the specified number.

In a preferred embodiment of the present invention, the first hybrid preemption sub-module includes: a sorting sub-module, which is used for sorting each client according to the number of queue shards occupied by it based on the queue lease table, from Sort from more to less; the second hybrid preemption module is used to preempt one or more preemption queue fragments from the first K client terminals each time, so that after the first K client terminals are preempted by J, the first K The average number of remaining queue fragments on the client side is greater than the number of queue fragments currently occupied by the K+1th client side, and the number of remaining queue fragments on each client side among the first K client terminals is equal to this The average number does not differ by more than the specified number, until NL queue fragments are successfully preempted.

In a preferred embodiment of the present invention, the preemption sub-module includes: a consumer field modification sub-module, which is used to queue the lease term in the preempted queue when a queue segment is preempted The consumer field of the fragment is modified to the current client.

In a preferred embodiment of the present invention, before the preemption submodule, it further includes: a map table judging submodule, which is used to obtain the queue lease period table every first time period and determine the queue lease period Whether the lease ID of the tenant field under each queue segment in the table is the same as the lease ID of the corresponding queue segment recorded last time in the map table of the local storage area; the map monitoring time update submodule is used for If the lease ID of a queue segment in the queue lease table is different from the lease ID of the corresponding queue segment recorded last time in the map table, update the lease ID of the queue segment in the map table to the queue The lease ID in the lease table, and the last monitoring time field of the queue segment in the updated map table is the current system time; the map monitoring time maintenance sub-module is used if a queue in the queue is listed The lease ID of the column segment is the same as the lease ID of the corresponding queue segment recorded last time in the map table. The system time under the last monitoring time field is maintained, and the current system time minus the last monitored system is determined Whether the time is greater than the first time period; the timeout judging sub-module is used to determine if the current system time minus the last monitored system time is greater than the first time period, determine the corresponding queue segment timeout.

In a preferred embodiment of the present invention, the queue fragmentation determining module includes: a first queue fragmentation determining sub-module, which is used to determine the waiting time of the current user end demand every first time period based on the lease agreement. The consumption queue fragments.

In a preferred embodiment of the present invention, the status data acquisition module 310 includes: a first status data acquisition sub-module, which is used for a queue segment to be consumed from the queue stored in the persistent storage space Obtain the status data of the queue segment in the lease table.

In a preferred embodiment of the present invention, the status data includes the value of the consumer field under the queue segment in the queue rental period table, the value of the renter field, and the consumption time of the update time field; the consumption The time is obtained by modifying the value of the update time field after the current client preempts the queue segment; then, the consumption judgment module 320 includes: a consumer field judgment sub-module for judging the queue segment Whether the value of the consumer field is the current client; the preemption time judging sub-module is used to determine the preemption time of the current client if the value of the consumer field of the queue segment is the current client Whether it is greater than the consumption time; the determining sub-module is used for determining that no other client is consuming the queue segment if the current client's preemption time is greater than the consumption time.

In a preferred embodiment of the present invention, the preemption time is the system time of the current client, and the consumption time is the sum of the system time when the client preempts the queue segment and the first cycle time.

In a preferred embodiment of the present invention, the progress obtaining module 330 includes: a lease sub-module for renting the queue segment, and storing the queue segment in the queue lease schedule in the persistent storage space. The consumer field under the slice is updated to the current client; the progress field reading module is used to obtain the progress field of the queue segment from the queue lease table stored in the persistent storage space Consumption progress.

In a preferred embodiment of the present invention, the consumption module 320 includes: a renter judging sub-module for judging the renter column of the queue segment in the queue lease period every second time period Whether the bit is the current client; the consumption progress update sub-module is used to update the consumption progress of the queue by the current client to the queue if the tenant field of the queue segment is the current client List under the progress column of the queue segment in the lease period table.

In a preferred embodiment of the present invention, after the submodule is leased, it further includes: a lease renewal judging submodule for judging the preemptor field and the sub-module under the queue segment in the queue lease period table Whether the consumer field is the current client; the first renewal sub-module is used to renew the lease for the current client if the preemptor field and the consumer field under the queue segment are the current client The queue fragment.

With reference to FIG. 2A, it shows an example of the distributed environment of the present invention, in which each computing node is a user terminal. The scheduling server can allocate queue fragments. Each client terminal includes: a status data acquisition module 310, a consumption judgment module 320, a progress acquisition module 330, and a consumption module 340. Of course, each client can also include a corresponding preferred module.

In a distributed environment, for a client, after it preempts the queue segment, the queue segment is the queue segment to be consumed for the client, then the client needs to obtain it based on the lease agreement first According to the status data of the queue segment, determine whether there is another client consuming the queue segment. After confirming that no other client is consuming the queue segment, the client will Obtain the current consumption progress of the queue segment, and update the status data of the queue segment, and then the client continues to consume the queue segment with the current consumption progress and records the new consumption progress. Through the above process, in the embodiment of the present invention, when the current client seizes the queue fragment that is being consumed by other clients, the consumption progress of the queue fragment can be seamlessly transferred to the current client, so that the queue fragmentation is in progress. During load balancing, or a client that is consuming a queue fragment crashes, after the queue fragment of a certain client is preempted by the current client, the current client can continue to consume the queue according to the consumption progress. To avoid repeated consumption of part of the data and make the consumption result more accurate

As for the device embodiment, since it is basically similar to the method embodiment, the description is relatively simple, and for related parts, please refer to the part of the description of the method embodiment.

The various embodiments in this specification are described in a progressive manner, and each embodiment focuses on the differences from other embodiments, and the same or similar parts between the various embodiments can be referred to each other.

Those skilled in the art should understand that the embodiments of the present invention can be provided as methods, devices, or computer program products. Therefore, the embodiments of the present invention may take the form of a completely hardware embodiment, a completely software embodiment, or an embodiment combining software and hardware. Moreover, the embodiments of the present invention may adopt computer program products implemented on one or more computer-usable storage media (including but not limited to disk storage, CD-ROM, optical storage, etc.) containing computer-usable program codes. form.

In a typical configuration, the computer device includes one or more consumer (processor) (CPU), input/output interfaces, network interfaces, and memory. Memory may include non-permanent memory in computer readable media, random access memory (RAM) and/or non-volatile memory, such as read-only memory (ROM) or flash memory ( flash RAM). Memory is an example of computer readable media. Computer-readable media includes permanent and non-permanent, removable and non-removable media, and information storage can be realized by any method or technology. Information can be computer-readable instructions, data structures, program modules, or other data. Examples of computer storage media include, but are not limited to, phase change memory (PRAM), static random access memory (SRAM), dynamic random access memory (DRAM), and other types of random access memory (RAM) , Read-only memory (ROM), electrically erasable programmable read-only memory (EEPROM), flash memory or other memory technology, CD-ROM, digital multi-function Optical discs (DVD) or other optical storage, magnetic cassettes, magnetic tape storage or other magnetic storage devices or any other non-transmission media that can be used to store information that can be accessed by computing devices. According to the definition in this article, computer-readable media does not include non-persistent computer-readable media (transitory media), such as modulated data signals and carrier waves.

The embodiments of the present invention are described with reference to the flowcharts and/or block diagrams of the methods, terminal devices (systems), and computer program products according to the embodiments of the present invention. It should be understood that each process and/or block in the flowchart and/or block diagram, and the combination of processes and/or blocks in the flowchart and/or block diagram can be realized by computer program instructions. These computer program instructions can be provided to general-purpose computers, dedicated computers, embedded consumer computers, or other programmable data consumer terminal devices to produce a machine, which can be executed by computers or other programmable data consumer terminal devices. The instructions generate means for realizing the functions specified in one process or multiple processes in the flowchart and/or one block or multiple blocks in the block diagram.

These computer program instructions can also be stored in a computer readable memory that can guide a computer or other programmable data consumer terminal equipment to work in a specific way, so that the instructions stored in the computer readable memory can be generated including the manufacturing of the instruction device The instruction device realizes the functions specified in one process or multiple processes in the flowchart and/or one block or multiple blocks in the block diagram.

These computer program instructions can also be loaded on a computer or other programmable data consumer terminal equipment, so that a series of operation steps are executed on the computer or other programmable terminal equipment to generate computer-realized consumption, so that the computer or other programmable terminal equipment The instructions executed on the design terminal device provide steps for implementing functions specified in one or more processes in the flowchart and/or one block or more in the block diagram.

Although the preferred embodiments of the embodiments of the present invention have been described, those skilled in the art can make additional changes and modifications to these embodiments once they learn the basic creative concepts. Therefore, the appended claims are intended to be interpreted as including the preferred embodiments and all changes and modifications falling within the scope of the embodiments of the present invention.

Finally, it should be noted that in this article, relational terms such as first and second are only used to distinguish one entity or operation from another entity or operation, and do not necessarily require or imply these entities. Or there is any such actual relationship or sequence between operations. Moreover, the terms "including", "including" or any other variants thereof are intended to cover non-exclusive inclusion, so that a process, method, article or terminal device including a series of elements not only includes those elements, but also includes those elements that are not explicitly listed. Other elements listed, or also include elements inherent to this process, method, article, or terminal device. Without more restrictions, the element defined by the sentence "including a..." does not exclude the existence of other identical elements in the process, method, article, or terminal device that includes the element.

The above provides a detailed introduction to a distributed environment coordinated consumption queue method and a distributed environment coordinated consumption queue system provided by the present invention. Specific examples are used in this article to illustrate the principle and implementation of the present invention. The description of the embodiments is only used to help understand the method and the core idea of the present invention; at the same time, for those with ordinary knowledge in the technical field, according to the idea of the present invention, there will be changes in the specific implementation and the scope of application In summary, the content of this specification should not be construed as a limitation of the present invention.

Claims (32)

A method for coordinating consumption queues in a distributed environment, which is characterized in that: for a queue segment to be consumed, based on a lease agreement, obtaining the queue lease time table stored in a persistent storage space The status data of the queue segment; based on the status data, determine whether other clients are consuming the queue segment; if it is determined that no other clients are consuming the queue segment, update the status of the queue segment Data, and obtain the current consumption progress of the queue segment; continue to consume the queue segment according to the current consumption progress, and record the new consumption progress of the queue segment; where the status data includes the queue The value of the consumer field, the value of the renter field, and the consumption time of the update time field in the queue segment in the lease period table. The method according to claim 1, wherein, before the step of obtaining the status data of the queue segment based on the lease agreement for a queue segment to be consumed, the method further includes: determining the current client demand based on the lease agreement Queue fragments to be consumed. The method according to claim 2, wherein the step of determining the queue fragments to be consumed based on the current client demand based on the lease agreement includes: Obtain the total number of active clients U, the total number of queue fragments P, and the total number of queue fragments consumed by the current client Q to calculate the number of queue fragments N that the current client needs to preempt; Among the fragments and/or queue fragments being consumed by other clients, preempt the N queue fragments as the queue fragments to be consumed by the current client. The method according to claim 3, wherein the step of obtaining the total number U of active clients, the total number P of queued shards, and the total number Q of queued shards consumed by the current client includes: obtaining from the persistent storage space The total number of active clients U is obtained from the client instance table stored in the Queue and the Queue Lease Table; the total number of shards P is obtained from the Queue System; the queue is stored from the persistent storage space Obtain the total number of queue fragments Q consumed by the current client in the column lease period table; pass N=[P/U]-Q to calculate the number of queue fragments N that the current client needs to preempt. The method according to claim 4, wherein the step of obtaining the total number of active clients U from the client instance table stored in the persistent storage space and the queue lease period table includes: obtaining the client instance table Record the number of clients started in the first time period live1; get the number of clients live2 corresponding to the queue segment without timeout in the queue lease table; the client without timeout is the queue segment The client recorded under the renter field of the film; Add live1 to live2 to get the total number of active clients U. The method according to claim 3, wherein, among the queue fragments that have timed out and/or the queue fragments being consumed by other clients, preempt the N queue fragments as the queue to be consumed by the current client The step of fragmentation includes: judging whether the N is greater than 0; if the N is greater than 0, judging whether the number of timeout queue fragments L is less than N; if the number of timeout queue fragments L is greater than or equal to N, then Preempt N queue fragments from the timeout queue fragments; if it is determined that the number of timeout queue fragments L is less than N, then seize L queue fragments from the timeout queue fragments, and Preempt NL queue fragments from the queue fragments being consumed by other clients. The method according to claim 6, wherein the step of preempting NL queue fragments from the queue fragments being consumed by other clients includes: the current client's queue fragments occupied by other clients Preempt the NL queue fragments in the middle, and make the difference between the number of the largest queue fragments occupied by each client and the number of the least queue fragments occupied by each client not exceed the specified number. The method according to claim 7, wherein the current client preempts each queue segment of NL from the queue segments occupied by other client terminals, and makes each client occupy the largest number of queue segments and the least occupied The steps for the difference between the number of queued shards not to exceed the specified number, including: based on the queued lease schedule, each client is queued according to its occupied queue The number of fragments is sorted from most to least; each time one or more preemption queue fragments are preempted from the first K client terminals, after the first K client terminals are preempted by J, the first K client terminals remain The average number of queue fragments is greater than the number of queue fragments currently occupied by the K+1th client, and the remaining queue fragments of each client in the first K client terminals are not different from the average number The specified number is exceeded, until NL queue fragments are successfully preempted. The method according to claim 3, wherein, among the queue fragments that have timed out and/or the queue fragments being consumed by other clients, preempt the N queue fragments as the queue to be consumed by the current client The fragmentation step includes: when a queue fragment is preempted, the consumer field of the preempted queue fragment in the queue lease table is modified to the current client. The method according to claim 3, wherein, among the queue fragments that have timed out and/or the queue fragments being consumed by other clients, preempt the N queue fragments as the queue to be consumed by the current client Before the fragmentation step, it also includes: every first time period, obtain the queue lease table, and determine whether the lease ID of the tenant field under each queue fragment in the queue lease table is the same as The lease ID of the corresponding queue segment recorded last time in the map table of the local storage area is the same; if the lease ID of a queue segment in the queue lease table is the same as the corresponding queue recorded last time in the map table If the lease ID of the segment is different, update the lease ID of the queue segment in the map table to the lease in the queue lease period table ID, and update the last monitoring time field of the queue segment in the map table to the current system time; if the lease ID of a queue segment in the queue lease period table is the same as the last recorded time in the map table If the lease ID of the corresponding queue shards is the same, the system time under the last monitoring time column is maintained, and it is judged whether the current system time minus the last monitored system time is greater than the first time period; if the current system time minus the last The system time of the second monitoring is greater than the first time period, then it is determined that the corresponding queue segment has timed out. The method according to claim 2, wherein the step of determining the queue fragments to be consumed for the current client demand based on the lease agreement comprises: determining the to-be-consumed current client demand based on the lease agreement every first time period The queue shards. The method according to claim 1, wherein, the consumption time is obtained by modifying the value of the update time field after the current client has preempted the queue segment; then, it is determined whether there are other clients on the basis of the status data The step of consuming the queue segment includes: determining whether the value of the consumer field of the queue segment is the current client; if the value of the consumer field of the queue segment is the current client, It is determined whether the preemption time of the current user terminal is greater than the consumption time; if the preemption time of the current user terminal is greater than the consumption time, it is determined that no other client terminal is consuming the queue segment. The method according to claim 12, wherein the preemption time Is the system time of the current user end, and the consumption time is the sum of the system time when the user end preempts the queue segment and the first cycle time. The method according to claim 1, wherein the step of updating the status data of the queue segment and obtaining the current consumption progress of the queue segment includes: renting the queue segment and making the queue persistent The consumer field under the queue shard in the queue lease table stored in the storage space is updated to the current client; the queue is obtained from the queue lease table stored in the persistent storage space The consumption progress under the progress column of the shard. The method according to claim 14, wherein the step of recording the new consumption progress of the queue segment includes: every second time period, determining that the queue segment is in the queue lease table Whether the tenant field of the queue is the current client; if the tenant field of the queue segment is the current client, update the current client's consumption progress for the queue segment to the queue lease Under the progress field of the queue segment in the schedule. The method according to claim 14, wherein, after the step of obtaining the consumption progress under the progress field of the queue segment from the queue lease time table stored in the persistent storage space, the method further includes: Determine whether the preemptor field and the consumer field under the queue segment in the queue lease table are the current client; if the preemptor field under the queue segment and the consumer field Whether the bit is the current client, the queue segment is renewed for the current client. A device for coordinating consumption queues in a distributed environment is characterized by comprising: a status data acquisition module for acquiring status data of a queue segment to be consumed based on a lease agreement; consumption judgment The module is used to determine whether other clients are consuming the queue segment based on the status data; the progress acquisition module is used to update the queue if it is determined that no other client is consuming the queue segment The status data of the shard and obtain the current consumption progress of the queue shard; the consumption module is used to continue to consume the queue shard according to the current consumption progress, and record the new consumption progress of the queue shard ; Wherein, the state data acquisition module includes: a first state data acquisition sub-module, used for a queue segment to be consumed, based on the lease agreement, from the queue lease table stored in the persistent storage space Obtain the status data of the queue segment in the queue; where the status data includes the value of the consumer field under the queue segment in the queue lease table, the value of the tenant field, and the update time field Consumption time. The device according to claim 17, wherein, before the status data acquisition module, it further includes: a queue segment determination module, configured to determine the queue segment to be consumed based on the current client demand based on the lease agreement. The device according to claim 18, wherein the queue fragment determination module includes: The demand quantity determination sub-module is used to obtain the total number of active clients U, the total number of queue shards P, and the total number of queue shards consumed by the current client Q to calculate the number of queue shards that the current client needs to preempt N; The preemption sub-module is used to preempt N queue segments from the timeout queue fragments and/or the queue fragments being consumed by other clients as the current client's queue fragments to be consumed. The device according to claim 19, wherein the demand quantity determination submodule includes: a client total number obtaining submodule, which is used to obtain a client instance table and the queue lease period table stored in the persistent storage space Obtain the total number of active clients U; obtain the total number of shards in the overall queue, which is used to obtain the total number of shards P from the queuing system; obtain the total number of shards in the single queue, obtain the sub-module, It is used to obtain the total number Q of queue shards consumed by the current client from the queue lease period table stored in the persistent storage space; the demand quantity calculation sub-module is used to pass N=[P/U]- Q. Calculate the number N of queue fragments that the current client needs to preempt. The apparatus according to claim 20, wherein the total number of client terminals obtaining submodule includes: a newly created client number obtaining submodule, which is used to obtain the users that have been activated in the most recent first time period recorded in the client instance table The number of clients live1; the sub-module for obtaining the number of clients that has not expired is used to obtain the lease period of the queue In the table, the number of clients corresponding to the queue fragment without timeout live2; the client without timeout is the client recorded under the tenant field of the queue fragment; the number of clients accumulates the submodule, using Add live1 to live2 to get the total number of active users U. The device according to claim 19, wherein the preemption sub-module includes: a first judging sub-module for judging whether the N is greater than 0; a second judging sub-module for if the N is greater than 0, then Determine whether the number of timeout queue fragments L is less than N; the full timeout preemption submodule is used to preempt N from the timeout queue fragments if the number of timeout queue fragments L is greater than or equal to N A queue shard; mixed preemption sub-module, used to determine that the number of timed out queue shards L is less than N, then preempt L queue shards from the timed out queue shards, and from other customers Among the queue fragments that the end is consuming, preempt the NL queue fragments. The device according to claim 22, wherein the hybrid preemption sub-module includes: a first hybrid preemption sub-module for the current client to preempt each queue segment of NL from the queue segments occupied by other client terminals , And make the difference between the number of shards occupying the largest queue and the number of shards occupying the least queued by each client does not exceed the specified number. The device according to claim 23, wherein the first hybrid preemption submodule includes: The sorting sub-module is used to sort each client according to the number of queue shards occupied by it based on the queue lease time table, from most to least; the second hybrid preemption module is used for the first K each time After one or more preemption queue fragments are preempted in the client, after the first K client terminals are preempted for J, the average number of remaining queue fragments for the first K client terminals is greater than that of the K+1th client terminal The number of queue fragments currently occupied, and the remaining number of queue fragments of each client in the first K client terminals does not differ from the average number by more than a specified number, until NL queue fragments are successfully preempted. The device according to claim 19, wherein the preemption submodule includes: a consumer field modification submodule, which is used to preempt a queue segment in the lease time table, and the preempted submodule The consumer field of the queue segment is modified to the current client. The device according to claim 19, wherein, before the preemption submodule, it further includes: a map table judging submodule, configured to obtain the queue lease table every first time period, and judge the queue Whether the lease ID of the tenant field under each queue segment in the lease table is the same as the lease ID of the corresponding queue segment recorded last time in the map table of the local storage area; map monitoring time update submodule , Used to update the lease of the queue segment in the map table if the lease ID of a queue segment in the queue lease table is different from the lease ID of the corresponding queue segment recorded last time in the map table ID is the lease ID in the queue lease period table, and the last monitoring time field of the queue segment in the updated map table is the current system time; The map monitoring time maintenance sub-module is used to maintain the last monitoring if the lease ID of a queue segment in the queue lease period table is the same as the lease ID of the corresponding queue segment recorded last time in the map table The system time under the time field, and determine whether the current system time minus the last monitored system time is greater than the first time period; the timeout judgment sub-module is used if the current system time minus the last monitored system time is greater than In the first time period, it is determined that the corresponding queue fragment has timed out. The device according to claim 18, wherein the queue fragmentation determination module includes: a first queue fragmentation determination sub-module configured to determine the current user end demand based on the lease agreement every first time period Queue fragments to be consumed. The device according to claim 17, wherein the consumption time is obtained by modifying the value of the update time field after the current user terminal preempts the queue segment; then, the consumption judgment module includes: a consumer field judgment sub The module is used to determine whether the value of the consumer field of the queue segment is the current client; the preemption time determination sub-module is used to determine if the value of the consumer field of the queue segment is the current client The client terminal determines whether the preemption time of the current client terminal is greater than the consumption time; the determining sub-module is used to determine that no other client terminal is consuming the queue segment if the preemption time of the current client terminal is greater than the consumption time. The device according to claim 28, wherein the preemption time Is the system time of the current user end, and the consumption time is the sum of the system time when the user end preempts the queue segment and the first cycle time. The device according to claim 17, wherein the progress acquisition module includes: a lease sub-module for renting the queue segment, and storing the queue in the queue lease time table in the persistent storage space The consumer field under the row segment is updated to the current client; the progress field reading module is used to obtain the queue segment’s data from the queue lease table stored in the persistent storage space The consumption progress under the progress bar. The device according to claim 30, wherein the consumption module includes: a renter judging sub-module for judging the lease of the queue segment in the queue lease period every second time period Whether the user field is the current client; the consumption progress update sub-module is used for if the tenant field of the queue segment is the current client, then the current client's consumption progress for the queue segment, Update to the progress column of the queue segment in the queue lease period table. The device according to claim 30, wherein, after the lease sub-module, it further includes: a lease renewal judging sub-module for judging the preemptor column under the queue segment in the queue lease period table Bit and whether the consumer field is the current client; The first lease renewal sub-module is used to renew the lease of the queue segment for the current client if the preemptor field and the consumer field under the queue segment are the current client.

Images