RabbitMQ 延遲隊列實現訂單支付結果異步階梯性通知

阿新 • • 發佈：2018-11-05

響應 ssa 訂單 nor idt ons fill comm port

在第三方支付中，例如支付寶、或者微信，對於訂單請求，第三方支付系統采用的是消息同步返回、異步通知+主動補償查詢的補償機制。

由於互聯網通信的不可靠性，例如雙方網絡、服務器、應用等因素的影響，不管是同步返回、異步通知、主動查詢報文都可能出現超時無響應、報文丟失等情況，所以像支付業務，對結果的通知一般采用幾種方案結合的補償機制，不能完全依賴某一種機制。
例如一個支付結果的通知，一方面會在支付頁面跳轉時候返回支付結果（一般只用作前端展示使用，非最終狀態），同時會采用後臺異步通知機制（有前臺、後臺通知的，以後臺異步通知結果為準），但由於前臺跳轉、後臺結果通知都可能失效，因此還以定時補單+請求方主動查詢接口作為輔助手段。

常見的補單操作，任務調度策略一般設定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

package org.delayQueue;
  
import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
 
public class DeclareQueue {
  public static String EXCHANGE_NAME = "notifyExchange";
 
    public static void init() {
       ConnectionFactory factory = new ConnectionFactory();
      factory.setHost("localhost");
      factory.setPort(5672);
 
        Connection connection = null;
     try {
         connection = factory.newConnection();
           Channel channel = connection.createChannel();
           channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.TOPIC);
          String routingKey = "AliPaynotify";
            String message = "http://localhost:8080/BossCenter/payGateway/notifyRecv.jsp?is_success=T?ify_id=4ab9bed148d043d0bf75460706f7774a?ify_time=2014-08-29+16%3A22%3A02?ify_type=trade_status_sync&out_trade_no=1421712120109862&total_fee=424.42&trade_no=14217121201098611&trade_status=TRADE_SUCCESS";
           channel.basicPublish(EXCHANGE_NAME, routingKey, null, message.getBytes());
            System.out.println(" [x] Sent :" + message);
       } catch (Exception e) {
           // TODO Auto-generated catch block
            e.printStackTrace();
        } finally {
           if (connection != null) {
               try {
                 connection.close();
             } catch (Exception ignore) {
              }
           }
       }
   }
 
  public static void main(String args[]) {
      init();
 }
 
}

DeclareConsumer.java

package org.delayQueue;
 
import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
 
import org.apache.http.HttpResponse;
import org.apache.http.client.ClientProtocolException;
import org.apache.http.client.HttpClient;
import org.apache.http.client.methods.HttpPost;
import org.apache.http.impl.client.DefaultHttpClient;
 
import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.Consumer;
import com.rabbitmq.client.DefaultConsumer;
import com.rabbitmq.client.Envelope;
 
public class DeclareConsumer {
 public static String EXCHANGE_NAME = "notifyExchange";
  public static String QU_declare_15S = "Qu_declare_15s";
 public static String EX_declare_15S = "EX_declare_15s";
 public static String ROUTINGKEY = "AliPaynotify";
   public static Connection connection = null;
 public static Channel channel = null;
   public static Channel DECLARE_15S_CHANNEL = null;
   public static String declare_queue = "init";
    public static String originalExpiration = "0";
  public static void init() throws Exception {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost");
       factory.setPort(5672);
      connection = factory.newConnection();
       channel = connection.createChannel();
       DECLARE_15S_CHANNEL = connection.createChannel();
   }
 
  public static void consume() {
      try {
           channel.exchangeDeclare(EXCHANGE_NAME, "topic");
            final String queueName = channel.queueDeclare().getQueue();
 
            channel.queueBind(queueName, EXCHANGE_NAME, ROUTINGKEY);
            System.out.println(" [*] Waiting for messages. To exit press CTRL+C");
          final Consumer consumer = new DefaultConsumer(channel) {
                @Override
               public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                    String message = new String(body, "UTF-8");
                 Map<String, Object> headers = properties.getHeaders();
                    if (headers != null) {
                      List<Map<String, Object>> xDeath = (List<Map<String, Object>>) headers.get("x-death");
                        System.out.println("xDeath--- > " + xDeath);
                     if (xDeath != null && !xDeath.isEmpty()) {
                          Map<String, Object> entrys = xDeath.get(0);
                           // for(Entry<String, Object>
                          // entry:entrys.entrySet()){
                            // System.out.println(entry.getKey()+":"+entry.getValue());
                         // }
                            originalExpiration = entrys.get("original-expiration").toString();
                      }
                   }
                   System.out.println(" [x] Received ‘" + envelope.getRoutingKey() + "‘:‘" + message + "‘" + "time" + System.currentTimeMillis());
                 HttpClient httpClient = new DefaultHttpClient();
                    HttpPost post = new HttpPost(message);
                  HttpResponse response = httpClient.execute(post);
                   BufferedReader inreader = null;
                 if (response.getStatusLine().getStatusCode() == 200) {
                      inreader = new BufferedReader(new InputStreamReader(response.getEntity().getContent(), "UTF-8"));
                       StringBuffer responseBody = new StringBuffer();
                     String line = null;
                     while ((line = inreader.readLine()) != null) {
                          responseBody.append(line);
                      }
                       if (!responseBody.equals("success")) {
                          // putDeclre15s(message);
                           if (originalExpiration.equals("0")) {
                               putDeclreQueue(message, 3000, QU_declare_15S);
                          }
                           if (originalExpiration.equals("3000")) {
                                putDeclreQueue(message, 30000, QU_declare_15S);
                         }
                           if (originalExpiration.equals("30000")) {
                               putDeclreQueue(message, 60000, QU_declare_15S);
                         }
                           if (originalExpiration.equals("60000")) {
                               putDeclreQueue(message, 120000, QU_declare_15S);
                            }
                           if (originalExpiration.equals("120000")) {
                              putDeclreQueue(message, 180000, QU_declare_15S);
                            }
                           if (originalExpiration.equals("180000")) {
                              putDeclreQueue(message, 300000, QU_declare_15S);
                           }
                          if (originalExpiration.equals("300000")) {
//                               channel.basicConsume(QU_declare_300S,true, this);
                              System.out.println("finish notify");
                           }
                      }
                  } else {
                       System.out.println(response.getStatusLine().getStatusCode());
                  }
              }
          };
 
           channel.basicConsume(queueName, true, consumer);
       } catch (Exception e) {
            e.printStackTrace();
       } finally {
        }
  }
 
 
 
    static Map<String, Object> xdeathMap = new HashMap<String, Object>();
  static List<Map<String, Object>> xDeath = new ArrayList<Map<String, Object>>();
  static Map<String, Object> xdeathParam = new HashMap<String, Object>();
 
  public static void putDeclre15s(String message) throws IOException {
       channel.exchangeDeclare(EX_declare_15S, "topic");
      Map<String, Object> args = new HashMap<String, Object>();
      args.put("x-dead-letter-exchange", EXCHANGE_NAME);// 死信exchange
        AMQP.BasicProperties.Builder builder = new AMQP.BasicProperties.Builder();
     builder.expiration("3000").deliveryMode(2);// 設置消息TTL
      AMQP.BasicProperties properties = builder.build();
     channel.queueDeclare(QU_declare_15S, false, false, false, args);
       channel.queueBind(QU_declare_15S, EX_declare_15S, ROUTINGKEY);
     channel.basicPublish(EX_declare_15S, ROUTINGKEY, properties, message.getBytes());
      System.out.println("send message in QA_DEFERRED_15S" + message + "time" + System.currentTimeMillis());
 }
 
    public static void putDeclreQueue(String message, int mis, String queue) throws IOException {
      channel.exchangeDeclare(EX_declare_15S, "topic");
      Map<String, Object> args = new HashMap<String, Object>();
      args.put("x-dead-letter-exchange", EXCHANGE_NAME);// 死信exchange
        AMQP.BasicProperties.Builder builder = new AMQP.BasicProperties.Builder();
     builder.expiration(String.valueOf(mis)).deliveryMode(2);// 設置消息TTL
     AMQP.BasicProperties properties = builder.build();
     channel.queueDeclare(queue, false, false, false, args);
        channel.queueBind(queue, EX_declare_15S, ROUTINGKEY);
      channel.basicPublish(EX_declare_15S, ROUTINGKEY, properties, message.getBytes());
      System.out.println("send message in " + queue + message + "time============" + System.currentTimeMillis());
    }
 
    public static void main(String args[]) throws Exception {
      init();
        consume();
 }
}

消息通過dlx轉發的情況下，header頭部會帶有x-death的一個數組，裏面包含消息的各項屬性，比如說消息成為死信的原因reason，original-expiration這個字段表示消息在原來隊列中的過期時間，根據這個值來確定下一次通知的延遲時間應該是多少秒。運行結果如下：

技術分享圖片

RabbitMQ 延遲隊列實現訂單支付結果異步階梯性通知

響應 ssa 訂單 nor idt ons fill comm port 在第三方支付中，例如支付寶、或者微信，對於訂單請求，第三方支付系統采用的是消息同步返回、異步通知+主動補償查詢的補償機制。由於互聯網通信的不可靠性，例如雙方網

RabbitMQ 延遲隊列實現訂單支付結果異步階梯性通知

RabbitMQ 延遲隊列實現訂單支付結果異步階梯性通知

RabbitMQ 延遲佇列實現訂單支付結果非同步階梯性通知

實現rabbitmq 延遲隊列功能

Spring Boot（十四）RabbitMQ延遲隊列

rabbitmq延遲隊列demo

spring boot Rabbitmq集成,延時消息隊列實現

RabbitMq初探——用隊列實現RPC

Python-RabbitMQ消息隊列實現rpc

C#中使用RabbitMQ收發隊列消息

循環隊列實現

隊列實現

數據結構-使用隊列實現棧

列車調度（隊列實現）

RabbitMQ入門-隊列

數據結構設計——用隊列實現迷宮問題的求解

Spring Boot RabbitMQ 延遲消息實現完整版

【校招面試之劍指offer】第9-2題用兩個隊列實現一個棧

有贊延遲隊列設計

LeetCode--225--用隊列實現棧

java使用優先級隊列實現哈夫曼編碼

RabbitMQ 延遲隊列實現訂單支付結果異步階梯性通知

相關推薦