CN117194413A

CN117194413A - Delay message processing method, delay message processing device, computer equipment and storage medium

Info

Publication number: CN117194413A
Application number: CN202310926136.4A
Authority: CN
Inventors: 罗成; 郑勇飞; 刘义
Original assignee: Hangzhou Differsoft Co ltd; Hangzhou Jike Cloud Network Technology Co ltd
Current assignee: Hangzhou Differsoft Co ltd; Hangzhou Jike Cloud Network Technology Co ltd
Priority date: 2023-07-26
Filing date: 2023-07-26
Publication date: 2023-12-08

Abstract

The embodiment of the application discloses a delay message processing method, a delay message processing device, computer equipment and a storage medium. The method comprises the following steps: acquiring a message generated by a service function; storing the information into a database, and writing the information meeting the requirements into a redis queue; ordering the messages in the redis queue according to the expected sending time to obtain an ordering result; periodically scanning the sequencing result, and putting the message with the predicted sending time longer than the current time stamp into a queue to be sent; the corresponding information is regularly fetched from the queue to be sent and submitted to the appointed delay information processing object; updating the messaging status within the database. By implementing the method of the embodiment of the application, the problems of single applicable scene, complex use, excessively high occupied memory and the like of the Rabbitmq delay message can be solved.

Description

Delay message processing method, delay message processing device, computer equipment and storage medium

Technical Field

The present application relates to a computer, and more particularly, to a delayed message processing method, apparatus, computer device, and storage medium.

Background

The current common message queue delay message scheme is to use the dead message of Ribbitmq to realize the delay delivery of the message, wherein the dead message of Ribbitmq refers to a queue ttl plus dead message exchange; the principle is that two queues are used, one queue receives messages without consumption, the messages die after waiting for a specified time, and the dead message exchange bound by the queue is routed to the other queue again to provide service consumption. However, the method can only support the message with fixed delay level, can not support the delay of the self-defined time, and has smaller use scene; the use mode is complex, a stack of queues and exchange are required to be declared, the large-scale use of delay messages is not facilitated, and great difficulty is brought to queue management; for the large memory pressure of the mq queue, delay messages are stored in the memory of the mq, if the delay messages are too many or the delay time is too long, the mq is easy to have the condition of insufficient memory, thereby triggering the rabkitmq page-changing blocking queue; cancellation of the delayed message cannot be supported, and once the delayed message is sent out, it must be judged at the consumer side whether to execute or not, and the delayed message cannot be cancelled in advance.

Therefore, a new method is necessary to be designed, and the problems of single applicable scene, complex use, excessively high occupied memory and the like of Rabbitmq delay messages are solved.

Disclosure of Invention

The application aims to overcome the defects of the prior art and provide a delay message processing method, a delay message processing device, computer equipment and a storage medium.

In order to achieve the above purpose, the present application adopts the following technical scheme: a method of deferred message processing, comprising:

acquiring a message generated by a service function;

storing the information into a database, and writing the information meeting the requirements into a redis queue;

ordering the messages in the redis queue according to the expected sending time to obtain an ordering result;

periodically scanning the sequencing result, and putting the message with the predicted sending time longer than the current time stamp into a queue to be sent;

the corresponding information is regularly fetched from the queue to be sent and submitted to the appointed delay information processing object;

updating the messaging status within the database.

The further technical scheme is as follows: after the message generated by the service function is acquired, the method further comprises the following steps:

a delay time and a message delivery object are specified.

The further technical scheme is as follows: the steps of storing the message into a database and writing the message meeting the requirements into a redis queue include:

storing the message in a database;

and carrying out message scheduling management by using a time-roulette algorithm, and writing the messages meeting the requirements into a redis queue.

The further technical scheme is as follows: the ordering the messages in the redis queue according to the expected sending time to obtain an ordering result, which comprises the following steps:

using the zset queue of the redis queue as a cache queue;

and using the queuing characteristic of the score, taking the predicted sending time stamp of the message as the score value, taking the message content as the queue value, and sequencing the message according to the score value by the queue to obtain a sequencing result.

The further technical scheme is as follows: the step of periodically scanning the sequencing result, and putting the message with the predicted sending time longer than the current timestamp into a queue to be sent, which comprises the following steps:

and periodically scanning the zset queue through the lua script to scan the sequencing result, taking out the message with the score value larger than the current time stamp, and putting the message into a queue to be sent.

The application also provides a delay message processing device, which comprises:

a message acquisition unit for acquiring a message generated by the service function;

a writing unit, configured to store the message into a database, and write a message meeting the requirement into a redis queue;

the ordering unit is used for ordering the messages in the redis queue according to the expected sending time so as to obtain an ordering result;

the scanning unit is used for periodically scanning the sequencing result and placing the message with the predicted sending time longer than the current time stamp into a queue to be sent;

the extraction unit is used for regularly extracting the corresponding message from the queue to be sent and submitting the message to the designated delay message processing object;

and the updating unit is used for updating the message sending state in the database.

The further technical scheme is as follows: further comprises:

and the designating unit is used for designating the delay time and the message delivery object.

The further technical scheme is as follows: the writing unit includes:

a storage subunit configured to store the message into a database;

and the scheduling management subunit is used for carrying out message scheduling management by using a time-roulette algorithm and writing the messages meeting the requirements into the redis queue.

The application also provides a computer device which comprises a memory and a processor, wherein the memory stores a computer program, and the processor realizes the method when executing the computer program.

The present application also provides a storage medium storing a computer program which, when executed by a processor, implements the above method.

Compared with the prior art, the application has the beneficial effects that: according to the application, the messages are stored in the database, the messages meeting the requirements are written into the redis queue, the messages are queued and cached by utilizing zset, the messages with the time longer than the current time stamp are sequenced according to the predicted sending time, and the messages with the time longer than the current time stamp are put into the queue to be sent for sending, so that the problems that the Rabbittmq delay message is single in applicable scene, complex to use, excessively high in occupied memory and the like are solved.

The application is further described below with reference to the drawings and specific embodiments.

Drawings

In order to more clearly illustrate the technical solutions of the embodiments of the present application, the drawings required for the description of the embodiments will be briefly described below, and it is obvious that the drawings in the following description are some embodiments of the present application, and other drawings may be obtained according to these drawings without inventive effort for a person skilled in the art.

Fig. 1 is a schematic diagram of an application scenario of a delay message processing method according to an embodiment of the present application;

fig. 2 is a flow chart of a delay message processing method according to an embodiment of the present application;

fig. 3 is a schematic sub-flowchart of a delayed message processing method according to an embodiment of the present application;

fig. 4 is a schematic sub-flowchart of a delayed message processing method according to an embodiment of the present application;

FIG. 5 is a schematic diagram of a time-roulette algorithm according to an embodiment of the present application;

fig. 6 is a schematic block diagram of a delayed message processing apparatus according to an embodiment of the present application;

FIG. 7 is a schematic block diagram of a write unit of a delayed message processing apparatus provided by an embodiment of the present application;

fig. 8 is a schematic block diagram of an ordering unit of a delayed message processing apparatus according to an embodiment of the present application;

FIG. 9 is a schematic block diagram of a computer device provided by an embodiment of the present application;

fig. 10 is a schematic diagram of filtering invalid data by a maximum timestamp screening mechanism according to an embodiment of the present application.

Detailed Description

The following description of the embodiments of the present application will be made clearly and fully with reference to the accompanying drawings, in which it is evident that the embodiments described are some, but not all embodiments of the application. All other embodiments, which can be made by those skilled in the art based on the embodiments of the application without making any inventive effort, are intended to be within the scope of the application.

It should be understood that the terms "comprises" and "comprising," when used in this specification and the appended claims, specify the presence of stated features, integers, steps, operations, elements, and/or components, but do not preclude the presence or addition of one or more other features, integers, steps, operations, elements, components, and/or groups thereof.

It is also to be understood that the terminology used in the description of the application herein is for the purpose of describing particular embodiments only and is not intended to be limiting of the application. As used in this specification and the appended claims, the singular forms "a," "an," and "the" are intended to include the plural forms as well, unless the context clearly indicates otherwise.

It should be further understood that the term "and/or" as used in the present specification and the appended claims refers to any and all possible combinations of one or more of the associated listed items, and includes such combinations.

Referring to fig. 1 and fig. 2, fig. 1 is a schematic diagram of an application scenario of a delay message processing method according to an embodiment of the present application. Fig. 2 is a schematic flow chart of a delay message processing method according to an embodiment of the present application. The delay message processing method is applied to the server. The server performs data interaction with the terminal, realizes overall management through the DMS service by linking the database, the cache and the mq, solves the problems of single applicable scene, complex use, excessively high occupied memory and the like of Rabbittmq delay message, and greatly enriches applicable scenes of delay message functions; solving a plurality of problems in the existing Rabbitmq scheme, such as supporting random delay time, simplifying use difficulty, supporting real-time cancellation of delay messages, and storing a large amount of messages in a database through a roulette algorithm, thereby reducing the internal memory pressure of a middleware

Fig. 2 is a flow chart of a delay message processing method according to an embodiment of the present application. As shown in fig. 2, the method includes the following steps S110 to S170.

S110, obtaining the information generated by the service function.

In this embodiment, the message refers to the content generated by the service function and is input by the terminal.

S120, storing the information into a database, and writing the information meeting the requirements into a redis queue.

In one embodiment, referring to fig. 3, the step S120 may include steps S121 to S122.

S121, storing the message into a database;

s122, performing message scheduling management by using a time-roulette algorithm, and writing the messages meeting the requirements into a redis queue.

In this embodiment, referring to fig. 5, the scheduling management of the message by using the time-carousel algorithm is more comprehensive and not easy to miss. The time wheel is a scheduling model that efficiently utilizes thread resources for batch scheduling. All the dispatching tasks in a large batch are bound to the same dispatcher, and the dispatcher is used for managing all tasks, triggering and running (runnable), so that various delay tasks, periodic tasks, notification tasks and the like can be efficiently managed.

S130, designating delay time and a message delivery object.

In this embodiment, the expected transmission time can be determined by specifying the delay time, and the expected transmission time is equal to the current timestamp plus the delay time. A messaging object refers to an object of a message transmission.

S140, ordering the messages in the redis queue according to the expected sending time to obtain an ordering result.

In this embodiment, the ordering result refers to a result formed after the messages in the redis queue are ordered according to the expected sending time.

In one embodiment, referring to fig. 4, the step S140 may include steps S141 to S142.

S141, using the zset queue of the redis queue as a cache queue;

s142, using the queuing characteristic of the score, taking the predicted sending time stamp of the message as the score value, taking the message content as the queue value, and sequencing the message according to the score value by the queue to obtain a sequencing result.

The properties of the zset-made message queue are: ordering of: all elements in zset are automatically ordered. This makes zset well suited for use with ordered message queues because messages can be retrieved on demand according to one or more criteria, such as time of arrival or priority of the message. Element uniqueness: each element of zset is unique, which is very helpful in achieving certain message requirements (e.g., idempotency). Mapping relationship between members and scores: each member of the ordered set has a score that allows the same data to be partitioned into different queues and different delays to be set for each queue. Efficient add delete operation: because zset automatically maintains the order between elements, manual ordering is not required when elements are added or deleted, thereby improving the speed of operation.

In summary, the zset of the Redis naturally supports the message queues according to the time sequence, so that the unique property of the members of the zset can be utilized to ensure that the messages are not repeatedly consumed, and the zset has great advantages in the aspects of high throughput rate and the like.

And S150, periodically scanning the sequencing result, and putting the message with the predicted sending time longer than the current time stamp into a queue to be sent.

In this embodiment, the zset queue is scanned periodically by the lua script to scan the sequencing result, and the message with the score value greater than the current timestamp is fetched and placed in the queue to be sent.

S160, regularly taking out the corresponding message from the queue to be sent, and submitting the message to a designated delay message processing object;

s170, updating the message sending state in the database.

Examples are as follows: all participants were alerted 15 minutes before the meeting start time. The current time stamp is T0, a meeting is reserved and initiated in the schedule management module, the meeting starting time is set at a certain time T2 after two days, the time stamp T1 represented by 15 minutes before the meeting is started is set, and a message is sent to remind all participants. Firstly, the schedule management module needs to call a message creation interface of the delay message service, package data such as message return time (T1), message content, unique service ID (B1), message ID (M1) and the like into data (D) and push the data to the delay message service. The deferred message service first persists the data D, stores the data to be returned in the Redis cache for a short period of time, or else stores the data in the MySQL database, in this case the data D will be stored in the MySQL database first. And (3) through a time round algorithm, turning a time slot, when the time slot wheel turns to a T1 interval, taking out the message data D from the MySQL database, writing the message data D into a Redis queue, submitting the data to a high-performance MQ cluster in the form of a message through an ack-retry mechanism, and finally sending the data D back to the schedule management module.

The processing flow after D is taken out by a time round algorithm is as follows: the data D is written into the waiting queue of the zset type ordered queue by the MySQL database, the score value is T1, the value M1 is the score value of the value, the data with the score value smaller than or equal to the current timestamp is migrated from the waiting queue to the sending queue by the lua script through the score value of the waiting queue, the atomicity of the lua script ensures the transactional property of the data transferred between the two queues, and the data D is migrated from the waiting queue to the sending queue. Then migrating the data D from the sending queue to the uncack queue through the lua script, submitting the data D to the high-performance MQ cluster, marking the data as ack if the submitting is successful, removing the data D from the uncack queue, and ending the flow; and migrating the data D from the uncack queue to the retry queue through the lua script after the delivery failure, and re-delivering the data D until the data D is successful in a timed task compensation mode. After the data D is submitted to the MQ, the data D is ensured to be finally processed and not repeatedly submitted through a consumption confirmation mechanism and a success callback mechanism of the MQ.

The delayed message service successfully delivers data D in the form of a message to the MQ. The MQ pushes the data D to the schedule management module, and the MQ ensures that the final processing of the data D is completed through an ack mechanism and a consumption retry mechanism.

And finally, notifying the delayed message service through a successful callback mechanism of the MQ, and changing the processing state of the data D into processing success.

The delayed message service filters invalid data through a maximum timestamp screening mechanism and is used for processing the situation that the data is changed, as shown in fig. 10, when the data D is submitted to the MQ in the fourth step, through M1 and B1, whether the message sending back timestamp of the data D is the maximum timestamp is judged in the database, if the data with the larger timestamp exists, it is indicated that the data D is changed after the creation is successful, and at this time, the data D corresponding to the current timestamp should be discarded without processing based on the data which is changed last time.

In this embodiment, the above steps are performed as a server, which carries the delayed message management service therein.

According to the delay message processing method, the messages are stored in the database, the messages meeting the requirements are written into the redis queue, the messages are cached by utilizing the zset queue, the messages with the time longer than the current time stamp are sequenced according to the expected sending time, and the messages with the time longer than the current time stamp are placed into the queue to be sent for sending, so that the problems that the applicable scene of the Rabbittmq delay message is single, the use is complex, the occupied memory is too high and the like are solved.

Fig. 6 is a schematic block diagram of a delayed message processing apparatus 300 according to an embodiment of the present application. As shown in fig. 6, the present application also provides a delayed message processing apparatus 300 corresponding to the above delayed message processing method. The delayed message processing apparatus 300 includes means for performing the delayed message processing method described above, and may be configured in a server. Specifically, referring to fig. 6, the delayed message processing apparatus 300 includes a message acquisition unit 301, a writing unit 302, a specifying unit 303, a sorting unit 304, a scanning unit 305, a fetching unit 306, and an updating unit 307.

A message obtaining unit 301, configured to obtain a message generated by a service function; a writing unit 302, configured to store the message in a database, and write a message meeting the requirement in a redis queue; a specifying unit 303 for specifying a delay time and a message delivery object; a sorting unit 304, configured to sort the messages in the redis queue according to the expected sending time, so as to obtain a sorting result; a scanning unit 305, configured to periodically scan the ordering result, and put the message with the predicted sending time greater than the current timestamp into a queue to be sent; a retrieving unit 306, configured to periodically retrieve a corresponding message from the queue to be sent, and submit the message to a specified delay message processing object; an updating unit 307 for updating the messaging status in the database.

In one embodiment, as shown in fig. 7, the writing unit 302 includes a storage subunit 3021 and a schedule management subunit 3022.

A storage subunit 3021, configured to store the message into a database; the scheduling management subunit 3022 is configured to perform message scheduling management by using a time-carousel algorithm, and write a message meeting the requirement into the redis queue.

In one embodiment, as shown in fig. 8, the sorting unit 304 includes a cache queue forming sub-unit 3041 and a message sorting sub-unit 3042.

A cache queue forming subunit 3041, configured to use the zset queue of the redis queue as a cache queue; the message ordering sub-unit 3042 is configured to use the queuing characteristic of the score, take the predicted sending timestamp of the message as the score value, take the message content as the queue value, and order the message according to the score value by the queue, so as to obtain an ordering result.

In an embodiment, the scanning unit 305 is configured to periodically scan the zset queue through the lua script to scan the sequencing result, and take out the message with the score value greater than the current timestamp and put the message in the queue to be sent.

It should be noted that, as will be clearly understood by those skilled in the art, the specific implementation process of the delay message processing apparatus 300 and each unit may refer to the corresponding description in the foregoing method embodiment, and for convenience and brevity of description, the description is omitted here.

The above-described delayed message processing apparatus 300 may be implemented in the form of a computer program which is executable on a computer device as shown in fig. 9.

Referring to fig. 9, fig. 9 is a schematic block diagram of a computer device according to an embodiment of the present application. The computer device 500 may be a server, where the server may be a stand-alone server or may be a server cluster formed by a plurality of servers.

With reference to FIG. 9, the computer device 500 includes a processor 502, memory, and a network interface 505 connected by a system bus 501, where the memory may include a non-volatile storage medium 503 and an internal memory 504.

The non-volatile storage medium 503 may store an operating system 5031 and a computer program 5032. The computer program 5032 includes program instructions that, when executed, cause the processor 502 to perform a delayed message processing method.

The processor 502 is used to provide computing and control capabilities to support the operation of the overall computer device 500.

The internal memory 504 provides an environment for the execution of a computer program 5032 in the non-volatile storage medium 503, which computer program 5032, when executed by the processor 502, causes the processor 502 to perform a delayed message processing method.

The network interface 505 is used for network communication with other devices. It will be appreciated by those skilled in the art that the architecture shown in fig. 9 is merely a block diagram of some of the architecture relevant to the present inventive arrangements and is not limiting of the computer device 500 to which the present inventive arrangements may be implemented, as a particular computer device 500 may include more or fewer components than shown, or may combine some of the components, or have a different arrangement of components.

Wherein the processor 502 is configured to execute a computer program 5032 stored in a memory to implement the steps of:

acquiring a message generated by a service function; storing the information into a database, and writing the information meeting the requirements into a redis queue; ordering the messages in the redis queue according to the expected sending time to obtain an ordering result; periodically scanning the sequencing result, and putting the message with the predicted sending time longer than the current time stamp into a queue to be sent; the corresponding information is regularly fetched from the queue to be sent and submitted to the appointed delay information processing object; updating the messaging status within the database.

In one embodiment, after implementing the message step generated by the get service function, the processor 502 further implements the following steps:

a delay time and a message delivery object are specified.

In one embodiment, when the step of storing the message in the database and writing the message meeting the requirements into the redis queue is implemented by the processor 502, the following steps are specifically implemented:

storing the message in a database; and carrying out message scheduling management by using a time-roulette algorithm, and writing the messages meeting the requirements into a redis queue.

In one embodiment, when implementing the step of ordering the messages in the redis queue according to the expected sending time to obtain the ordering result, the processor 502 specifically implements the following steps:

using the zset queue of the redis queue as a cache queue; and using the queuing characteristic of the score, taking the predicted sending time stamp of the message as the score value, taking the message content as the queue value, and sequencing the message according to the score value by the queue to obtain a sequencing result.

In an embodiment, when the processor 502 performs the step of periodically scanning the ordering result and placing the message with the predicted sending time greater than the current timestamp into the queue to be sent, the following steps are specifically implemented:

It should be appreciated that in an embodiment of the application, the processor 502 may be a central processing unit (Central Processing Unit, CPU), the processor 502 may also be other general purpose processors, digital signal processors (Digital Signal Processor, DSPs), application specific integrated circuits (Application Specific Integrated Circuit, ASICs), off-the-shelf programmable gate arrays (Field-Programmable Gate Array, FPGAs) or other programmable logic devices, discrete gate or transistor logic devices, discrete hardware components, or the like. Wherein the general purpose processor may be a microprocessor or the processor may be any conventional processor or the like.

Those skilled in the art will appreciate that all or part of the flow in a method embodying the above described embodiments may be accomplished by computer programs instructing the relevant hardware. The computer program comprises program instructions, and the computer program can be stored in a storage medium, which is a computer readable storage medium. The program instructions are executed by at least one processor in the computer system to implement the flow steps of the embodiments of the method described above.

Accordingly, the present application also provides a storage medium. The storage medium may be a computer readable storage medium. The storage medium stores a computer program which, when executed by a processor, causes the processor to perform the steps of:

In an embodiment, after the step of executing the computer program to implement the step of obtaining the message generated by the service function, the processor further implements the steps of:

a delay time and a message delivery object are specified.

In one embodiment, when the processor executes the computer program to implement the steps of storing the message in a database and writing the message meeting the requirements into a redis queue, the steps are specifically implemented as follows:

In one embodiment, when the processor executes the computer program to implement the step of ordering the messages in the redis queue according to the expected sending time to obtain an ordering result, the following steps are specifically implemented:

In one embodiment, when the processor executes the computer program to realize the step of periodically scanning the sequencing result and put the message with the predicted sending time greater than the current timestamp into the queue to be sent, the method specifically comprises the following steps:

The storage medium may be a U-disk, a removable hard disk, a Read-Only Memory (ROM), a magnetic disk, or an optical disk, or other various computer-readable storage media that can store program codes.

Those of ordinary skill in the art will appreciate that the elements and algorithm steps described in connection with the embodiments disclosed herein may be embodied in electronic hardware, in computer software, or in a combination of the two, and that the elements and steps of the examples have been generally described in terms of function in the foregoing description to clearly illustrate the interchangeability of hardware and software. Whether such functionality is implemented as hardware or software depends upon the particular application and design constraints imposed on the solution. Skilled artisans may implement the described functionality in varying ways for each particular application, but such implementation decisions should not be interpreted as causing a departure from the scope of the present application.

In the several embodiments provided by the present application, it should be understood that the disclosed apparatus and method may be implemented in other manners. For example, the device embodiments described above are merely illustrative. For example, the division of each unit is only one logic function division, and there may be another division manner in actual implementation. For example, multiple units or components may be combined or may be integrated into another system, or some features may be omitted, or not performed.

The steps in the method of the embodiment of the application can be sequentially adjusted, combined and deleted according to actual needs. The units in the device of the embodiment of the application can be combined, divided and deleted according to actual needs. In addition, each functional unit in the embodiments of the present application may be integrated in one processing unit, or each unit may exist alone physically, or two or more units may be integrated in one unit.

The integrated unit may be stored in a storage medium if implemented in the form of a software functional unit and sold or used as a stand-alone product. Based on such understanding, the technical solution of the present application is essentially or a part contributing to the prior art, or all or part of the technical solution may be embodied in the form of a software product stored in a storage medium, comprising several instructions for causing a computer device (which may be a personal computer, a terminal, a network device, etc.) to perform all or part of the steps of the method according to the embodiments of the present application.

While the application has been described with reference to certain preferred embodiments, it will be understood by those skilled in the art that various changes and substitutions of equivalents may be made and equivalents will be apparent to those skilled in the art without departing from the scope of the application. Therefore, the protection scope of the application is subject to the protection scope of the claims.

Claims

1. A method of deferred message processing, comprising:

acquiring a message generated by a service function;

updating the messaging status within the database.

2. The method for processing delayed messages according to claim 1, further comprising, after said obtaining the message generated by the service function:

a delay time and a message delivery object are specified.

3. The method of claim 1, wherein storing the message in a database and writing the satisfactory message in a redis queue comprises:

storing the message in a database;

4. The method for processing delayed messages according to claim 1, wherein said ordering the messages in said redis queue according to the expected transmission time to obtain an ordering result comprises:

using the zset queue of the redis queue as a cache queue;

5. The method for processing delayed messages according to claim 1, wherein said periodically scanning said ordering result, placing said message with an estimated time of transmission greater than a current time stamp into a queue to be transmitted, comprises:

6. A delay message processing apparatus, comprising:

7. The delayed message processing apparatus of claim 6, further comprising:

8. The delay message processing apparatus of claim 6, wherein the writing unit comprises:

a storage subunit configured to store the message into a database;

9. A computer device, characterized in that it comprises a memory on which a computer program is stored and a processor which, when executing the computer program, implements the method according to any of claims 1-5.

10. A storage medium storing a computer program which, when executed by a processor, implements the method of any one of claims 1 to 5.