CN102023974B - A method, device and system for processing messages through a message queue - Google Patents

Abstract

The embodiment of the present invention discloses a method for processing messages through a message queue, which includes the following steps: receiving a message from a message production device, sending the message to the first message queue in the memory, and The message is added to the database; the message in the database is sent to the second message queue in the memory, and the second message queue is used to store the message obtained and processed by the message using device; receive the message Using the message processed by the device, sending the message to the first message queue, and deleting or modifying the processed message from the database according to the flag bit of the processed message. The embodiment of the present invention takes both performance and reliability of the message queue into consideration. The embodiment of the present invention also discloses a device and a system for applying the above method.

Description

A method, device and system for processing messages through a message queue

technical field

The invention relates to the technical field of the Internet, in particular to a method, device and system for processing messages through a message queue.

Background technique

In data business technology, a message refers to a data unit transmitted between two computers, which can be very simple, such as containing only text strings, or more complex, possibly containing embedded objects. A message queue is a container for storing messages during message transmission. It is a message processing technology for exchanging information between distributed applications. It can reside in memory or on disk. It is mainly used to provide routing and ensure message delivery. An application provides message processing and message queuing functionality, regardless of whether the computer on which the application is running is on the same network or online at the same time. If the recipient is unavailable when the message is sent, the message queue will hold the message until it is read by the application. Through the message queue, the application program can independently access the message without knowing the positional relationship between the messages, and the message queue does not need to wait for the receiving program to receive the message before continuing to perform subsequent steps.

The message queue provides a loosely coupled method for constructing distributed applications implemented in a synchronous or asynchronous manner. The API (Application Programming Interface, Application Programming Interface) call of the message queue is embedded in a new or existing application and sent to In-memory or disk-based message queues. Message queues can be used in applications to perform functions such as requesting services, exchanging information, or asynchronous processing.

The criteria to measure whether the message queue is excellent include performance and reliability. The message queue used in large-scale distributed systems has a huge amount of messages. Therefore, the performance requirements for the message queue are very high, and high concurrency and high throughput are required. In order to meet the requirements of high efficiency, the existing technology usually uses the memory message queue, and the message is stored in the memory to ensure the message processing speed; in order to achieve reliability, some message queues use the memory message queue + file system to realize the message Persistence, or completely use the database to implement the message queue; there are also some message queues that use the memory message queue + database to achieve message persistence, but do not use a trigger method to achieve message synchronization between the message queue and the database.

In the process of realizing the present invention, the inventor finds that the prior art has at least the following defects:

The memory message queue reduces the reliability of the system. In case of power failure, unexpected crash, or program crash, the integrity of the message cannot be guaranteed, and a large number of messages that cannot be processed in the future may be lost; while using the memory message queue + file system or When the memory message queue + database method is used to achieve message persistence, the efficiency is very low, and the implementation is complicated and difficult to use.

Contents of the invention

Embodiments of the present invention provide a method, device and system for processing messages through a message queue, which are used to balance the performance and reliability of the message queue.

An embodiment of the present invention provides a method for processing messages through a message queue, including the following steps:

Receive a message from the message production device, send the message to the first message queue in the memory, and add the message to the database according to the flag bit in the message;

Sending the messages in the database to a second message queue in memory, where the second message queue is used to store messages acquired and processed by the message using device;

receiving the message processed by the message usage device, sending the message to the first message queue, and deleting or modifying the processed message from the database according to the tag bit of the processed message .

The embodiment of the present invention also provides a message queue server, including:

The transceiver module is configured to receive a message from a message production device, send the message to the first message queue in the memory, receive a message processed by the message usage device, and send the processed message to the first message queue. in the message queue;

A database module, configured to store messages, and send the messages to the second message queue in memory;

An adding module, connected to the database module and the transceiver module, configured to add the message to the message according to the flag bit of the message from the message production device sent by the transceiver module to the first message queue In the database module;

A modifying module, connected to the transceiver module and the database module, configured to convert the processed The message is deleted from the database module or the processed message is modified in the database module.

The embodiment of the present invention also provides a system for processing messages through a message queue, including a message production device, a message queue server, and a message usage device,

The message production device is configured to generate a message and send the message to the message queue server;

The message queue server is configured to receive the message from the message production device, send the message to the first message queue in the memory, and add the message to the database according to the flag bit in the message; Send the message in the database to the second message queue in the memory, so that the message using device acquires and processes the message; receives the message processed by the message using device, and sends the message to the first message in the queue, and delete or modify the message from the database according to the flag bit of the processed message;

The message usage device is configured to acquire and process messages from the second message queue, add a flag to the processed messages, and send them to the first message queue.

Compared with the prior art, the embodiment of the present invention has the following advantages: the advantages of high concurrency and high throughput of the memory message queue are used to improve the efficiency of processing messages, ensure the reliability of data, and fully retain the transactional nature of the database Advantages: In addition to completing the distribution and delivery of messages, the processing results of messages are also retained, which can be used for other purposes at any time; since the messages are stored in the database in real time, and the database can easily perform various operations, which is convenient for tracking the current usage of the message queue , fault detection and optimization.

Description of drawings

In order to more clearly illustrate the technical solutions in the embodiments of the present invention or the prior art, the following will briefly introduce the drawings that need to be used in the description of the embodiments of the present invention or the prior art. Obviously, the accompanying drawings in the following description These are only some embodiments of the present invention, and those skilled in the art can also obtain other drawings based on these drawings without any creative effort.

FIG. 1 is a flowchart of a method for processing messages through a message queue in an embodiment of the present invention;

FIG. 2 is a schematic structural diagram of a system for processing messages through a message queue in an application scenario of an embodiment of the present invention;

FIG. 3 is a flow chart of processing messages through a message queue in an application scenario of an embodiment of the present invention;

FIG. 4 is a schematic structural diagram of a message queue server in an embodiment of the present invention;

FIG. 5 is a schematic structural diagram of a message queue server in an application scenario according to an embodiment of the present invention;

FIG. 6 is a schematic structural diagram of a system for processing messages through a message queue in an embodiment of the present invention.

Detailed ways

In the technical solution provided by the embodiment of the present invention, the synchronization between the database and the message queue is realized by triggering, and the high performance of the memory message queue and the high reliability of the database are used to combine the advantages of the memory message queue and the data security to realize A message queue using a database as a persistent tool is developed, which solves the problems of performance and reliability.

The following will clearly and completely describe the technical solutions of the embodiments of the present invention with reference to the accompanying drawings in the embodiments of the present invention. Obviously, the described embodiments are some of the embodiments of the present invention, but not all of them. Based on the embodiments of the present invention, all other embodiments obtained by persons of ordinary skill in the art without making creative efforts belong to the protection scope of the present invention.

As shown in Figure 1, it is a flow chart of a method for processing messages through a message queue in an embodiment of the present invention, including the following steps:

Step 101, receiving a message from a message production device, sending the message to a first message queue in memory, and adding the message to a database according to a flag bit in the message.

Step 102, sending the messages in the database to the second message queue in the memory.

Wherein, the second message queue is used to store messages acquired and processed by the message using device.

Step 103, receiving the message processed by the message usage device, sending the message to the first message queue, and deleting or modifying the processed message from the database according to the tag bit of the processed message.

As shown in FIG. 2, it is a schematic structural diagram of a system for processing messages through a message queue in the application scenario of the embodiment of the present invention. The system includes a message production device 100, a first message queue 200, a transaction processing service program 300, a database, and a database. A trigger 400 , a trigger external process 500 , a second message queue 600 and a message consuming device 700 . Wherein, the first message queue 200, the transaction processing service program 300, the database 400 (including the database trigger), the trigger external process 500 and the second message queue 600 form a complete trigger database persistent message queue server 800, which is a message The sending device 100 and the message using device 700 provide a message exchange service.

The method for processing messages through the message queue in the embodiment of the present invention will be described in detail below in conjunction with the above application scenarios.

As shown in Figure 3, it is a flow chart of processing messages through message queues in the application scenario of the embodiment of the present invention, specifically including the following steps:

Step 301, create a first message queue 200.

Specifically, a memory message queue is created in the message queue server 800 as the first message queue 200, and the first message queue 200 is used to receive messages from each message production device 100 and transmit the messages to the transaction processing service program 200 .

Step 302 , the message production device 100 generates a message, and sends the message to the first message queue 200 .

Wherein, the message carries a flag bit, which is used to indicate how to process the message, including adding, deleting, and modifying the message to the database. Since the message has just been generated by the message producing device 100 and has not been processed by the message consuming device 700 , the flag bit carried in the message indicates that the message should be added to the database 400 .

Step 303, start the transaction processing service program 300.

Step 304 , the transaction processing service program 300 obtains the message in the first message queue 200 , and adds the message to the database 400 according to the insertion tag carried in the message.

Step 305 , creating a database trigger in the database 400 .

Step 306, create a trigger external process 500.

Specifically, establishing the trigger external process 500 includes establishing the external trigger external process 500 according to the external process specification of the database system, and loading the trigger external process 500 for the database system according to the database system specification. The trigger external process 500 is used to be triggered to start working when the message in the database 400 increases, and to send the newly added message to the memory.

Step 307 , when an add operation occurs to a message in the database, the database trigger is triggered, and the message is sent to the trigger external process 500 .

In step 308, the second message queue 600 is established.

Specifically, an in-memory message queue is created on the message queue server 800 as the second message queue 600 , and the second message queue 600 is used to receive messages from the trigger external process 500 and transmit the messages to the message consuming device 700 .

Step 309 , the trigger external process 500 puts the message into the second message queue 600 .

Step 310 , the message usage device 700 obtains and processes messages from the second message queue 600 , adds a flag to the processed messages, and sends them to the first message queue 200 .

Step 311 , the transaction processing service program 300 obtains the message in the first message queue 200 , and deletes or modifies the message in the database 400 according to the modification flag carried in the message.

Specifically, since the message has been processed by the message consuming device 700 , the modification mark carried by the message deletes the message from the database 400 or modifies the message in the database 100 .

It should be noted that, in the method of the embodiment of the present invention, the order of each step may be adjusted according to actual needs.

In the technical solution provided by the embodiment of the present invention, the first message queue 200 sent to after the message is generated and the second message queue 600 sent to after the trigger are both memory message queues, therefore, the processing speed is extremely fast, ensuring the smoothness of message processing Efficiency: After the message enters the first message queue 200, it is put into the database 400 immediately, and the message is reliably saved. For the message production device 100 and the message consumption device 700, the message queue server 800 is equivalent to a common memory message queue, and the difference from the common memory message queue is that after the message consumption device 700 completes the message operation, it needs to send The first message queue 200 sends a processed message to delete or modify the message recorded in the database 400, that is, the message will be deleted or modified only after the message using device 700 correctly receives and processes the message, and clearly issues an operation instruction. , therefore, the messages in the database will not be lost due to reasons such as server power failure or unexpected crash.

Embodiments of the present invention provide a method and application scenarios for processing messages through a message queue in the above implementation manners. Correspondingly, embodiments of the present invention also provide an apparatus and system for applying the above method for processing messages through a message queue.

As shown in Figure 4, it is a schematic structural diagram of a message queue server in an embodiment of the present invention, including:

The transceiver module 410 is configured to receive a message from a message producing device, send the message to the first message queue in memory, receive a message processed by the message using device, and send the processed message to the first message queue. in a message queue.

The database module 420 is configured to store messages and send the messages to the second message queue in memory.

The adding module 430 is connected with the database module 420 and the transceiver module 410, and is used to add the message to the database module according to the flag bit of the message from the message production device sent by the transceiver module 410 to the first message queue 420 in.

The modification module 440 is connected with the transceiver module 410 and the database module 420, and is configured to convert the processed message to The above-mentioned processed messages are deleted from the database module 420 or modified in the database module 420 .

As shown in Figure 5, it is a schematic structural diagram of the message queue server in the application scenario of the embodiment of the present invention, specifically including a creation module 510, a database module 520, a transceiver module 530, an addition module 540 and a modification module 550, wherein,

The creation module 510 is connected with the database module 520 and the transceiver module 530, and is used to create the first message queue and the second message queue, which are respectively used by the transceiver module 530 and the database module 520.

The database module 520 is configured to store messages and send the messages to the second message queue in memory.

The database module 520 is specifically configured to send the message to the trigger external process through the database trigger, and send the message to the second message queue through the trigger external process.

The transceiver module 530 is configured to receive a message from a message production device, send the message to the first message queue in the memory, receive a message processed by the message consuming device, and send the processed message to the in the first message queue.

The adding module 540 is connected with the database module 520 and the transceiver module 530, and is used to add the message to the database module according to the flag bit of the message from the message production device sent by the transceiver module 530 to the first message queue 520 in.

The adding module 540 is specifically configured to add the message in the first message queue to the database module 520 according to the insertion marker carried by the message in the first message queue.

The modification module 550 is connected with the transceiver module 530 and the database module 520, and is configured to convert the processed message to The above-mentioned processed messages are deleted from the database module 520 or modified in the database module 520 .

The embodiment of the present invention makes full use of the advantages of high concurrency and high throughput of the memory message queue, improves the efficiency of message processing, ensures the reliability of data, and fully retains the transactional advantages of the database. In addition to completing the distribution and delivery of messages , also retains the processing results of the message, which can be used for other purposes at any time; because the message is stored in the database in real time, and the database can easily perform various operations, which is convenient for tracking the current usage of the message queue, finding faults and optimizing.

As shown in FIG. 6, it is a schematic structural diagram of a system for processing messages through a message queue in an embodiment of the present invention, including a message production device 610, a message queue server 620, and a message usage device 630, wherein,

The message production device 610 is configured to generate a message and send the message to the message queue server 620 .

The message queue server 620 is configured to receive the message from the message production device 610, send the message to the first message queue in the memory, and add the message to the database according to the flag bit in the message; Send the message in the database to the second message queue in the memory, make the message using device 630 acquire and process the message; receive the message processed by the message using device, and send the message to the first message queue, and delete or modify the message from the database according to the flag bit of the processed message;

The message using device 630 is configured to obtain and process messages from the second message queue, add a flag to the processed messages, and send them to the first message queue.

The above-mentioned message queue server 620 is further configured to create an in-memory message queue, and use the in-memory message queue as the first message queue.

The above-mentioned message queue server 620 is further configured to start a transaction processing service program, and the transaction processing service program is used to add the message to the database according to the tag bit carried by the message in the first message queue.

The above-mentioned message queue server 620 is further configured to start a transaction processing service program, and the transaction processing service program is used to delete or modify messages in the database according to the flag bit carried by the messages in the first message queue.

The above-mentioned message queue server 620 is further configured to establish a database trigger in the database, and establish an external process of the trigger, and the database trigger is used to send messages in the database to the external process of the trigger. A trigger external procedure is used to send messages from the database trigger to the second message queue.

The above-mentioned message queue server 620 is further configured to create an in-memory message queue, and use the in-memory message queue as the second message queue.

The embodiments of the present invention make full use of the advantages of high concurrency and high throughput of memory message queues, and the efficiency far exceeds that of using file system persistence or message queues implemented solely on the basis of databases. In internal implementation, transaction processing service programs and databases The connection pool is used among them to reduce database concurrency and make full use of database performance. It adopts trigger operation, which is more efficient than traditional polling or other methods, and the CPU utilization rate is lower; using database as a persistent medium makes full use of Utilizing the reliability advantages of the database, by using the backup and recovery function carried by the database itself, the reliability of the data is guaranteed; the transactional advantages of the database are completely preserved, and in addition to completing the distribution and delivery of messages, it also retains the processing results of messages. It can be used for other purposes at any time; because the messages are stored in the database in real time, and the database can be easily used for various operations, which is convenient for tracking the current usage of the message queue, finding faults and optimizing them.

Through the description of the above embodiments, those skilled in the art can clearly understand that the present invention can be implemented by means of software plus a necessary general-purpose hardware platform, and of course also by hardware, but in many cases the former is a better implementation Way. Based on this understanding, the essence of the technical solution of the present invention or the part that contributes to the prior art can be embodied in the form of a software product. The computer software product is stored in a storage medium and includes several instructions to make a A terminal device (which may be a mobile phone, a personal computer, a server, or a network device, etc.) executes the methods described in various embodiments of the present invention.

The above is only a preferred embodiment of the present invention, it should be pointed out that, for those of ordinary skill in the art, without departing from the principle of the present invention, some improvements and modifications can also be made, and these improvements and modifications can also be made. It should be regarded as the protection scope of the present invention.

Those skilled in the art can understand that the modules in the device in the embodiment can be distributed in the device in the embodiment according to the description in the embodiment, or can be located in one or more devices different from the embodiment according to corresponding changes. The modules in the above embodiments can be integrated or deployed separately; they can be combined into one module, or further split into multiple sub-modules.

The serial numbers of the above embodiments of the present invention are for description only, and do not represent the advantages and disadvantages of the embodiments.

The above disclosures are only a few specific embodiments of the present invention, however, the present invention is not limited thereto, and any changes conceivable by those skilled in the art shall fall within the protection scope of the present invention.

Claims (10)

1. A method for processing messages by a message queue, characterized in that, comprising: Receive a message from the message production device, send the message to the first message queue in the memory, and add the message to the database according to the flag bit in the message; Sending the messages in the database to a second message queue in memory, where the second message queue is used to store messages acquired and processed by the message using device; receiving the message processed by the message usage device, sending the message to the first message queue, and deleting or modifying the processed message from the database according to the tag bit of the processed message . 2. The method according to claim 1, wherein the message is added to the database according to the tag position in the message, specifically: Add the message to the database according to the insertion mark carried by the message in the first message queue. 3. The method according to claim 1, characterized in that, deleting or revising the processed message from the database according to the tag position of the processed message, specifically: According to the modification flag carried by the processed message, the processed message is deleted from the database or the processed message is modified in the database. 4. The method according to claim 1, wherein the sending of the message in the database to the second message queue in the internal memory is specifically: A message in the database is sent to a trigger external procedure by a database trigger in the database, and the message is sent to the second message queue by the trigger external procedure. 5. The method according to claim 1, characterized in that before receiving the message processed by the device, further comprising: The message usage device acquires and processes messages from the second message queue, adds a flag to the processed messages, and sends them to the first message queue. 6. A message queue server, characterized in that, comprising: The transceiver module is configured to receive a message from a message production device, send the message to the first message queue in the memory, receive a message processed by the message usage device, and send the processed message to the first message queue. in the message queue; A database module, configured to store messages, and send the messages to the second message queue in memory; An adding module, connected to the database module and the transceiver module, configured to add the message to the message according to the flag bit of the message from the message production device sent by the transceiver module to the first message queue In the database module; A modifying module, connected to the transceiver module and the database module, configured to convert the processed The message is deleted from the database module or the processed message is modified in the database module. 7. message queue server as claimed in claim 6, is characterized in that, The adding module is specifically configured to add the message in the first message queue to the database module according to the insertion marker carried by the message in the first message queue. 8. The message queue server according to claim 6, further comprising: A creation module, connected to the transceiver module, is used to create the first message queue and the second message queue, which are respectively used by the transceiver module and the database module. 9. message queue server as claimed in claim 6, is characterized in that, The database module is specifically configured to send the message to the trigger external process through the database trigger, and send the message to the second message queue through the trigger external process. 10. A system for processing messages through a message queue, comprising a message production device, a message queue server and a message usage device, characterized in that, The message production device is configured to generate a message and send the message to the message queue server; The message queue server is configured to receive the message from the message production device, send the message to the first message queue in the memory, and add the message to the database according to the flag bit in the message; Send the message in the database to the second message queue in the memory, so that the message using device acquires and processes the message; receives the message processed by the message using device, and sends the message to the first message in the queue, and delete or modify the message from the database according to the flag bit of the processed message; The message usage device is configured to acquire and process messages from the second message queue, add a flag to the processed messages, and send them to the first message queue.

Images