全部产品
存储与CDN 数据库 安全 应用服务 数加·人工智能 数加·大数据基础服务 互联网中间件 视频服务 阿里云办公 培训与认证 物联网
消息服务

事务消息

更新时间:2017-06-07 13:26:11

背景描述

有时候我们需要实现本地操作和消息发送的事务一致性功能。即:消息发送成功,则本地操作成功;反之,如果消息发送失败,本地操作失败(成功也需要rollback)。保证不出现操作成功但消息发送失败;或者操作失败但消息发送成功的情况;

另外,消费端,我们也希望消息一定被成功处理一次,不会因为消息端程序崩溃而导致消息没有成功处理,进而需要人工重置消费进度。

解决方案

利用消息服务MNS的延迟消息功能来实现。

准备工作

创建两个队列:

1.事务消息队列

消息的有效期小于消息延迟时间。即如果生产者不主动修改(提交)消息可见时间,消息对消费者不可见;

2.操作日志队列

记录事务消息的操作记录信息。消息延迟时间为事务操作超时时间。日志队列中的消息确认(删除)后将对消费者不可见。

具体步骤

1.发送一条事务准备消息到事务消息队列;

2.写操作日志信息到操作日志队列,日志中包含步骤1消息的消息句柄;

3.执行本地事务操作;

4.如果步骤3成功,提交消息(消息对消费者可见);反之,回滚消息;

5.确认步骤2中的操作日志(删除该日志消息);

6.步骤4后,消费者可以接收到事务消息;

7.消费者处理消息;

8.消费者确认删除消息;

如下图:

异常分析:

生产者异常(例如:进程重启):

A.读取操作日志队列超时未确认日志

B.检查事务结果

C.如果检查得到事务已经成功,则提交消息(重复提交无副作用,同一句柄的消息只能成功提交一次)

D.确认操作日志

消费者异常(例如:进程重启):

消息服务提供至少保证消费一次的特性,只要步骤8不成功,消息在一段时间后可以继续可见,被当前消费者或者其他消费者处理。

消息服务不可达(例如:断网)

消息发送和接收处理状态以及操作日志都在消息服务端,消息服务本身具备高可靠和高可用的特点,所以只要网络恢复,事务可以继续,能保证只要生产者:操作成功,则消费者一定能够拿到消息并处理成功;或操作失败, 则消费者收不到消息的最终一致性。

事务图片

代码实现:

MNS最新的Java SDK(1.1.5)中的TransactionQueue支持上述事务消息方案。使用者只需要在TransactionOperations和TransactionChecker 两个接口添加业务操作和检查逻辑,就可以方便的实现事务消息。

Demo 代码

  1. public class TransactionMessageDemo{
  2. public class MyTransactionChecker implements TransactionChecker
  3. {
  4. public boolean checkTransactionStatus(Message message)
  5. {
  6. boolean checkResult = false;
  7. String messageHandler = message.getReceiptHandle();
  8. try{
  9. //TODO: check if the messageHandler related transaction is success.
  10. checkResult = true;
  11. }catch(Exception e)
  12. {
  13. checkResult = false;
  14. }
  15. return checkResult;
  16. }
  17. }
  18. public class MyTransactionOperations implements TransactionOperations
  19. {
  20. public boolean doTransaction(Message message)
  21. {
  22. boolean transactionResult = false;
  23. String messageHandler = message.getReceiptHandle();
  24. String messageBody = message.getMessageBody();
  25. try{
  26. //TODO: do your local transaction according to the messageHandler and messageBody here.
  27. transactionResult = true;
  28. }catch(Exception e)
  29. {
  30. transactionResult = false;
  31. }
  32. return transactionResult;
  33. }
  34. }
  35. public static void main(String[] args) {
  36. System.out.println("Start TransactionMessageDemo");
  37. String transQueueName = "transQueueName";
  38. String accessKeyId = ServiceSettings.getMNSAccessKeyId();
  39. String accessKeySecret = ServiceSettings.getMNSAccessKeySecret();
  40. String endpoint = ServiceSettings.getMNSAccountEndpoint();
  41. CloudAccount account = new CloudAccount(accessKeyId, accessKeySecret, endpoint);
  42. MNSClient client = account.getMNSClient(); //this client need only initialize once
  43. // create queue for transaction queue.
  44. QueueMeta queueMeta = new QueueMeta();
  45. queueMeta.setQueueName(transQueueName);
  46. queueMeta.setPollingWaitSeconds(15);
  47. TransactionMessageDemo demo = new TransactionMessageDemo();
  48. TransactionChecker transChecker = demo.new MyTransactionChecker();
  49. TransactionOperations transOperations = demo.new MyTransactionOperations();
  50. TransactionQueue transQueue = client.createTransQueue(queueMeta, transChecker);
  51. // do transaction.
  52. Message msg = new Message();
  53. String messageBody = "TransactionMessageDemo";
  54. msg.setMessageBody(messageBody);
  55. transQueue.sendTransMessage(msg, transOperations);
  56. // delete queue and close client if we won't use them.
  57. transQueue.delete();
  58. // close the client at the end.
  59. client.close();
  60. System.out.println("End TransactionMessageDemo");
  61. }
  62. }
本文导读目录