Patent Description:
A publish-subscribe function in a distributed system enables that a production device on a production platform distributes, to a message stream platform, log messages that are generated in a process of executing a service of a topic by the production device and that are used to record the executed service. A consumer device on a consumption platform subscribes to, on the message stream platform, the log messages corresponding to the topic of the service executed by the production platform. The message stream platform sends the log messages corresponding to the topic to the consumer device based on the subscribed topic. The consumer device processes content recorded in the log messages, to implement a preset function. The publish-subscribe function enables a looser coupling relationship between the production device and the consumer device. The production device does not care about an address of the consumer device, and the consumer device also does not care about an address of the production device.

The production platform includes a plurality of production devices, and the plurality of production devices may process the service in parallel. During service processing, each production device simultaneously generates log messages describing an execution process of the service. For example, in a scenario in which the production platform is a cloud storage platform that provides a cloud storage service for a user and the cloud storage platform is an object storage system, an enterprise user that provides an album service for a terminal user may request to allocate one or more buckets on the cloud storage platform. After the terminal user applies for the album service, the enterprise user may allocate, in the bucket, storage space to the terminal user for the user to store a photo. In this scenario, the topic of the service may be defined as the album. In a process in which the user stores the photo, each time the terminal user stores one photo in the bucket, the production platform generates one log message recording related information of the photo, for example, a topic identifier of the service, a photo generation time, a photo size, and an identifier of the bucket in which the photo is located. The log messages generated on the production platform are sent to the message stream platform.

The message stream platform includes one management device and a plurality of brokers. After receiving the log messages published by the production platform, the management device distributes the received log messages to the plurality of brokers. Each broker records the received messages in a log file corresponding to the topic identifier in the log message. There is a conventional technology for log data storage. For example, <CIT> discloses that raw log data is stored chronologically in the raw log data database. <CIT> discloses that a time series that is responsive to the request for time series may be determined by comparing event data for the time series stored in an event log to the filter criteria. <CIT> discloses that the log data is transmitted as a plurality of shards to a router of router-writer pairs.

The consumer platform includes a plurality of consumer devices, which are used to obtain the log messages recorded by the message stream platform and further process the obtained log messages. For log messages of a same topic, the consumption platform selects several consumer devices to form a consumer group to send, to the message stream platform, a topic identifier of log messages that need to be consumed by the consumer group, to subscribe to the log messages of the topic. The broker on the message stream platform may send, in parallel, to all consumer devices in the consumer group, the log messages recorded in a log file, corresponding to the topic, to consume the log messages of the same topic in parallel.

In the distributed system, although the messages of the same topic can be processed in parallel, due to performance differences of the brokers or the consumer devices, when messages of a topic are consumed in parallel, some log files are consumed comparatively quickly, and some log files are consumed comparatively slowly. As a result, the messages in the log files cannot be consumed synchronously.

Embodiments of the present invention provide a message stream processing method. A time segment mark is added to each of log messages on a message stream platform, and log messages generated in a same time segment are sent to a consumption platform, so that the consumption platform consumes the log messages based on the time segment.

The invention is defined by a method according to claim <NUM> and an apparatus according to claim <NUM>. Further embodiments are set out in the dependent claims. A first aspect of the present invention provides a message processing method applied to a message stream platform. The method includes: receiving a plurality of log messages from a production platform, where the plurality of log messages are used to record information about a first service executed by the production platform; storing the plurality of log messages in a log file based on a time segment; sending log messages that are related to the first service and that are in a first time segment to a consumption platform based on a subscription relationship between a consumer group and a topic of the plurality of log messages, wherein the log messages that are related to the first service are consumed by at least one consumer device in the consumer group included in the consumption platform; wherein the method further comprises: receiving an association request sent by the production platform, wherein the association request is sent by the production platform during splitting or aggregation of a shard to which the log messages belong; and according to the association request, setting a new time segment mark for each of subsequently received log messages related to the first service.

According to the method provided in the embodiment of the present invention, the time segment mark is added to each of the log messages on the message stream platform, and log messages generated in a same time segment are sent to the consumption platform, so that the consumption platform consumes the log messages based on the time segment, and the log messages are synchronized based on the time segment.

Optionally, in a possible implementation of the first aspect of the present invention, the sending log messages that are related to the first service and that are in a first time segment to a consumption platform includes: sending the log messages that are related to the first service and that are in the first time segment to the consumption platform according to a request from the consumption platform.

The log messages are sent to the consumption platform according to the request from the consumption platform, to implement on-demand consumption on the consumption platform.

Optionally, in a possible implementation of the first aspect of the present invention, the sending log messages that are related to the first service and that are in a first time segment to a consumption platform includes: after the log messages that are related to the first service and that are in the first time segment are sent to the consumption platform, sending log messages that are related to the first service and that are in a second time segment to the consumption platform.

The log messages are sent according to a time segment sequence, so that the consumption platform can consume the log messages according to the time segment sequence.

Optionally, in a possible implementation of the first aspect of the present invention, the storing the plurality of log messages in a log file based on a time segment includes: allocating a time segment mark to each of the plurality of log messages, where the time segment mark is used to identify a segment of time; and storing the plurality of log messages in the log file based on the time segment mark.

The time segment mark is set for the time segment, and consumption of log messages in each time segment may be controlled based on the time segment mark, to facilitate management.

Optionally, in a possible implementation of the first aspect of the present invention, the allocating a time segment mark to each of the plurality of log messages includes: periodically updating the time segment mark, and marking a time segment for each of subsequently received log messages by using the updated time segment mark.

Optionally, in a possible implementation of the first aspect of the present invention, the method further includes:
receiving the association request sent by the production platform, where the association request carries shard identifiers of both a second shard and a first shard or shard identifiers of both a second shard and a third shard, the second shard is obtained by splitting the first shard or by aggregating the first shard and the third shard, the first shard and the third shard each are a subset formed by the log messages about executing the first service, wherein shards are stored in a bucket, the splitting or aggregation are performed by the primary consumer device included in the production platform, the association request is used to set a new time segment mark for each of subsequently received log messages related to the first service and the shard identifiers are used to determine a partition to which the messages belong; and setting, according to the association request, a new time segment mark for each of subsequently received log messages belonging to the first shard and the second shard.

After a shard dynamically changes, the new time segment mark is added to each of the log messages, so that order-preserving consumption can still be implemented when a same object belongs to a different shard because of the dynamical change of the shard.

Optionally, in a possible implementation of the first aspect of the present invention, the method further includes:.

After a shard dynamically changes, the new time segment mark is added to each of the log messages, so that order-preserving consumption can be implemented when a same object belongs to a different shard because of the dynamical change of the shard.

A second aspect of the present invention provides a message stream platform. The message stream platform includes a management module and a broker module. The management module is configured to receive a plurality of log messages from a production platform, where the plurality of log messages are used to record information about a first service executed by the production platform; and forward the plurality of log messages to the broker module. The broker module is configured to store the plurality of log messages in a log file based on a time segment, and send log messages that are related to the first service and that are in a first time segment to a consumption platform based on a subscription relationship between a consumer group and a topic of the plurality of log messages, wherein the log messages that are related to the first service are consumed by at least one consumer device in the consumer group included in the consumption platform. The management module is further configured to receive an association request sent by the production platform, wherein the association request is sent by the production platform during splitting or aggregation of a shard to which the log messages belong. The broker module is further configured to set, according to the association request, a new time segment mark for each of subsequently received log messages related to the first service.

Optionally, in a possible implementation of the second aspect of the present invention, the broker module is specifically configured to: send the log messages that are related to the first service and that are in the first time segment to the consumption platform according to a request from the consumption platform.

Optionally, in a possible implementation of the second aspect of the present invention, the broker module is specifically configured to: send the log messages that are related to the first service and that are in the first time segment to the consumption platform based on a subscription relationship; and after the log messages that are related to the first service and that are in the first time segment are sent to the consumption platform, send log messages that are related to the first service and that are in a second time segment to the consumption platform.

Optionally, in a possible implementation of the second aspect of the present invention, the management module is further configured to set a time segment mark, where the time segment mark is used to identify a segment of time, and send the time segment mark to the broker module; and the broker module stores the plurality of log messages in the log file based on the time segment mark.

Optionally, in a possible implementation of the second aspect of the present invention, the management module is further configured to update the time segment mark periodically, and send the updated time segment mark to the broker module; and the broker module is further configured to mark a time segment for each of received log messages by using the updated time segment mark.

Optionally, in a possible implementation of the second aspect of the present invention, the management module is further configured to, when the association request sent by the production platform is received, update the time segment mark, and send the updated time segment mark to the broker module, where the association request carries shard identifiers of both a second shard and a first shard or shard identifiers of both a second shard and a third shard, the second shard is obtained by splitting the first shard or by aggregating the first shard and the third shard, the first shard and the third shard each are a subset formed by the log messages about executing the first service, wherein shards are stored in a bucket, the splitting or aggregation are performed by the primary consumer device included in the production platform, the association request is used to set a new time segment mark for each of subsequently received log messages related to the first service and the shard identifiers are used to determine a partition to which the messages belong; and
the broker module is configured to mark a time segment for each of subsequently received log messages by using the updated time segment mark.

Optionally, in a possible implementation of the second aspect of the present invention, the management module is further configured to, when the association request sent by the production platform is received, update the time segment mark, and send the updated time segment mark to the broker module, where the association request carries a shard identifier of a second shard, the second shard is obtained by splitting a first shard or by aggregating a first shard and a third shard, the first shard and the third shard each are a subset formed by the log messages about executing the first service, wherein shards are stored in a bucket, the splitting or aggregation are performed by the primary consumer device included in the production platform and the shard identifiers are used to determine a partition to which the messages belong; and the broker module is configured to set, according to the association request, a new time segment mark for each of subsequently received log messages related to the first service.

Effects achieved by various possible implementations of the message stream platform provided in the second aspect of the present invention are basically the same as those achieved by various possible implementations of the method provided in the first aspect.

A third aspect of the present invention which is not claimed, provides a message processing method applied to a message stream platform, including:
receiving log messages sent by a production platform, and storing the log messages based on a topic of a service executed by the production platform, where the log messages are used to record information about the service executed by the production platform, and the topic is used to identify the service executed by the production platform; receiving a consumption request sent by a consumption platform, where the consumption request carries a topic corresponding to log messages that the consumption platform requests to obtain; and sending log messages that are related to the topic and that are in a time segment to the consumption platform.

The storing the log messages based on a topic of a service executed by the production platform includes:.

Various possible implementations of the message processing method provided in the third aspect of the present invention are basically the same as the various possible implementations provided in the first aspect of the present invention, and have basically the same beneficial effects.

A fourth aspect of the present invention which is not claimed, provides a message stream platform, including.

Various possible implementations of the message processing method provided in the fourth aspect of the present invention are basically the same as the various possible implementations provided in the second aspect of the present invention, and have basically the same beneficial effects.

A fifth aspect of the present invention which is not claimed, provides a distributed system. The system includes:.

Various possible implementations of the distributed system provided in the fifth aspect of the present invention are basically the same as the various possible implementations provided in the second aspect of the present invention, and have basically the same beneficial effects.

A sixth aspect of the present invention which is not claimed, provides a distributed system. The system includes:.

Various possible implementations of the distributed system provided in the sixth aspect of the present invention are basically the same as the various possible implementations provided in the second aspect of the present invention, and have basically the same beneficial effects.

A seventh aspect of the present invention which is not claimed, provides a distributed system. The system includes:.

Various possible implementations of the distributed system provided in the seventh aspect of the present invention are basically the same as the various possible implementations provided in the second aspect of the present invention, and have basically the same beneficial effects.

An eighth aspect of the present invention which is not claimed, provides a message stream platform. The message stream platform includes:.

A ninth aspect of the embodiments of the present invention which is not claimed, provides a non-transient computer storage medium. The computer-readable storage medium stores a computer program. The computer program is executed by a processor to implement the method described in the first aspect or any possible implementation of the first aspect.

<FIG> is an architectural diagram of a distributed system according to an embodiment of the present invention. The distributed system includes a production platform <NUM>, a message stream platform <NUM>, and a consumption platform <NUM>. The production platform <NUM> includes a primary production device <NUM> and a plurality of production devices <NUM>. In addition to performing functions of the production device <NUM>, the primary production device <NUM> is further responsible for managing all production devices. When executing a service, the primary production device <NUM> and the production device <NUM> simultaneously generate log messages that describe a process of executing the service. In this embodiment of the present invention, the production platform is still used as a cloud storage platform. The cloud storage platform is used as an object storage system. An enterprise user requests to allocate a bucket on the cloud storage platform to provide an album service for a terminal user.

In this embodiment of the present invention, the primary production device <NUM> and the production device <NUM> on the production platform may be independent servers, or may be virtual devices. When the primary production device <NUM> and the production device <NUM> are the virtual devices, the plurality of virtual devices may be constructed on a plurality of servers, or may be constructed on a same server. A layout of the virtual machine may be determined based on an actual running environment.

The message stream platform <NUM> includes a management device <NUM> and a plurality of brokers <NUM>-<NUM> to <NUM>-N. The management device <NUM> is configured to receive the log messages sent by the production devices <NUM> on the production platform <NUM>, and send the received log messages to the brokers <NUM>-<NUM> to <NUM>-N for processing. Each of the brokers <NUM>-<NUM> to <NUM>-N stores the received log messages into a log file corresponding to a topic identifier in the log messages in a storage device <NUM>. The consumption platform <NUM> includes a primary consumer device <NUM> and a plurality of consumer devices <NUM>. In addition to performing a function of the consumer device <NUM>, the primary consumer device is further responsible for managing the consumer device <NUM>.

In this embodiment of the present invention, the management device <NUM> and the brokers <NUM>-<NUM> to <NUM>-N on the message stream platform <NUM> may be independent servers, or may be virtual devices. When the management device <NUM> and the brokers <NUM>-<NUM> to <NUM>-N are the virtual devices, the plurality of virtual devices may be constructed on a plurality of servers, or may be constructed on a same server. A layout of the virtual machine may be determined based on an actual running environment. When the brokers <NUM>-<NUM> to <NUM>-N are the independent servers, the storage device <NUM> is a built-in memory in the brokers <NUM>-<NUM> to <NUM>-N. When receiving the log messages sent by the message stream platform <NUM>, the brokers <NUM>-<NUM> to <NUM>-N directly store the received log messages in the built-in memory of the brokers <NUM>-<NUM> to <NUM>-N.

The consumption platform <NUM> includes the primary consumer device <NUM> and the consumer device <NUM>. In addition to performing the function of the consumer device <NUM>, the primary consumer device <NUM> is further responsible for managing all the consumer device <NUM>. When log messages of a topic on the message stream platform need to be used, the consumption platform first sets a consumer group formed by the consumer devices that use the log messages of the topic, and then sends a subscription request for the log messages of the topic. After receiving the subscription request, the message stream platform <NUM> sends the log messages of the topic, subscribed by the consumption platform <NUM> to each consumer device in the consumer group.

In this embodiment of the present invention, the primary consumer device <NUM> and the consumer device <NUM> on the consumption platform <NUM> may be independent servers, or may be virtual devices. When the primary consumer device <NUM> and the consumer device <NUM> are the virtual devices, the plurality of virtual devices may be constructed on a plurality of servers, or may be constructed on a same server. A layout of the virtual machine may be determined based on an actual running environment.

Currently, although some distributed systems can process the log messages in parallel, the distributed systems cannot process the log messages synchronously. If the messages cannot be processed synchronously, users' requirements in some specific scenarios cannot be met, for example, in a scenario the consumption platform needs to collect statistics on usage of a bucket by <NUM>:<NUM> p. to charge an enterprise user. When the usage of the bucket by <NUM>:<NUM> p. is determined, a sum of sizes of all photos stored in the bucket before <NUM>:<NUM> p. is collected. However, because the message stream platform processes the log messages in parallel, the log messages recording a photo size are scattered in log files corresponding to different brokers. Therefore, when the consumption platform obtains the log messages, each broker cannot synchronously send the log messages in the log file of each broker to the consumption platform. For example, one of the brokers sends, to the consumption platform, log messages with photos stored in the bucket before <NUM> p. Another broker sends, to the consumption platform, log messages with photos stored in the bucket before <NUM> p. As a result, the consumption platform cannot collect the statistics on the usage of the bucket by <NUM>:<NUM> p. , and the enterprise user who use the bucket cannot be charged. In this embodiment of the present invention, a mark is periodically set on the message stream platform for each log file having the same topic identifier. The mark is used to mark log messages generated in different time segments in each log file and a time segment sequence. In this way, when the consumer device consumes a plurality of log files, the log messages recorded in the plurality of log files of the broker in a same time segment are synchronously sent to the consumer device. In addition, after sending log messages in a time segment, log messages in a next time segment are sent to the consumer device according to the time segment sequence marked by the mark, to consume the log files synchronously. In this embodiment of the present invention, the broker sends one log message to the consumer device, to consume the log message.

The following describes publishing and consumption of log messages of a topic in detail in embodiments of the present invention with reference to flowcharts <NUM> to <NUM>.

<FIG> is a flowchart of a method for creating a topic corresponding to a topic service on a message stream platform <NUM>.

Step S201: A primary production device <NUM> creates the topic in response to a user's operation, and sets a quantity N of partitions of an object corresponding to the topic.

For example, when the topic defined by a user is recording usage of each bucket in an object storage system on a cloud platform, a primary production server creates a topic identifier topic <NUM> for the topic. To enable log messages about the topic, generated by a production platform <NUM> to be processed by a plurality of brokers on a message stream platform in parallel, when the topic is created, a quantity of log messages about the topic that can be processed in parallel, namely, a quantity N of partitions is set.

Step S202: The primary production device <NUM> sends a topic creation request to a management device <NUM> on the message stream platform <NUM>, where the topic creation request includes the topic identifier and the quantity N of partitions for the topic.

Step S203: After receiving the topic creation request, the management device <NUM> creates, in a storage device <NUM>, a file corresponding to the topic.

For example, if the topic identifier is the topic <NUM>, the file topic <NUM> is created in the storage device <NUM>, and an attribute of the topic is recorded in the created file, for example, the quantity N of partitions for the topic, a time segment in which the log messages recorded in the log file are generated, and an initial value of a mark of a time segment sequence. In this embodiment of the present invention, the mark is represented by a global sequence number (GLS). The GLS is used to mark log messages generated in each time segment, and the GLS may be incremented periodically to track a publishing sequence of the log messages of the topic. For example, assuming that a time for creating the file is <NUM>:<NUM> a. , and the GLS is incremented every one minute, an initial value of the GLS is recorded as <NUM> at <NUM>:<NUM> a. , the GLS is incremented to <NUM> at <NUM>:<NUM> a. , the GLS is incremented to <NUM> at <NUM>:<NUM> a. , the GLS is incremented to <NUM> at <NUM>:<NUM> a. , and so on. In this way, a mark in a time segment from <NUM>:<NUM> a. to <NUM>:<NUM> a. is <NUM>, a mark in a time segment from <NUM>:<NUM> a. to <NUM>:<NUM> a. is <NUM>, a mark in a time segment from <NUM>:<NUM> a. to <NUM>:<NUM> a. is <NUM>, and so on. Specific usage about the GLS will be described in detail below. It should be noted herein that the GLS is only an implementation of this embodiment of the present invention, and other manners used to mark different time segments and the time segment sequence are all included in the solutions disclosed in the present invention.

Step S204: The management device <NUM> determines, based on the quantity N of partitions carried in the topic creation request, brokers <NUM>-<NUM> to <NUM>-N that process the log messages about the topic in parallel.

After obtaining the quantity N of partitions from the topic creation request, the management device <NUM> obtains the N brokers <NUM>-<NUM> to <NUM>-N from the message stream platform, and generates N identifiers for the partitions, for example, pt(i), where i is a natural number from <NUM> to N. After the identifiers of the partitions are generated, an identifier of a partition processed by each of the brokers <NUM>-<NUM> to <NUM>-N is allocated in sequence. In other words, a mapping relationship between the partition pt(i) and each of the brokers <NUM>-<NUM> to <NUM>-N is established.

Step S205: The management device <NUM> initiates a partition creation request to each of the brokers <NUM>-<NUM> to <NUM>-N, to indicate each of the brokers <NUM>-<NUM> to <NUM>-N to create a log file corresponding to the partition processed by each of the brokers <NUM>-<NUM> to <NUM>-N.

When the management device <NUM> sends the partition creation request, the topic identifier, the partition identifier pt(i) of the partition processed by each of the brokers <NUM>-<NUM> to <NUM>-N, and the initial value <NUM> of the GLS may be carried in the partition creation request.

Step S206: After receiving the partition creation request, each of the brokers <NUM>-<NUM> to <NUM>-N creates a log file that corresponds to the identifier of the partition and a topic name in the partition creation request and the initial value of the GLS. In another implementation, each broker includes only one log file. Log messages about different topics are recorded in the log file. The log messages about services of the different topics are identified by using the topic name. In this embodiment of the present invention, for ease of description, an example in which one log file is generated for each topic is used for description.

For example, the log file may be represented as topic <NUM>/pt(i)/meslog, where the topic <NUM> is the topic name, pt(i) is the identifier of the partition, and meslog is an identifier of the log file. The log file is used to record the log messages of the partition pt(i) corresponding to the topic. The initial value of the GLS and an initial value of an offset location of the log file may be recorded as, for example, {gls = <NUM>, offt = <NUM>}, where gls = <NUM> indicates that the initial value of the GLS is <NUM>, as GLS is incremented, the value of the GLS is updated, and offt = <NUM> indicates that the initial value of the offset location of the log file, which is used to record a time point at which the GLS is updated and the offset location of the log file.

Step S207: After creating the log file for each partition, each of the brokers <NUM>-<NUM> to <NUM>-N returns response information to the management device <NUM>.

Step S208: After receiving the response information from each of the brokers <NUM>-<NUM> to <NUM>-N, the management device <NUM> sends a response message to the primary production device <NUM> on the production platform <NUM>, to notify the primary production device <NUM> that the topic is created.

After receiving the response information, the primary production device <NUM> may notify each production device <NUM> to publish the log messages about the topic to the message stream platform <NUM>.

A service of one topic includes a plurality of objects. For example, each photo stored in the bucket by a user is one object for a service of an album topic. To process the photos in parallel, the primary production device <NUM> may send the received photos to different production devices for processing in a load balancing or a polling manner. In this way, a photo processed by one production device belongs to one shard. After the photo is stored in the bucket, log messages recording photo information are log messages that belong to the shard. In other words, each shard is a subset of log messages generated when the service of the topic is executed. For example, if the bucket in the object storage system stores photos A to P, namely, <NUM> objects, and A to D, E and H, I to L, and M to P are separately processed by different production devices, four fragments are formed. If shard identifiers: a shard <NUM>, a shard <NUM>, a shard <NUM>, and a shard <NUM> are respectively set for the four shards, each shard may be represented as: a shard <NUM>: A to D, a shard <NUM>: E and H, a shard <NUM>: I to L and a shard <NUM>: M to P.

In the embodiments of the present invention, two scenarios are provided. One scenario is that a quantity of production devices on the production platform does not change in a process of executing the service. In other words, a quantity of shards does not change. The other scenario is that a quantity of production devices on the production platform changes in a process of executing the service. In other words, a quantity of shards changes dynamically. The following separately describes the two scenarios.

The scenario that the quantity of production devices on the production platform does not change in the process of executing the service is first described. In other words, the quantity of shards does not change. For a procedure in which a production platform publishes log messages in the process of executing the service, refer to <FIG>.

Step S301: Each production device <NUM> publishes log messages corresponding to each shard to the message stream platform <NUM>, where the log messages carry a topic identifier and a shard identifier.

For example, after storing the received photo into the bucket, the production device <NUM> may generate the log messages describing related information of the photo, and send the log messages including an identifier of a partition corresponding to the production device and the topic identifier of the service to the message stream platform <NUM>.

Step S302: After receiving the log messages, the management device <NUM> on the message stream platform <NUM> determines, based on the shard identifier included in the log messages, a partition to which the messages belong.

In the embodiments of the present invention, the partition to which the messages belong may be determined by performing hash calculation on the shard identifier. For example, the partition to which the shard belongs may be determined based on a remainder obtained by dividing the shard identifier by N. The hash algorithm is merely an example for description. Another method for mapping the message to the partition based on the shard identifier also falls within the protection scope of the present invention.

Step S303: The management device <NUM> determines, based on the determined partition, one of the brokers <NUM>-<NUM> to <NUM>-N that is corresponding to the partition.

In the step S204 in <FIG>, in a process in which the management device <NUM> creates the topic, the relationship between the partition and each of the brokers <NUM>-<NUM> to <NUM>-N is established. Therefore, the management device <NUM> may find, based on the mapping relationship, one of the brokers <NUM>-<NUM> to <NUM>-N that is corresponding to the partition.

Step S304: The management device <NUM> forwards the received log messages to the broker <NUM>-i corresponding to the partition.

Step S305: The broker <NUM>-i that receives the log messages finds, based on the topic identifier included in the log messages, a log file corresponding to the topic, and writes the log messages into the log file.

In the step S206 in <FIG>, when the broker <NUM>-i creates the log file, the topic of the log file is recorded. Therefore, the log file may be found based on the topic identifier in the messages.

Step <NUM> and step S307: The broker <NUM>-i sends, to the management device <NUM>, a response indicating that message writing is completed, and the management device <NUM> further sends, to the production device <NUM>, the response indicating that the message writing is completed.

The production device <NUM> may continuously write data into a plurality of log files corresponding to the topics based on the procedure.

Step S308: When detecting that a current time is a time for updating the GLS, the management device <NUM> increments the GLS in the file corresponding to the topic, for example, adding the GLS by one.

Step S309: The management device <NUM> sends an updated GLS to each broker <NUM>-i corresponding to the topic.

Step S310: After receiving the updated GLS, each of the brokers <NUM>-<NUM> to <NUM>-N records the updated GLS in the log file corresponding to the topic managed by the recorder and an offset location of the current log file, for example, if the updated GLS is <NUM> and the offset location of the current log file is XX, {gls = <NUM>, offt = XX} may be recorded.

Messages written in a current time segment and a writing sequence of the log messages written into the log file in the current time segment may be recorded by recording the GLS number and the offset location of the current log file. In this embodiment of the present invention, the GLS number is incremented to record the sequence of log messages written into the log file in different time segments. In another embodiment, another mark that may identify the sequence may also be used to indicate the sequence of the log messages written into the log file in the different time segments.

The consumption platform <NUM> may send a subscription request for the topic to the message stream platform, to consume the log messages in the log file corresponding to the topic. For example, if the topic subscribed by the consumption platform <NUM> is an album, the consumption platform <NUM> sends a subscription message including a topic identifier that identifies the album. Before consuming the log messages corresponding to the topic, the consumption platform <NUM> first establishes a consumption relationship with the brokers <NUM>-<NUM> to <NUM>-N on the message stream platform <NUM>. <FIG> is a flowchart of a method in which the message stream platform <NUM> creates the consumption relationship for the consumer device on the consumption platform <NUM>.

Step S401: The primary consumer device <NUM> on the consumption platform <NUM> constructs a consumer group that consumes the topic service, where the consumer group includes a plurality of consumer devices <NUM> that are specified to consume the topic.

When consuming messages of a topic, which are recorded on the message stream platform, the consumption platform <NUM> needs to first specify the consumer group for consuming the messages of the topic. Each consumer group includes at least one consumer device. There may be one or more consumer groups, and each consumer group has a unique identifier.

Step S402: The primary consumer device <NUM> sends the subscription request, where the subscription message carries the topic identifier and information of the consumer group, and the information of the consumer group includes an identifier of the consumer group and a quantity of consumer devices <NUM> included in the consumer group.

Step S403: After receiving the subscription request, the management device <NUM> generates a consumption record for the consumer group, where the consumption record includes the topic identifier, the identifier of the consumer group, the quantity of consumer devices included in the consumer group, and an initial GLS of a topic service, consumed by the consumer group, and the initial GLS is usually <NUM>. The consumption record is used to record a process in which the consumer group consumes a message of the topic. A specific method for the consumption record recording the process in which the consumer group consumes the message of the topic is described in detail below.

Step S404: After generating the consumption record, the management device <NUM> returns response information of the subscription request.

Step S405: After receiving the response message, the primary consumer device <NUM> creates a consumer device <NUM> for the consumer group, and indicates each consumer device <NUM> to send a creation request to the management device <NUM>.

After the consumption record is created, the primary consumer device <NUM> may specify the consumer device <NUM> that specifically consumes the message of the topic. In the step S402, when the consumer group is created, only the quantity of consumer devices <NUM> included in the consumer group is specified, but consumer devices specifically included in the consumer group are not specified. In the step S405, the consumer devices specifically included in the consumer group are specified. In another embodiment of the present invention, in step S402, a consumer device included in a consumer group may be specified when a consumer group is created. In this case, in step S405, each consumer device <NUM> is indicated to send a creation request to the management device <NUM>.

Step S406: The management device <NUM> allocates a partition corresponding to the subscribed topic to each consumer device.

The manager <NUM> allocates, based on a quantity of partitions, a data volume of a log file in each partition, and the quantity of consumer devices included in the consumer group, the partition corresponding to the subscribed topic to each consumer device <NUM> according to a load balancing principle. The consumption partition corresponding to the subscribed topic, allocated to each consumer device may be recorded in one consumption relationship, for example, {consumerGroup, topic <NUM>, pt(i), consumer(j)}, where consumerGroup indicates the identifier of the consumer group, and consumer(j) indicates the identifier of the consumer device.

When the quantity of consumer devices is equal to the quantity of partitions, one partition is allocated to each consumer device. When the quantity of consumer devices is less than the quantity of partitions, a plurality of partitions may be allocated to one consumer device.

Step S407: The management device <NUM> determines a broker corresponding to each consumer device.

After the consumption partition is allocated to each consumer device, the broker corresponding to each consumer device may be determined based on a correspondence between the partition and the broker. The mapping relationship between the consumer device <NUM> and the broker is also established. Because a plurality of partitions may be allocated to each consumer device <NUM>, each consumer device <NUM> may also correspond to a plurality of brokers.

Step S408: The management device <NUM> sends response information of the creation request to each consumer device <NUM>, where the response information carries an identifier of the broker corresponding to each consumer device.

Step S409: After receiving the response message, each consumer device <NUM> initiates a connection request to the broker indicated by the identifier of the broker, carried in the response message.

Step S410: After receiving the connection request, the broker establishes a connection to the consumer device <NUM>.

Step S411: After the connection is established, the broker <NUM> detects a link status between the broker <NUM> and the connected consumer device <NUM>, and reports to the management device <NUM> when the link is disconnected.

Step S412: After receiving the notification message, the management device <NUM> updates the quantity of consumer devices included in the consumer group recorded in the consumption record, and re-allocates a consumer device <NUM> to the broker <NUM> according to the load balancing principle.

After creating a consumption relationship for each consumer device <NUM>, the management device <NUM> may indicate each of the brokers <NUM>-<NUM> to <NUM>-N to send, to a corresponding consumer device <NUM> for consumption, the log messages corresponding to the topic subscribed to by a consumption platform managed by each of the brokers <NUM>-<NUM> to <NUM>-N. For a specific method, refer to <FIG>.

Step S501: The management device <NUM> sends a sending request to each of the brokers <NUM>-<NUM> to <NUM>-N that process the messages of the topic, where the sending request carries the topic identifier, the identifier of the consumer group, and a GLS that needs to be consumed currently.

Step S502: After receiving the sending request, each of the brokers <NUM>-<NUM> to <NUM>-N finds, based on the topic identifier, a log file corresponding to a partition processed by the broker <NUM>-i, determines an offset of a log file corresponding to the current GLS, and sends at the offset location, messages recorded in the log file.

In the log file, each GLS corresponds to one offset location in the log file. When the sending request is received, the broker <NUM>-i starts to send the messages in the log file at the offset location corresponding to the current GLS. One file is used to record a consumption progress of each consumer group on the log file. The consumption progress is an offset location, in the log file, of log messages that are sent to the consumption platform last time in the log file, for example, topic <NUM>/pt(i)/consumergroup <NUM>/off_progress, where off_progress indicates the offset location of the messages that is sent last time in the log file. Because one topic may be consumed by a plurality of consumer groups, a consumption progress of each consumer group on the log file is recorded.

Step S503: A consumer device <NUM> receives and stores the messages sent by the broker <NUM>-i.

Step S504: After detecting that the messages corresponding to the current GLS in the log file are sent, the broker <NUM>-i sends a notification message to the management device <NUM>, to notify the management device <NUM> that the messages corresponding to the current GLS in the log file are consumed.

That the messages corresponding to the current GLS are sent indicates that off_progress recorded in the consumption progress is adjacent to an offset location of a log file corresponding to a next GLS of the current GLS.

The notification message carries the topic identifier, the identifier of the consumer group, a GLS, and a message partition identifier. The management device may identify, based on the topic identifier, the identifier of the consumer group, and the message partition identifier, that a message corresponding to a GLS, in a partition of a specific topic, corresponding to a specific consumer group is consumed.

Step S505: After the management device <NUM> receives notification messages that are for the consumer group and that are sent by all brokers <NUM>-<NUM> to <NUM>-N corresponding to the topic service, the management device <NUM> updates the consumption progress of the consumer group on the topic, namely, incrementing the GLS in the consumption record, and returns to perform the step S501 until all log files corresponding to each partition are consumed.

In the scenario, after the log files of each partition in one time segment are consumed by setting the SGL, log messages of the log file in a next time segment are sent, so that the log files of each partition are consumed synchronously.

For the log messages sent to the consumption platform <NUM>, the consumption platform <NUM> may perform further processing on the log messages as required, for example, charging based on usage of a bucket at one moment. The log messages generated in each time segment are marked and the messages generated in the same time segment are sent to the consumption platform, so that the consumption platform can charge the usage of the bucket at the moment.

The second scenario provided in this embodiment of the present invention is that in the another scenario, in the process of executing the service by the distributed system, the quantity of production devices on the production platform changes, and the quantity of shards also dynamically changes. For example, because of service volume increase, the production devices may be increased, or because of some faulty production devices, the production devices <NUM> are reduced. The dynamic change of the quantity of production devices <NUM> may cause the change of a shard for each topic service. For example, if the production devices are increased, a shard processed by a production device with a heavy load is split to generate two shards.

For example, as shown in <FIG>, service objects included in a shard <NUM> are E to H. Because of production device increase, the shard <NUM> may be split into two shards (in another embodiment, an original shard may be separated into one shard, for example, the shard <NUM> is separated into a shard <NUM>): a shard <NUM>: E to F and a shard <NUM>: G to H. As time goes by, the shard <NUM> and the shard <NUM> may further be split. For example, the shard <NUM> is further split into a shard <NUM> and a shard <NUM>. The shard <NUM> is further split into a shard <NUM> and a shard <NUM>. A shard <NUM> and a shard <NUM> may be aggregated into a shard <NUM>. Shard splitting and aggregation cause a change of a partition to which log messages of the shard belong. As a result, log messages of a same object are distributed to different partitions. For example, an object E is processed by a production device A during storage. A shard to which the object E belongs is the shard <NUM>, and a partition to which the shard <NUM> is mapped is P(<NUM>). However, because the production device A is faulty, or a production device is added, objects processed by the production device A are processed by a production device B and a production device C separately. Therefore, the shard <NUM> is split into the shard <NUM> and the shard <NUM>, and the shard to which the service object E belongs changes from the shard <NUM> to the shard <NUM>. As described above, a partition to which a shard belongs is determined based on the shard identifier. Because the shard to which the service object E belongs changes from the shard <NUM> to the shard <NUM>, the partition to which the object belongs may also change from P(<NUM>) to P(<NUM>). Therefore, because log files in different shards are consumed in parallel, order-preserving consumption cannot be implemented after the shard for the service object E changes.

For example, for a photo that initially belongs to the shard <NUM>, messages in the shard <NUM> are mapped to the partition P(<NUM>). When the photo is stored in the bucket, a production device publishing the log messages of the shard <NUM> publishes log messages that record generation of the photo, a broker <NUM>-<NUM> corresponding to P(<NUM>) stores the log messages in a log file corresponding to P(<NUM>). Because of the splitting of the shard <NUM>, the photo belongs to the shard <NUM>, and the shard <NUM> is mapped to the partition P(<NUM>). In this case, if the photo is deleted from the bucket, the production device <NUM> publishing the shard <NUM> generates messages that record deletion of the photo. A broker <NUM>-<NUM> corresponding to P(<NUM>) stores the messages in a log file corresponding to P(<NUM>). In this way, in a disaster recovery and backup scenario, after the photo in the bucket on the production platform is backed up to the consumption platform, the consumption platform further obtains the log messages of the photo from the message stream platform, to ensure that data on the consumption platform is consistent with data on the production platform. However, in a process of obtaining the log messages, because of the dynamic change of the shard, the log messages of the same photo may be recorded in different log files. The log messages recording that the photo is generated may be consumed after the log messages recording the deleting of the photo. This causes log management confusion on the consumption platform.

A method shown in <FIG> in an embodiment of the present invention is a method for ensuring order-preserving consumption of objects after the shard dynamically changes in the second scenario.

Step S601: When the shard dynamically changes, for example, after splitting or aggregation occurs, a new shard is generated, a primary production device <NUM> publishes an association request, where the association request carries a topic identifier of a topic service, the new shard identifier, and an identifier of an associated shard.

The identifier of the associated shard is an identifier of a shard before splitting or aggregation into the new shard, namely, an identifier of a shard that requires order preserving. For example, a shard <NUM> and a shard <NUM> are split from a shard <NUM>, and the shard <NUM> and the shard <NUM> needs to be order-preserved with the shard <NUM>. In this case, the shard <NUM> and the shard <NUM> are the identifiers of the new shards, and the shard <NUM> is the identifier of the associated shard.

Step S602: After receiving the association request, a management device <NUM> finds, based on the topic identifier, a current GLS corresponding to the topic identifier, and updates the GLS, for example, adds <NUM>.

As described above, a file in the management device records the current GLS number. After the association request is received, the current GLS number may be found in the file.

Step S603: The management device <NUM> determines a partition to which the new shard belongs, and determines a broker <NUM>-i corresponding to the partition.

Step S604: The management device <NUM> notifies the broker <NUM>-i corresponding to the new shard and a broker corresponding to the associated shard to update the GLS.

Step S605: The broker <NUM>-i that receives the notification updates the GLS in the log file corresponding to the topic, and records an offset address of the log file corresponding to the GLS.

Step S606: The broker <NUM>-i sends a response to the notification message to the management device <NUM>.

Step S607: After receiving the response information of the notification message, the management device <NUM> sends response information of the association request to the primary production device <NUM>.

Step S608: After receiving the response message, the primary production device <NUM> allocates a production device <NUM> to the new shard, and notifies the new production device <NUM> to publish a message corresponding to the new shard.

After receiving the notification, the production device may normally publish the messages of the new shard.

In another implementation of the second scenario, in step <NUM>, a production device <NUM> that publishes a new shard may publish an association request which does not include an identifier of an associated shard. However, in step <NUM>, all brokers associated with the topic are notified to update a GLS. In this way, the changed shard is consumed after a shard before the change, to implementing order preserving of a same object in different shards.

In the second scenario, an order-preserving relationship between the shard before the change and the changed shard is set, and the GLS is updated to ensure that the changed shard is consumed after the shard before the change. In this way, even if the same object is allocated to the different partitions after the change, the order-preserving consumption may also be ensured.

<FIG> and <FIG> are a schematic diagram in which after a shard <NUM> dynamically changes in a message publishing process, order preserving is performed, by using a GLS, for an object E that belongs to different shards after the object E dynamically changes. As shown in the figure, T1 to T9 are preset times for updating the GLS. After the shard <NUM> is split between T2 and T3, a shard to which the object E belongs changes from the shard <NUM> to a shard <NUM>, and a management device <NUM> adds the GLS once, for example, <NUM>. The added GLS is sent to a broker <NUM>-i corresponding to an associated shard <NUM>, the new shard <NUM>, and a shard <NUM>. In log files shard <NUM> meglog, shard <NUM> meglog, and shard <NUM> meglog corresponding to the shard <NUM>, the shard <NUM>, and the shard <NUM>, brokers <NUM>-<NUM> to <NUM>-N record offsets corresponding to the GLS in the log files. Then, between T4 and T5, the shard <NUM> is split into a shard <NUM> and a shard <NUM>, and the shard <NUM> is split into a shard <NUM> and a shard <NUM>. Between T6 and T7, the shard <NUM> and the shard <NUM> are aggregated into a shard <NUM>. The shard to which the object E belongs changes from the shard <NUM> to the shard <NUM>, and then to the shard <NUM> and the shard <NUM>. The log file that records the object E changes from shard <NUM> meglog to shard <NUM> meglog, and then to shard <NUM> meglog, and shard <NUM> meglog. In this way, when a consumer group consumes the log files in a GLS sequence, even if messages of the object E are recorded in different log files before and after the shard change, the order-preserving consumption may further be performed.

An example in which the object is a photo and the photo belongs to the shard <NUM> is used as example, when the photo is stored in a bucket in an object storage system, messages recording that the photo is generated are stored in a log file corresponding to P(<NUM>). Then, because of splitting of the shard <NUM>, the photo belongs to the new shard <NUM> and a partition corresponding to the new shard <NUM> is P(<NUM>). After the shard <NUM> is split, the management device increases the GLS from original <NUM> to <NUM>. In this case, a GLS number is increased to the log files corresponding to P(<NUM>) and P(<NUM>), and a current offset location of the log file is recorded. When the photo is deleted from the bucket, messages recording that the photo is deleted are recorded in the log file corresponding to P(<NUM>), and are recorded after a corresponding offset location when the GLS is <NUM>. The messages recording that the photo is generated are recorded before the corresponding offset location when the GLS is <NUM>. When the consumption platform consumes the log files according to a sequence of the GLS, messages before the corresponding offset location when the GLS is <NUM> in the log files are consumed first, and then messages after the corresponding offset location when the GLS is <NUM> in the log files are consumed. In this way, even if change messages of the photo are written into the different log files after the splitting, the order-preserving consumption can be implemented.

<FIG> is a structural diagram of a message stream platform <NUM> according to an embodiment of the present invention. The message stream platform <NUM> includes a management module <NUM> and a plurality of brokers <NUM>-<NUM> to <NUM>-N. The management module <NUM> is applied to the management device <NUM>, and the plurality of brokers <NUM>-<NUM> to <NUM>-N are respectively applied to the brokers <NUM>-<NUM> to <NUM>-N. The message stream platform <NUM> may be implemented by using hardware, or may be implemented by using hardware. The management module <NUM> is configured to receive a plurality of log messages from the production platform <NUM>, and forward the plurality of log messages to the brokers <NUM>-<NUM> to <NUM>-N. Any one broker <NUM>-i is configured to store the plurality of log messages in a log file based on a time segment, and send log messages that are related to the first service and that are in a first time segment to a consumption platform. Specifically, in an implementation, the broker module <NUM>-i is configured to send the log messages that are related to the first service and that are in the first time segment to the consumption platform according to a request from the consumption platform. In another implementation, the broker module <NUM>-i is configured to send the log messages that are related to the first service and that are in the first time segment to the consumption platform based on a subscription relationship; and after the log messages that are related to the first service and that are in the first time segment are sent to the consumption platform, send log messages that are related to the first service and that are in a second time segment to the consumption platform. The management module <NUM> is further configured to set a time segment mark, where the time segment mark is used to identify a segment of time, and send the time segment mark to the broker module; and the broker module <NUM>-i stores the plurality of log messages in the log file based on the time segment mark. The management module <NUM> periodically updates the time segment mark, and sends the updated time segment mark to the broker module <NUM>-i. The broker module <NUM>-i is further configured to mark a time segment for each of received log messages by using the updated time segment mark.

In an implementation of the present invention, the management module <NUM> is further configured to, when an association request sent by the production platform is received, update the time segment mark, and send the updated time segment mark to the broker module <NUM>-i, where the association request carries shard identifiers of both a second shard and a first shard or shard identifiers of both a second shard and a third shard, the second shard is obtained by splitting the first shard or by aggregating the first shard and the third shard, the first shard and the third shard each are a subset formed by the log messages about executing the first service, and the association request is sent by the production platform during the splitting or aggregation; and the broker module <NUM>-i is configured to mark a time segment for each of subsequently received log messages by using the updated time segment mark.

In another implementation of the present invention, the management module <NUM> is further configured to, when an association request sent by the production platform is received, update the time segment mark, and send the updated time segment mark to the broker module <NUM>-i, where the association request carries a shard identifier of a second shard, the second shard is obtained by splitting a first shard or by aggregating a first shard and a third shard, the first shard and the third shard each are a subset formed by the log messages about executing the first service, and the association request is sent by the production platform during the splitting or aggregation; and the broker module <NUM>-i is configured to set, according to the association request, a new time segment mark for each of subsequently received log messages related to the first service.

Specifically, the management module <NUM> performs steps S303, S304, S308, and S309 in the method shown in <FIG>, steps S501 and S505 in <FIG>, steps S602, S603, S604, and S607 in <FIG> to implement the functions. The broker module <NUM>-i performs steps S305, S306, and S310 in the method shown in <FIG>, steps S502 and S504 in <FIG>, and steps S605 and S606 in <FIG> to implement the functions.

<FIG> is a structural diagram of hardware of the management device <NUM> and any broker <NUM>-i on the message stream platform <NUM>. As shown in the figure, the management device <NUM> includes a processor <NUM>, a memory <NUM>, a bus <NUM>, and a communications interface <NUM>. A structure of the broker <NUM> is basically the same as that of the management device, and includes a processor <NUM>, a memory <NUM>, a bus <NUM>, and a communications interface <NUM>. The processor <NUM> and the processor <NUM> may have a plurality of specific implementation forms, for example, may be a central processing unit (CPU) or an image processing unit (GPU), or the processor <NUM> may be a single-core processor or a multi-core processor. The processor <NUM> and the processor <NUM> may be a combination of a CPU and a hardware chip. The hardware chip may be an application-specific integrated circuit (ASIC), a programmable logic device (PLD), or a combination thereof. The PLD may be a complex programmable logic device (CPLD), a field programmable logic gate array (FPGA), generic array logic (GAL), or any combination thereof. The processor <NUM> and the processor <NUM> may also be separately implemented by using a logic device with built-in processing logic, for example, an FPGA or a digital signal processor (DSP).

The communications interface <NUM> and the communications interface <NUM> may be a wired interface or a wireless interface, and are configured to communicate with another unit or device. The wired interface may be an Ethernet interface or a local interconnect network (LIN). The wireless interface may be a cellular network interface, a wireless local area network interface, or the like. The bus <NUM> may be classified into an address bus, a data bus, a control bus, and the like.

Storage media of the memory <NUM> and the memory <NUM> may be a volatile memory and a non-volatile memory, or may include both a volatile memory and a non-volatile memory. The non-volatile memory may be a read-only memory (ROM), a programmable read-only memory (PROM), an erasable programmable read-only memory (EPROM), an electrically erasable programmable read-only memory (EEPROM), or a flash memory. The volatile memory may be a random access memory (RAM), used as an external cache. Through example but not limitative description, many forms of RAMs may be used, for example, a static random access memory (SRAM), a dynamic random access memory (DRAM), a synchronous dynamic random access memory (SDRAM), a double data rate synchronous dynamic random access memory (DDR SDRAM), an enhanced synchronous dynamic random access memory (ESDRAM), a synchlink dynamic random access memory (SLDRAM), and a direct rambus dynamic random access memory (DR RAM). The memory <NUM> and the memory <NUM> are configured to store program code and data. The processor <NUM> executes the program code in the memory <NUM> to implement the method steps performed by the management device in <FIG>, <FIG>, <FIG>, <FIG>, and <FIG>. The processor <NUM> executes the program code in the memory <NUM> to implement the method steps performed by the broker in <FIG>, <FIG>, <FIG>, <FIG>, and <FIG>.

Claim 1:
A message processing method applied to a message stream platform, wherein the method comprises:
receiving a plurality of log messages from a production platform, wherein the plurality of log messages are used to record information about a first service executed by the production platform;
storing the plurality of log messages in a log file based on time segments;
sending log messages that are related to the first service and that are in a first time segment to a consumption platform based on a subscription relationship between a consumer group and a topic of the plurality of log messages, wherein the log messages that are related to the first service are consumed by at least one consumer device in the consumer group included in the consumption platform;
characterized in that the method further comprises:
receiving an association request sent by the production platform, wherein the association request is sent by the production platform during splitting or aggregation of a shard to which log messages from the production platform belong; and
setting, according to the association request, a new time segment mark for each of subsequently received log messages related to the first service.