RabbitMQ 延迟队列实现订单支付结果异步阶梯性通知[通俗易懂]

2022-08-27 13:25:36 浏览数 (1)

大家好，又见面了，我是你们的朋友全栈君。

在第三方支付中，例如支付宝、或者微信，对于订单请求，第三方支付系统采用的是消息同步返回、异步通知 主动补偿查询的补偿机制。

由于互联网通信的不可靠性，例如双方网络、服务器、应用等因素的影响，不管是同步返回、异步通知、主动查询报文都可能出现超时无响应、报文丢失等情况，所以像支付业务，对结果的通知一般采用几种方案结合的补偿机制，不能完全依赖某一种机制。 例如一个支付结果的通知，一方面会在支付页面跳转时候返回支付结果（一般只用作前端展示使用，非最终状态），同时会采用后台异步通知机制（有前台、后台通知的，以后台异步通知结果为准），但由于前台跳转、后台结果通知都可能失效，因此还以定时补单 请求方主动查询接口作为辅助手段。 常见的补单操作，任务调度策略一般设定30秒、60秒、3分钟、6分钟、10分钟调度多次（以自己业务需要），如果调度接收到响应确认报文，补单成功，则中止对应订单的调度任务；如果超过补单上限次数，则停止补单，避免无谓的资源浪费。请求端随时可以发起请求报文查询对应订单的状态。

在日常开发中，对于网站前端来说，支付计费中心对于订单请求信息的处理也是通过消息同步返回、异步通知 主动补偿查询相结合的机制，其中对于订单的异步通知，目前的通知策略为3s、30s、60s、120s、180、300s的阶梯性通知。返回成功情况下就不继续通知了，本来打算使用将失败的消息写到数据库等待发送，然后每秒查询数据库获取消息通知前端。但觉得这样的处理方式太粗暴。存在以下缺点：

1 、每秒请求有点儿浪费资源； 2 、通知方式不稳定； 3 、无法承受大数据量等等

所以最终打算使用rabbitmq的消息延迟 死信队列来实现。消息模型如下：

producer发布消息，通过exchangeA的消息会被分发到QueueA,Consumer监听queueA，一旦有消息到来就被消费，这边的消费业务就是通知前端，如果通知失败，就创建一个延迟队列declareQueue，设置每个消息的ttl然后通过declare_exchange将消息分发到declare_queue,因为declare_queue没有consumer并且declare_queue中的消息设置了ttl，当ttl到期后，将通过DEX路由到queueA,被重新消费。

代码如下：DeclareQueue.java

  1. package org.delayQueue;
  3. import com.rabbitmq.client.BuiltinExchangeType;
  4. import com.rabbitmq.client.Channel;
  5. import com.rabbitmq.client.Connection;
  6. import com.rabbitmq.client.ConnectionFactory;
  8. public class DeclareQueue {
  9. public static String EXCHANGE_NAME = "notifyExchange";
  11. public static void init() {
  12. ConnectionFactory factory = new ConnectionFactory();
  13. factory.setHost("localhost");
  14. factory.setPort(5672);
  16. Connection connection = null;
  17. try {
  18. connection = factory.newConnection();
  19. Channel channel = connection.createChannel();
  20. channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.TOPIC);
  21. String routingKey = "AliPaynotify";
  22. String message = "http://localhost:8080/BossCenter/payGateway/notifyRecv.jsp?is_success=T¬ify_id=4ab9bed148d043d0bf75460706f7774a¬ify_time=2014-08-29 16:22:02¬ify_type=trade_status_sync&out_trade_no=1421712120109862&total_fee=424.42&trade_no=14217121201098611&trade_status=TRADE_SUCCESS";
  23. channel.basicPublish(EXCHANGE_NAME, routingKey, null, message.getBytes());
  24. System.out.println(" [x] Sent :" message);
  25. } catch (Exception e) {
  26. // TODO Auto-generated catch block
  27. e.printStackTrace();
  28. } finally {
  29. if (connection != null) {
  30. try {
  31. connection.close();
  32. } catch (Exception ignore) {
  33. }
  34. }
  35. }
  36. }
  38. public static void main(String args[]) {
  39. init();
  40. }
  42. }

DeclareConsumer.java

  1. package org.delayQueue;
  3. import java.io.BufferedReader;
  4. import java.io.IOException;
  5. import java.io.InputStreamReader;
  6. import java.util.ArrayList;
  7. import java.util.HashMap;
  8. import java.util.List;
  9. import java.util.Map;
  10. import java.util.Map.Entry;
  12. import org.apache.http.HttpResponse;
  13. import org.apache.http.client.ClientProtocolException;
  14. import org.apache.http.client.HttpClient;
  15. import org.apache.http.client.methods.HttpPost;
  16. import org.apache.http.impl.client.DefaultHttpClient;
  18. import com.rabbitmq.client.AMQP;
  19. import com.rabbitmq.client.Channel;
  20. import com.rabbitmq.client.Connection;
  21. import com.rabbitmq.client.ConnectionFactory;
  22. import com.rabbitmq.client.Consumer;
  23. import com.rabbitmq.client.DefaultConsumer;
  24. import com.rabbitmq.client.Envelope;
  26. public class DeclareConsumer {
  27. public static String EXCHANGE_NAME = "notifyExchange";
  28. public static String QU_declare_15S = "Qu_declare_15s";
  29. public static String EX_declare_15S = "EX_declare_15s";
  30. public static String ROUTINGKEY = "AliPaynotify";
  31. public static Connection connection = null;
  32. public static Channel channel = null;
  33. public static Channel DECLARE_15S_CHANNEL = null;
  34. public static String declare_queue = "init";
  35. public static String originalExpiration = "0";
  36. public static void init() throws Exception {
  37. ConnectionFactory factory = new ConnectionFactory();
  38. factory.setHost("localhost");
  39. factory.setPort(5672);
  40. connection = factory.newConnection();
  41. channel = connection.createChannel();
  42. DECLARE_15S_CHANNEL = connection.createChannel();
  43. }
  45. public static void consume() {
  46. try {
  47. channel.exchangeDeclare(EXCHANGE_NAME, "topic");
  48. final String queueName = channel.queueDeclare().getQueue();
  50. channel.queueBind(queueName, EXCHANGE_NAME, ROUTINGKEY);
  51. System.out.println(" [*] Waiting for messages. To exit press CTRL C");
  52. final Consumer consumer = new DefaultConsumer(channel) {
  53. @Override
  54. public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
  55. String message = new String(body, "UTF-8");
  56. Map<String, Object> headers = properties.getHeaders();
  57. if (headers != null) {
  58. List<Map<String, Object>> xDeath = (List<Map<String, Object>>) headers.get("x-death");
  59. System.out.println("xDeath--- > " xDeath);
  60. if (xDeath != null && !xDeath.isEmpty()) {
  61. Map<String, Object> entrys = xDeath.get(0);
  62. // for(Entry<String, Object>
  63. // entry:entrys.entrySet()){
  64. // System.out.println(entry.getKey() ":" entry.getValue());
  65. // }
  66. originalExpiration = entrys.get("original-expiration").toString();
  67. }
  68. }
  69. System.out.println(" [x] Received '" envelope.getRoutingKey() "':'" message "'" "time" System.currentTimeMillis());
  70. HttpClient httpClient = new DefaultHttpClient();
  71. HttpPost post = new HttpPost(message);
  72. HttpResponse response = httpClient.execute(post);
  73. BufferedReader inreader = null;
  74. if (response.getStatusLine().getStatusCode() == 200) {
  75. inreader = new BufferedReader(new InputStreamReader(response.getEntity().getContent(), "UTF-8"));
  76. StringBuffer responseBody = new StringBuffer();
  77. String line = null;
  78. while ((line = inreader.readLine()) != null) {
  79. responseBody.append(line);
  80. }
  81. if (!responseBody.equals("success")) {
  82. // putDeclre15s(message);
  83. if (originalExpiration.equals("0")) {
  84. putDeclreQueue(message, 3000, QU_declare_15S);
  85. }
  86. if (originalExpiration.equals("3000")) {
  87. putDeclreQueue(message, 30000, QU_declare_15S);
  88. }
  89. if (originalExpiration.equals("30000")) {
  90. putDeclreQueue(message, 60000, QU_declare_15S);
  91. }
  92. if (originalExpiration.equals("60000")) {
  93. putDeclreQueue(message, 120000, QU_declare_15S);
  94. }
  95. if (originalExpiration.equals("120000")) {
  96. putDeclreQueue(message, 180000, QU_declare_15S);
  97. }
  98. if (originalExpiration.equals("180000")) {
  99. putDeclreQueue(message, 300000, QU_declare_15S);
  100. }
  101. if (originalExpiration.equals("300000")) {
  102. // channel.basicConsume(QU_declare_300S,true, this);
  103. System.out.println("finish notify");
  104. }
  105. }
  106. } else {
  107. System.out.println(response.getStatusLine().getStatusCode());
  108. }
  109. }
  110. };
  112. channel.basicConsume(queueName, true, consumer);
  113. } catch (Exception e) {
  114. e.printStackTrace();
  115. } finally {
  116. }
  117. }
  121. static Map<String, Object> xdeathMap = new HashMap<String, Object>();
  122. static List<Map<String, Object>> xDeath = new ArrayList<Map<String, Object>>();
  123. static Map<String, Object> xdeathParam = new HashMap<String, Object>();
  125. public static void putDeclre15s(String message) throws IOException {
  126. channel.exchangeDeclare(EX_declare_15S, "topic");
  127. Map<String, Object> args = new HashMap<String, Object>();
  128. args.put("x-dead-letter-exchange", EXCHANGE_NAME);// 死信exchange
  129. AMQP.BasicProperties.Builder builder = new AMQP.BasicProperties.Builder();
  130. builder.expiration("3000").deliveryMode(2);// 设置消息TTL
  131. AMQP.BasicProperties properties = builder.build();
  132. channel.queueDeclare(QU_declare_15S, false, false, false, args);
  133. channel.queueBind(QU_declare_15S, EX_declare_15S, ROUTINGKEY);
  134. channel.basicPublish(EX_declare_15S, ROUTINGKEY, properties, message.getBytes());
  135. System.out.println("send message in QA_DEFERRED_15S" message "time" System.currentTimeMillis());
  136. }
  138. public static void putDeclreQueue(String message, int mis, String queue) throws IOException {
  139. channel.exchangeDeclare(EX_declare_15S, "topic");
  140. Map<String, Object> args = new HashMap<String, Object>();
  141. args.put("x-dead-letter-exchange", EXCHANGE_NAME);// 死信exchange
  142. AMQP.BasicProperties.Builder builder = new AMQP.BasicProperties.Builder();
  143. builder.expiration(String.valueOf(mis)).deliveryMode(2);// 设置消息TTL
  144. AMQP.BasicProperties properties = builder.build();
  145. channel.queueDeclare(queue, false, false, false, args);
  146. channel.queueBind(queue, EX_declare_15S, ROUTINGKEY);
  147. channel.basicPublish(EX_declare_15S, ROUTINGKEY, properties, message.getBytes());
  148. System.out.println("send message in " queue message "time============" System.currentTimeMillis());
  149. }
  151. public static void main(String args[]) throws Exception {
  152. init();
  153. consume();
  154. }
  155. }

消息通过dlx转发的情况下，header头部会带有x-death的一个数组，里面包含消息的各项属性，比如说消息成为死信的原因reason，original-expiration这个字段表示消息在原来队列中的过期时间，根据这个值来确定下一次通知的延迟时间应该是多少秒。

运行结果如下：

发布者：全栈程序员栈长，转载请注明出处：https://javaforall.cn/145842.html原文链接：https://javaforall.cn

编程算法 c++ java hashmap

0 人点赞