快捷導(dǎo)航

RabbitMQ?延遲隊列實現(xiàn)訂單支付結(jié)果異步階梯性通知(實例代碼)

更新時間：2022年02月17日 11:49:06 作者：星朝

這篇文章主要介紹了RabbitMQ?延遲隊列實現(xiàn)訂單支付結(jié)果異步階梯性通知,本文通過實例代碼給大家介紹的非常詳細(xì)，對大家的學(xué)習(xí)或工作具有一定的參考借鑒價值，需要的朋友可以參考下

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

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

常見的補單操作，任務(wù)調(diào)度策略一般設(shè)定30秒、60秒、3分鐘、6分鐘、10分鐘調(diào)度多次（以自己業(yè)務(wù)需要），如果調(diào)度接收到響應(yīng)確認(rèn)報文，補單成功，則中止對應(yīng)訂單的調(diào)度任務(wù)；如果超過補單上限次數(shù)，則停止補單，避免無謂的資源浪費。請求端隨時可以發(fā)起請求報文查詢對應(yīng)訂單的狀態(tài)。在日常開發(fā)中，對于網(wǎng)站前端來說，支付計費中心對于訂單請求信息的處理也是通過消息同步返回、異步通知+主動補償查詢相結(jié)合的機制，其中對于訂單的異步通知，目前的通知策略為3s、30s、60s、120s、180、300s的階梯性通知。返回成功情況下就不繼續(xù)通知了，本來打算使用將失敗的消息寫到數(shù)據(jù)庫等待發(fā)送，然后每秒查詢數(shù)據(jù)庫獲取消息通知前端。但覺得這樣的處理方式太粗暴。

存在以下缺點：

1 、每秒請求有點兒浪費資源；

2 、通知方式不穩(wěn)定；

3 、無法承受大數(shù)據(jù)量等等

所以最終打算使用rabbitmq的消息延遲+死信隊列來實現(xiàn)。消息模型如下：

producer發(fā)布消息，通過exchangeA的消息會被分發(fā)到QueueA,Consumer監(jiān)聽queueA，一旦有消息到來就被消費，這邊的消費業(yè)務(wù)就是通知前端，如果通知失敗，就創(chuàng)建一個延遲隊列declareQueue，設(shè)置每個消息的ttl然后通過declare_exchange將消息分發(fā)到declare_queue,因為declare_queue沒有consumer并且declare_queue中的消息設(shè)置了ttl，當(dāng)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);// 設(shè)置消息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 {
		builder.expiration(String.valueOf(mis)).deliveryMode(2);// 設(shè)置消息TTL
		channel.queueDeclare(queue, false, false, false, args);
		channel.queueBind(queue, EX_declare_15S, ROUTINGKEY);
		System.out.println("send message in " + queue + message + "time============" + System.currentTimeMillis());
	public static void main(String args[]) throws Exception {
		init();
		consume();
}

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

到此這篇關(guān)于RabbitMQ 延遲隊列實現(xiàn)訂單支付結(jié)果異步階梯性通知的文章就介紹到這了,更多相關(guān)RabbitMQ 延遲隊列實現(xiàn)訂單支付內(nèi)容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

您可能感興趣的文章: