CN112822091B - Message processing method and device - Google Patents
Message processing method and device Download PDFInfo
- Publication number
- CN112822091B CN112822091B CN201911125238.6A CN201911125238A CN112822091B CN 112822091 B CN112822091 B CN 112822091B CN 201911125238 A CN201911125238 A CN 201911125238A CN 112822091 B CN112822091 B CN 112822091B
- Authority
- CN
- China
- Prior art keywords
- message
- transaction
- state
- preparation
- cache queue
- Prior art date
- Legal status (The legal status is an assumption and is not a legal conclusion. Google has not performed a legal analysis and makes no representation as to the accuracy of the status listed.)
- Active
Links
- 238000003672 processing method Methods 0.000 title abstract description 10
- 238000002360 preparation method Methods 0.000 claims abstract description 96
- 238000000034 method Methods 0.000 claims abstract description 37
- 230000008569 process Effects 0.000 claims abstract description 11
- 238000003860 storage Methods 0.000 claims description 9
- 238000004590 computer program Methods 0.000 claims description 6
- 238000011084 recovery Methods 0.000 claims description 3
- 238000010586 diagram Methods 0.000 description 7
- 230000007246 mechanism Effects 0.000 description 7
- 238000004891 communication Methods 0.000 description 5
- 230000004044 response Effects 0.000 description 5
- 238000005516 engineering process Methods 0.000 description 3
- 238000004519 manufacturing process Methods 0.000 description 3
- 230000003287 optical effect Effects 0.000 description 2
- 230000002159 abnormal effect Effects 0.000 description 1
- 230000005540 biological transmission Effects 0.000 description 1
- 230000006872 improvement Effects 0.000 description 1
- 230000010365 information processing Effects 0.000 description 1
- 230000003993 interaction Effects 0.000 description 1
- 230000004048 modification Effects 0.000 description 1
- 238000012986 modification Methods 0.000 description 1
Images
Classifications
-
- H—ELECTRICITY
- H04—ELECTRIC COMMUNICATION TECHNIQUE
- H04L—TRANSMISSION OF DIGITAL INFORMATION, e.g. TELEGRAPHIC COMMUNICATION
- H04L51/00—User-to-user messaging in packet-switching networks, transmitted according to store-and-forward or real-time protocols, e.g. e-mail
- H04L51/07—User-to-user messaging in packet-switching networks, transmitted according to store-and-forward or real-time protocols, e.g. e-mail characterised by the inclusion of specific contents
- H04L51/18—Commands or executable codes
-
- H—ELECTRICITY
- H04—ELECTRIC COMMUNICATION TECHNIQUE
- H04L—TRANSMISSION OF DIGITAL INFORMATION, e.g. TELEGRAPHIC COMMUNICATION
- H04L67/00—Network arrangements or protocols for supporting network services or applications
- H04L67/50—Network services
- H04L67/56—Provisioning of proxy services
- H04L67/568—Storing data temporarily at an intermediate stage, e.g. caching
- H04L67/5682—Policies or rules for updating, deleting or replacing the stored data
Landscapes
- Engineering & Computer Science (AREA)
- Computer Networks & Wireless Communication (AREA)
- Signal Processing (AREA)
- Information Retrieval, Db Structures And Fs Structures Therefor (AREA)
- Computer And Data Communications (AREA)
- Debugging And Monitoring (AREA)
Abstract
The application provides a message processing method, which comprises the following steps: scanning the prepared message in the cache queue; if the fact that the overtime preparation message exists is determined, inquiring a transaction state corresponding to the preparation message from the generating equipment; the transaction state is recorded when the generation device processes the local transaction after sending the transaction message; when the queried transaction state is determined to be a successful state, writing the submitted message into a log database; creating an index for the transaction, obtaining the transaction to be consumed by the equipment, and deleting the preparation message in the cache queue; when the queried transaction state is determined to be a failure state, writing a rollback message into a log database; and delete the ready message in the cache queue. The method can keep the transaction states on the generating device and the intermediate device consistent.
Description
Technical Field
The present invention relates to the field of information processing technologies, and in particular, to a method and an apparatus for processing a message.
Background
At present, data interaction is carried out among different applications, in order to achieve the purposes of application decoupling and the like, a message middleware is introduced, an application A is deployed in a first machine room and serves as a message producer, and an application B is deployed in a second machine room and serves as a message consumer. The producer stores the message in the BROKER of the message middleware, and the consumer obtains the message from the BROKER to execute corresponding business logic.
On the one hand, the sender needs to be able to store messages to the token quickly and reliably, and on the other hand, it is desirable to support transaction characteristics and local transaction combinations. The transaction at the BROKER commits if the local transaction is successful, and the transaction at the BROKER rolls back if the local transaction fails.
ROKETMQ provides DB-based transaction message functionality. By writing the transaction message sent by the client first into the DB if the transaction is committed the message is made visible to the consumer and if the transaction rolls back the message is deleted. This implementation meets the user's needs for transaction messages to some extent.
However, the above implementation does not have a perfect transaction compensation mechanism: when a local transaction succeeds, when the transaction is submitted to the MQ, a situation of inconsistent state can be caused if a failure or other scene occurs. The method comprises the following steps: the success of the local transaction is unknown, and the MQ transaction can only execute some default operations to terminate the transaction finally, and the default operations can be commit or rollback, which causes the transaction to be inconsistent.
Disclosure of Invention
In view of this, the present application provides a message processing method and apparatus that can keep the transaction states on the generating device and the intermediate device consistent.
In order to solve the technical problems, the technical scheme of the application is realized as follows:
a message processing method, the method comprising:
scanning the prepared message in the cache queue;
if the fact that the overtime preparation message exists is determined, inquiring a transaction state corresponding to the preparation message from the generating equipment; the transaction state is recorded when the generation device processes the local transaction after sending the transaction message;
when the queried transaction state is determined to be a successful state, writing the submitted message into a log database; creating an index for the transaction, obtaining the transaction to be consumed by the equipment, and deleting the preparation message in the cache queue;
when the queried transaction state is determined to be a failure state, writing a rollback message into a log database; and delete the ready message in the cache queue.
A message processing apparatus, the apparatus comprising: the device comprises a scanning unit, an acquisition unit, a determination unit and a processing unit;
the scanning unit is used for scanning the preparation messages in the cache queue;
the acquisition unit is used for inquiring the transaction state corresponding to the preparation message from the generating equipment if the overtime preparation message exists in the preparation message scanned by the scanning unit; the transaction state is recorded when the generation device processes the local transaction after sending the transaction message;
the determining unit is used for determining whether the transaction state acquired by the acquiring unit is a success state or a failure state;
the processing unit is used for writing the submitted information into the log database when the determining unit determines that the queried transaction state is a successful state; creating an index for the transaction, obtaining the transaction to be consumed by the equipment, and deleting the preparation message in the cache queue; when the determining unit determines that the queried transaction state is a failure state, writing a rollback message into a log database; and delete the ready message in the cache queue.
An electronic device comprising a memory, a processor and a computer program stored on the memory and executable on the processor, the processor implementing the steps of the message processing method when the program is executed.
A computer readable storage medium having stored thereon a computer program which when executed by a processor realizes the steps of the message processing method.
According to the technical scheme, the processing state of the corresponding transaction is obtained by performing overtime processing on the preparation message in the cache queue, and the corresponding transaction message is processed according to the processing state, so that the transaction compensation mechanism is perfected, and the consistency of the transaction states on the generating device and the intermediate device can be maintained.
Drawings
FIG. 1 is a schematic diagram of a message processing flow in a first embodiment of the present application;
fig. 2 is a schematic diagram of a message processing flow in a second embodiment of the present application;
FIG. 3 is a schematic view of a device structure applied to the above technology in an embodiment of the present application;
fig. 4 is a schematic diagram of an entity structure of an electronic device according to an embodiment of the present invention.
Detailed Description
In order to make the objects, technical solutions and advantages of the present invention more apparent, the technical solutions of the present invention will be described in detail below by referring to the accompanying drawings and examples.
According to the message processing method, the processing state of the corresponding transaction is obtained by performing overtime processing on the preparation (preparation) message in the cache queue, and the corresponding transaction message is processed according to the processing state, so that a transaction compensation mechanism is perfected, and the consistency of the transaction states on the generating equipment and the intermediate equipment can be maintained.
The message processing system in the embodiment of the application comprises: generating equipment, intermediate equipment and consumption equipment; the production equipment is used as a producer of the message, the intermediate equipment is used as a temporary storage equipment of the message, the consumer is used as a consumer of the message, and the intermediate equipment is provided with a reader.
After the producer generates the message, the message is sent to the intermediate equipment, and the intermediate equipment stores the message temporarily; the consumption device obtains the message from the intermediate device to execute the corresponding business logic.
In the embodiment of the application, after sending the transaction message to the sender, the producer processes the local transaction, and when the local transaction is processed, not only sends a Commit (Commit) message or a rollback (rollback) message to the intermediate device according to the processing result; the processing result of the local processing transaction is recorded locally, namely the transaction state is recorded, and the transaction state comprises: success, failure.
In a particular implementation, it may be necessary to maintain multiple transactions whose transaction states are correspondingly recorded using the transaction ID of the intermediary response:
for example, transaction 1, the state is: success;
transaction 2, state is failed.
In specific implementation, the transaction ID and the transaction state may be recorded in a table manner or may be recorded in a log manner, and the method of recording the correspondence between the transaction ID and the transaction state is not limited in the embodiment of the present application.
The intermediate device sequentially reads and writes the messages sent by the received production device; and copy the data stored locally to the backup device.
The problem of slow writing performance is solved by adding the system image file to the local file sequence storage, so that the writing of a single brooker transaction message into TPS can reach 3W+, which is about 10 times of the prior scheme, and the performance can be improved by transversely expanding the number of the brookers for one topic. The copying of files among the brookers ensures the high availability of the system, and when one brooker has a problem, the system can be timely switched to the other brooker to continue to provide services.
The following is a conventional message processing procedure, that is, a processing procedure that the intermediate device normally receives the command message or the rollback message sent by the generating device, specifically:
first, the generating device sends a preparation message to the intermediate device.
Step two, when the intermediate equipment receives the preparation message, storing the preparation message into a cache queue, and writing the preparation message into a log database; and carries the transaction ID allocated thereto when transmitting a message responding to the preparation message to the generating device.
And thirdly, when the production equipment receives the response to the preparation message, sending the transaction message by using the transaction ID carried by the preparation message.
And fourthly, when the intermediate equipment receives the transaction information sent by the generating equipment, writing the transaction information into a log database, and sending a response message to the generating equipment.
And fifthly, after the generating device sends the transaction information, carrying out local transaction processing, marking the transaction processing result as the transaction state of the transaction, and sending a Commit message or a rollback message to the intermediate device according to the processing result.
When the processing result is successful, a Commit message is sent; and when the processing result is failure, sending a rollback message.
Sixthly, when receiving a Commit message sent by the generating device, the intermediate device writes the Commit message into a log database; creating an index for the transaction, obtaining the transaction by the consumption equipment, and deleting the preparation message in the cache queue;
when the rollback message is received, writing the rollback message into a log database; and deleting the preparation message in the cache queue.
An index is created for a transaction message, i.e., it can be retrieved by the consumer device for correlation.
To this end, a completed message processing flow is completed.
The following describes the transaction message processing procedure under abnormal conditions in the embodiment of the present application in detail with reference to the accompanying drawings:
example 1
In this embodiment, the case where the intermediate device does not receive the command message and the rollback message and the intermediate device operates normally is taken as an example.
Referring to fig. 1, fig. 1 is a schematic diagram of a message processing flow in a first embodiment of the present application. The method comprises the following specific steps:
Here, determining that the preparation message times out may be: by setting a timer for each preparation message in advance, when the timer times out, the preparation message is still in the cache queue, and the preparation message is determined to time out.
Inquiring the transaction state corresponding to the preparation message from the generating device specifically comprises the following steps:
the intermediate device sends a transaction state query request to the generating device; the query request carries the transaction ID of the corresponding transaction.
When the generating equipment receives the inquiry request, the transaction state corresponding to the locally stored transaction ID is acquired and responded to the intermediate equipment.
And when the intermediate device receives the response transaction state of the generating device, acquiring the transaction state.
The transaction state is a transaction state recorded by the generating device when the local transaction is finished after the transaction message is sent; including success and failure states.
104, when the intermediate device determines that the queried transaction state is a failure state, writing a rollback message into a log database; and deleting the preparation message in the cache queue.
No matter what the transaction state is, the corresponding message needs to be written into the log database, the corresponding preparation message is deleted, and for the transaction with the successful transaction state, an index is created for the transaction, wherein the created index is the position of each message in log, that is, the position for providing the transaction message for the consumer equipment.
The specific indexing and the process of obtaining the corresponding message by the consumer are the same as in the prior art and will not be described in detail here.
By the compensation mechanism, the consistency of the transaction state in the scene of failure of submitting the transaction can be ensured.
Example two
In this embodiment, the case where the intermediate device fails and the Commit message and the rollback message are not received is taken as an example.
Referring to fig. 2, fig. 2 is a schematic diagram of a message processing flow in a second embodiment of the present application. The method comprises the following specific steps:
If the intermediate equipment fails, the preparation message in the buffer queue is lost;
after the equipment is recovered from the fault, the preparation message corresponding to the transaction message which is not determined whether the transaction message can be acquired by the consumer equipment is required to be put into a cache queue.
Because all the corresponding messages are stored for each transaction in the log database, it can be judged which preparation message has no corresponding Commit message or roller ack message to determine whether the transaction related message is processed, and if not, the corresponding preparation message is required to be put into a cache queue for continuous processing; if there is a corresponding Commit message, or a rollback message, then the process is determined to end.
In step 202, the intermediate device queries the generating device for the transaction state corresponding to the preparation message.
Inquiring the transaction state corresponding to the preparation message from the generating device specifically comprises the following steps:
the intermediate device sends a transaction state query request to the generating device; the query request carries the transaction ID of the corresponding transaction.
When the generating equipment receives the inquiry request, the transaction state corresponding to the locally stored transaction ID is acquired and responded to the intermediate equipment.
And when the intermediate device receives the response transaction state of the generating device, acquiring the transaction state.
The transaction state is a transaction state recorded by the generating device when the local transaction is finished after the transaction message is sent; including success and failure states.
No matter what the transaction state is, the corresponding message needs to be written into the log database, the corresponding preparation message is deleted, and for the transaction with the successful transaction state, an index is created for the transaction, wherein the created index is the position of each message in log, that is, the position for providing the transaction message for the consumer equipment.
The specific indexing and the process of obtaining the corresponding message by the consumer are the same as in the prior art and will not be described in detail here.
By the aid of the compensation mechanism, the consistency of the transaction states in the scene of failure of submitting the transaction caused by failure recovery of the intermediate equipment can be ensured.
Based on the implementation, the intermediate device in the embodiment of the application can provide the capability of transaction information, reduce external dependence and ensure high availability of transmission; the performance of processing the transaction information by the BROKER of the intermediate equipment is improved, the sending delay is reduced, and the resource utilization rate is improved; and provides a sophisticated transaction compensation mechanism.
In summary, the present application perfects the transaction compensation mechanism by performing timeout processing on the preparation message in the cache queue, or storing the preparation message corresponding to the unprocessed message in the log database in the cache queue, and obtaining the processing state of the corresponding transaction, and processing the corresponding transaction message according to the processing state, so as to keep the consistency of the transaction states on the generating device and the intermediate device.
Based on the same inventive concept, a message processing device is also provided in the embodiments of the present application. Referring to fig. 3, fig. 3 is a schematic structural diagram of an apparatus to which the above technology is applied in the embodiment of the present application. The device comprises: a scanning unit 301, an acquisition unit 302, a determination unit 303, and a processing unit 304;
a scanning unit 301, configured to scan the preparation message in the buffer queue;
an obtaining unit 302, configured to query, from the generating device, a transaction state corresponding to a preparation message that has been scanned by the scanning unit 301 if it is determined that the preparation message has a timeout; the transaction state is recorded when the generation device processes the local transaction after sending the transaction message;
a determining unit 303, configured to determine whether the transaction state acquired by the acquiring unit 302 is a success state or a failure state;
a processing unit 304, configured to write the commit message into the log database when the determining unit 303 determines that the queried transaction state is a successful state; creating an index for the transaction, obtaining the transaction to be consumed by the equipment, and deleting the preparation message in the cache queue; when the determining unit 303 determines that the queried transaction state is a failure state, writing a rollback message into the log database; and delete the ready message in the cache queue.
Preferably, the apparatus further comprises: a receiving unit 305;
the receiving unit is used for receiving the preparation message sent by the generating equipment; receiving a transaction message sent by generating equipment;
a processing unit 304, configured to store, when the receiving unit receives a preparation message sent by a generating device, the preparation message in a buffer queue, and write the preparation message into a log database; when the receiving unit receives the transaction message sent by the generating device, the transaction message is written into a log database.
Preferably, the apparatus further comprises: a receiving unit 305;
a receiving unit 305, configured to receive a commit message sent by the generating device, and receive a rollback message sent by the generating device;
a processing unit 304, further configured to, when the receiving unit 305 receives a commit message sent by the generating device, write the commit message into the log database; creating an index for the transaction, obtaining the transaction to be consumed by the equipment, and deleting the preparation message in the cache queue; when receiving a rollback message sent by the generating equipment, writing the rollback message into a log database; and delete the ready message in the cache queue.
Preferably, the method comprises the steps of,
the obtaining unit 302 is further configured to obtain, from the log database, a preparation message corresponding to a transaction in which the rollback message and the commit message are not written after the device where the unit is located is recovered from a fault; and storing the data in a cache queue; inquiring the transaction state corresponding to the preparation message from the generating equipment;
a determining unit 303, configured to determine whether the transaction state acquired by the acquiring unit 302 is a success state or a failure state;
a processing unit 304, configured to write the commit message into the log database when the determining unit 303 determines that the queried transaction state is a successful state; creating an index for the transaction, obtaining the transaction to be consumed by the equipment, and deleting the preparation message in the cache queue; when the determining unit 303 determines that the queried transaction state is a failure state, writing a rollback message into the log database; and delete the ready message in the cache queue.
Preferably, the method comprises the steps of,
the processing unit 304 is further configured to sequentially read and write when the receiving unit 305 receives the message sent by the sender; the data stored locally is replicated and stored to the backup device.
The units of the above embodiments may be integrated or may be separately deployed; can be combined into one unit or further split into a plurality of sub-units.
In addition, an electronic device is provided in an embodiment of the present application, which includes a memory, a processor, and a computer program stored in the memory and capable of running on the processor, where the processor executes the program to implement the steps of the message processing method.
In addition, in an embodiment of the present application, there is further provided a computer readable storage medium having stored thereon a computer program, wherein the program when executed by a processor implements the steps of the message processing method.
Fig. 4 is a schematic diagram of an entity structure of an electronic device according to an embodiment of the present invention. As shown in fig. 4, the electronic device may include: processor 410, communication interface (Communications Interface) 420, memory 430 and communication bus 440, wherein processor 410, communication interface 420 and memory 430 communicate with each other via communication bus 440. Processor 510 may invoke logic instructions in memory 430 to perform the following method:
scanning the prepared message in the cache queue;
if the fact that the overtime preparation message exists is determined, inquiring a transaction state corresponding to the preparation message from the generating equipment; the transaction state is recorded when the generation device processes the local transaction after sending the transaction message;
when the queried transaction state is determined to be a successful state, writing a commit message into a log database; creating an index for the transaction, obtaining the transaction to be consumed by the equipment, and deleting the preparation message in the cache queue;
when the queried transaction state is determined to be a failure state, writing a rollback message into a log database; and delete the ready message in the cache queue.
Further, the logic instructions in the memory 430 described above may be implemented in the form of software functional units and may be stored in a computer-readable storage medium when sold or used as a stand-alone product. Based on this understanding, the technical solution of the present invention may be embodied essentially or in a part contributing to the prior art or in a part of the technical solution, 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 server, a network device, etc.) to perform all or part of the steps of the method according to the embodiments of the present invention. And the aforementioned storage medium includes: a U-disk, a removable hard disk, a Read-Only Memory (ROM), a random access Memory (RAM, random Access Memory), a magnetic disk, or an optical disk, or other various media capable of storing program codes.
The apparatus embodiments described above are merely illustrative, wherein the elements illustrated as separate elements may or may not be physically separate, and the elements shown as elements may or may not be physical elements, may be located in one place, or may be distributed over a plurality of network elements. Some or all of the modules may be selected according to actual needs to achieve the purpose of the solution of this embodiment. Those of ordinary skill in the art will understand and implement the present invention without undue burden.
From the above description of the embodiments, it will be apparent to those skilled in the art that the embodiments may be implemented by means of software plus necessary general hardware platforms, or of course may be implemented by means of hardware. Based on this understanding, the foregoing technical solution may be embodied essentially or in a part contributing to the prior art in the form of a software product, which may be stored in a computer readable storage medium, such as ROM/RAM, a magnetic disk, an optical disk, etc., including several instructions for causing a computer device (which may be a personal computer, a server, or a network device, etc.) to execute the method described in the respective embodiments or some parts of the embodiments. The foregoing description of the preferred embodiments of the invention is not intended to be limiting, but rather to enable any modification, equivalent replacement, improvement or the like to be made within the spirit and principles of the invention.
Claims (10)
1. A method of message processing, the method comprising:
scanning the prepared message in the cache queue;
if the fact that the overtime preparation message exists is determined, inquiring a transaction state corresponding to the preparation message from the generating equipment; the transaction state is recorded when the generation device processes the local transaction after sending the transaction message;
when the queried transaction state is determined to be a successful state, writing the submitted message into a log database; creating an index for the transaction, obtaining the transaction to be consumed by the equipment, and deleting the preparation message in the cache queue;
when the queried transaction state is determined to be a failure state, writing a rollback message into a log database; deleting the preparation message in the cache queue;
wherein the method further comprises:
after fault recovery, acquiring a preparation message corresponding to the transaction in which the rollback message and the commit message are not written from the log database, and storing the preparation message into a cache queue;
inquiring the transaction state corresponding to the preparation message from the generating equipment;
when the queried transaction state is determined to be a successful state, writing the submitted message into a log database; creating an index for the transaction, obtaining the transaction to be consumed by the equipment, and deleting the preparation message in the cache queue;
when the queried transaction state is determined to be a failure state, writing a rollback message into a log database; deleting the preparation message in the cache queue;
wherein determining whether the prepare message has timed out comprises:
setting a timer for the preparation message, and determining that the preparation message is overtime if the preparation message is still in a cache queue when the timer is overtime.
2. The method according to claim 1, wherein the method further comprises:
when receiving a preparation message sent by generating equipment, storing the preparation message into a cache queue, and writing the preparation message into a log database;
and when receiving the transaction information sent by the generating equipment, writing the transaction information into a log database.
3. The method according to claim 1, wherein the method further comprises:
when receiving a commit message sent by generating equipment, writing the commit message into a log database; creating an index for the transaction, obtaining the transaction to be consumed by the equipment, and deleting the preparation message in the cache queue;
when receiving a rollback message sent by the generating equipment, writing the rollback message into a log database; and delete the ready message in the cache queue.
4. A method according to any one of claims 1-3, wherein the method further comprises:
when receiving a message sent by a sender, sequentially reading and writing;
the data stored locally is replicated and stored to the backup device.
5. A message processing apparatus, the apparatus comprising: the device comprises a scanning unit, an acquisition unit, a determination unit and a processing unit;
the scanning unit is used for scanning the preparation messages in the cache queue;
the acquisition unit is used for inquiring the transaction state corresponding to the preparation message from the generating equipment if the overtime preparation message exists in the preparation message scanned by the scanning unit; the transaction state is recorded when the generation device processes the local transaction after sending the transaction message;
the determining unit is used for determining whether the transaction state acquired by the acquiring unit is a success state or a failure state;
the processing unit is used for writing the submitted information into the log database when the determining unit determines that the queried transaction state is a successful state; creating an index for the transaction, obtaining the transaction to be consumed by the equipment, and deleting the preparation message in the cache queue; when the determining unit determines that the queried transaction state is a failure state, writing a rollback message into a log database; deleting the preparation message in the cache queue;
wherein,,
the obtaining unit is further used for obtaining the preparation message corresponding to the transaction in which the rollback message and the commit message are not written from the log database after the fault of the device where the obtaining unit is located is recovered; and storing the data in a cache queue; inquiring the transaction state corresponding to the preparation message from the generating equipment;
the determining unit is used for determining whether the transaction state acquired by the acquiring unit is a success state or a failure state;
the processing unit is used for writing the submitted information into the log database when the determining unit determines that the queried transaction state is a successful state; creating an index for the transaction, obtaining the transaction to be consumed by the equipment, and deleting the preparation message in the cache queue; when the determining unit determines that the queried transaction state is a failure state, writing a rollback message into a log database; deleting the preparation message in the cache queue;
the obtaining unit is further configured to set a timer for the preparation message, and when the timer expires, determine that the preparation message expires if the preparation message is still in the cache queue.
6. The apparatus of claim 5, wherein the apparatus further comprises: a receiving unit;
the receiving unit is used for receiving the preparation message sent by the generating equipment; receiving a transaction message sent by generating equipment;
the processing unit is used for storing the preparation message into a cache queue and writing the preparation message into a log database when the receiving unit receives the preparation message sent by the generating device; when the receiving unit receives the transaction message sent by the generating device, the transaction message is written into a log database.
7. The apparatus of claim 5, wherein the apparatus further comprises: a receiving unit;
the receiving unit is used for receiving the commit message sent by the generating equipment and receiving the rollback message sent by the generating equipment;
the processing unit is further used for writing the commit message into a log database when the receiving unit receives the commit message sent by the generating device; creating an index for the transaction, obtaining the transaction to be consumed by the equipment, and deleting the preparation message in the cache queue; when receiving a rollback message sent by the generating equipment, writing the rollback message into a log database; and delete the ready message in the cache queue.
8. The apparatus according to any one of claims 5 to 7, wherein,
the processing unit is further used for sequentially reading and writing when the receiving unit receives the message sent by the sender; the data stored locally is replicated and stored to the backup device.
9. An electronic device comprising a memory, a processor and a computer program stored on the memory and executable on the processor, wherein the processor implements the method of any of claims 1-4 when the program is executed by the processor.
10. A computer readable storage medium, on which a computer program is stored, characterized in that the program, when being executed by a processor, implements the method of any of claims 1-4.
Priority Applications (1)
| Application Number | Priority Date | Filing Date | Title |
|---|---|---|---|
| CN201911125238.6A CN112822091B (en) | 2019-11-18 | 2019-11-18 | Message processing method and device |
Applications Claiming Priority (1)
| Application Number | Priority Date | Filing Date | Title |
|---|---|---|---|
| CN201911125238.6A CN112822091B (en) | 2019-11-18 | 2019-11-18 | Message processing method and device |
Publications (2)
| Publication Number | Publication Date |
|---|---|
| CN112822091A CN112822091A (en) | 2021-05-18 |
| CN112822091B true CN112822091B (en) | 2023-05-30 |
Family
ID=75852162
Family Applications (1)
| Application Number | Title | Priority Date | Filing Date |
|---|---|---|---|
| CN201911125238.6A Active CN112822091B (en) | 2019-11-18 | 2019-11-18 | Message processing method and device |
Country Status (1)
| Country | Link |
|---|---|
| CN (1) | CN112822091B (en) |
Families Citing this family (4)
| Publication number | Priority date | Publication date | Assignee | Title |
|---|---|---|---|---|
| CN113626457A (en) * | 2021-08-19 | 2021-11-09 | 福建天晴数码有限公司 | Method and system for realizing database and cache consistency by cache deletion retry mechanism |
| CN113779149A (en) * | 2021-09-14 | 2021-12-10 | 北京知道创宇信息技术股份有限公司 | Message processing method and device, electronic equipment and readable storage medium |
| CN113742354B (en) * | 2021-09-18 | 2024-02-09 | 中国银行股份有限公司 | Message confirmation method and device, storage medium and electronic equipment |
| CN114116262B (en) * | 2021-12-02 | 2022-08-23 | 北京宇信科技集团股份有限公司 | Processing method, device, medium and equipment for distributed asynchronous data communication |
Citations (3)
| Publication number | Priority date | Publication date | Assignee | Title |
|---|---|---|---|---|
| CN103312549A (en) * | 2013-06-26 | 2013-09-18 | 华为技术有限公司 | Transaction management method, device and system |
| CN106462601A (en) * | 2014-03-31 | 2017-02-22 | 亚马逊科技公司 | Atomic writes for multiple-extent operations |
| CN108009027A (en) * | 2017-11-23 | 2018-05-08 | 北京百度网讯科技有限公司 | Implementation method, device, equipment and the storage medium of queue message uniformity |
Family Cites Families (8)
| Publication number | Priority date | Publication date | Assignee | Title |
|---|---|---|---|---|
| US5335343A (en) * | 1992-07-06 | 1994-08-02 | Digital Equipment Corporation | Distributed transaction processing using two-phase commit protocol with presumed-commit without log force |
| US5781910A (en) * | 1996-09-13 | 1998-07-14 | Stratus Computer, Inc. | Preforming concurrent transactions in a replicated database environment |
| US7725432B2 (en) * | 2004-05-21 | 2010-05-25 | Bea Systems, Inc. | Logging last resource system |
| US9892160B2 (en) * | 2015-04-07 | 2018-02-13 | International Business Machines Corporation | Database statistics based on transaction state |
| US10884869B2 (en) * | 2015-04-16 | 2021-01-05 | Nuodb, Inc. | Backup and restore in a distributed database utilizing consistent database snapshots |
| CN107918620B (en) * | 2016-10-10 | 2022-04-19 | 阿里巴巴集团控股有限公司 | Database writing method and device and electronic equipment |
| CN106503257B (en) * | 2016-11-15 | 2019-09-20 | 北京京东金融科技控股有限公司 | Distributed transaction server method and system based on binlog compensation mechanism |
| US10318521B2 (en) * | 2016-11-29 | 2019-06-11 | International Business Machines Corporation | Query processing with bounded staleness for transactional mutations in NoSQL database |
-
2019
- 2019-11-18 CN CN201911125238.6A patent/CN112822091B/en active Active
Patent Citations (3)
| Publication number | Priority date | Publication date | Assignee | Title |
|---|---|---|---|---|
| CN103312549A (en) * | 2013-06-26 | 2013-09-18 | 华为技术有限公司 | Transaction management method, device and system |
| CN106462601A (en) * | 2014-03-31 | 2017-02-22 | 亚马逊科技公司 | Atomic writes for multiple-extent operations |
| CN108009027A (en) * | 2017-11-23 | 2018-05-08 | 北京百度网讯科技有限公司 | Implementation method, device, equipment and the storage medium of queue message uniformity |
Also Published As
| Publication number | Publication date |
|---|---|
| CN112822091A (en) | 2021-05-18 |
Similar Documents
| Publication | Publication Date | Title |
|---|---|---|
| CN112822091B (en) | Message processing method and device | |
| CN102265277B (en) | Operation method and device for data memory system | |
| US6873995B2 (en) | Method, system, and program product for transaction management in a distributed content management application | |
| US7624112B2 (en) | Asynchronously storing transaction information from memory to a persistent storage | |
| JP2558052B2 (en) | Transaction processing system using hypothetical commit two-phase commit protocol and operating method thereof | |
| TW522320B (en) | Apparatus and method for recovering a failed database data set | |
| US7693882B2 (en) | Replicating data across the nodes in a cluster environment | |
| US20190129976A1 (en) | Apparatus for controlling synchronization of metadata on network and method for the same | |
| JPH11184744A (en) | Message queuing system | |
| CN111225012A (en) | Transaction processing method, device and equipment | |
| US20150261626A1 (en) | Data restoration method and system | |
| US20070143362A1 (en) | Database system including center server and local servers | |
| CN112597249A (en) | Synchronous distribution and storage method and system for service data | |
| CN113010549B (en) | Data processing method based on remote multi-activity system, related equipment and storage medium | |
| CN111475583A (en) | Transaction processing method and device | |
| EP0834122B1 (en) | Synchronisation procedure in a routing node | |
| US12111817B2 (en) | Log execution method and apparatus, computer device and storage medium | |
| CN101551822B (en) | Method and apparatus for copying database | |
| US20150286496A1 (en) | Systems and methods for enlisting single phase commit resources in a two phase commit transaction | |
| CN115617571A (en) | Data backup method, device, system, equipment and storage medium | |
| CN109726211B (en) | Distributed time sequence database | |
| CN112749172A (en) | Data synchronization method and system between cache and database | |
| US7529783B2 (en) | Log shipping data replication with parallel log writing and log shipping at the primary site | |
| US20100145933A1 (en) | Dynamic Restoration of Message Object Search Indexes | |
| US11693844B2 (en) | Processing delete requests based on change feed of updates |
Legal Events
| Date | Code | Title | Description |
|---|---|---|---|
| PB01 | Publication | ||
| PB01 | Publication | ||
| SE01 | Entry into force of request for substantive examination | ||
| SE01 | Entry into force of request for substantive examination | ||
| GR01 | Patent grant | ||
| GR01 | Patent grant |