2. 程式人生 > 其它 >rabbitmq保證訊息投遞成功的對比分析

rabbitmq保證訊息投遞成功的對比分析

阿新 • • 發佈：2021-01-06

技術標籤：javarabbitmq生產者投遞訊息可靠性javarabbitmq

  • pom.xml

引入rabbitmq所需jar包

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd"
 
>
    <modelVersion>4.0.0</modelVersion>

    <groupId>fun.gosuncn</groupId>
    <artifactId>rabbitmq</artifactId>
    <version>1.0</version>

    <dependencies>
        <dependency>
            <groupId>com.rabbitmq</groupId>
            <
 
artifactId>amqp-client</artifactId>
            <version>5.10.0</version>
        </dependency>
    </dependencies>

</project>
  • RabbitmqUtils

設定連線資訊，獲取連線工廠。

import com.rabbitmq.client.ConnectionFactory;

public class RabbitmqUtils {
    public static ConnectionFactory getConnectionFactory
 
() {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("10.0.8.6");
        factory.setPort(5672);
        factory.setUsername("guest");
        factory.setPassword("guest");
        factory.setVirtualHost("/");
        return factory;
    }
}
  • Receiver

作為消費者不斷的監聽佇列，消費訊息。

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;

public class Receiver {
    public static void main(String[] args) {
        ConnectionFactory factory = RabbitmqUtils.getConnectionFactory();
        try {
            Connection connection = factory.newConnection();
            Channel channel = connection.createChannel();
            channel.queueDeclare("direct-queue", false, false, false, null);
            channel.basicConsume("direct-queue", true, (consumerTag, message) -> {
                System.out.println("DeliverCallback's consumerTag is {" + consumerTag + "}");
                System.out.println("消費訊息的內容 = " + new String(message.getBody()));
            }, consumerTag -> System.out.println("CancelCallback's consumerTag is {" + consumerTag + "}"));
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}
  • 第一種方式 - 事務方式
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;

import java.io.IOException;
import java.util.concurrent.TimeoutException;

public class Sender {
    public static void main(String[] args) {
        ConnectionFactory factory = RabbitmqUtils.getConnectionFactory();
        Connection connection = null;
        Channel channel = null;
        try {
            connection = factory.newConnection();
            channel = connection.createChannel();
            channel.queueDeclare("direct-queue", false, false, false, null);
            channel.txSelect();
            channel.basicPublish("", "direct-queue", null, "hello world".getBytes());
            //int i = 1 / 0;
            channel.txCommit();
        } catch (Exception e) {
            if (channel != null) {
                try {
                    channel.txRollback();
                    // TODO 事務回滾 邏輯 [再次傳送、存庫操作等]
                    System.out.println("事務回滾 邏輯 [再次傳送、存庫操作等]");
                } catch (IOException ioException) {
                    ioException.printStackTrace();
                }
            }
            e.printStackTrace();
        } finally {
            if (channel != null) {
                try {
                    channel.close();
                } catch (IOException | TimeoutException e) {
                    e.printStackTrace();
                }
            }
            if (connection != null) {
                try {
                    connection.close();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
        }
    }
}

開啟事務channel.txSelect();。

執行channel.basicPublish("", "direct-queue", null, "hello world".getBytes());訊息的投遞。

使用int i = 1 / 0;模擬上一步訊息的投遞過程發生了IO異常，此時將不會執行channel.txCommit();操作。

而是進入channel.txRollback();程式碼塊，進行事務回滾邏輯操作。

假如，沒有發生任何異常，將會執行channel.txCommit();操作。

以上過程是如何確認訊息的投遞成功的呢？

首先，客戶端開啟了事務，客戶端將訊息投遞給伺服器，假如投遞過程中因網路等原因發生了IOException，那麼將會進入channel.txRollback();邏輯；假如投遞到伺服器了，後續的邏輯發生異常，進入channel.txRollback();依然是屬於投遞失敗，消費者是消費不到訊息的。

因此，事務的方式可以處理訊息傳送失敗後的解決方案，以確保對訊息的完全掌控。

  • 第二種方式 - 同步confirm方式
package fun.gosuncn.test1;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;

import java.io.IOException;
import java.util.concurrent.TimeoutException;

public class Sender {
    public static void main(String[] args) {
        ConnectionFactory factory = RabbitmqUtils.getConnectionFactory();
        Connection connection = null;
        Channel channel = null;
        try {
            connection = factory.newConnection();
            channel = connection.createChannel();
            channel.queueDeclare("direct-queue", false, false, false, null);
            channel.confirmSelect();
            channel.basicPublish("", "direct-queue", null, "hello world".getBytes());
            //int i = 1 / 0;
            if (channel.waitForConfirms()) {
                System.out.println("訊息已經確認投遞成功");
                Thread.sleep(3000L);
            }
            System.out.println("驗證waitForConfirms是阻塞方法");
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            if (channel != null) {
                try {
                    channel.close();
                } catch (IOException | TimeoutException e) {
                    e.printStackTrace();
                }
            }
            if (connection != null) {
                try {
                    connection.close();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
        }
    }
}

通過channel.confirmSelect();開啟confirm模式。

而channel.waitForConfirms()的作用是，阻塞程式，等待確認結果的返回，進行邏輯操作。

假如channel.basicPublish與channel.waitForConfirms()之間出現了任何異常錯誤，訊息是有可能投遞成功的。

  • 第三種方式 - 同步confirm方式（批量）
package fun.gosuncn.test1;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;

import java.io.IOException;
import java.util.concurrent.TimeoutException;

public class Sender {
    public static void main(String[] args) {
        ConnectionFactory factory = RabbitmqUtils.getConnectionFactory();
        Connection connection = null;
        Channel channel = null;
        try {
            connection = factory.newConnection();
            channel = connection.createChannel();
            channel.queueDeclare("direct-queue", false, false, false, null);
            channel.confirmSelect();
            channel.basicPublish("", "direct-queue", null, "hello world1".getBytes());
            channel.basicPublish("", "direct-queue", null, "hello world2".getBytes());
            channel.basicPublish("", "direct-queue", null, "hello world3".getBytes());
            if (channel.waitForConfirms()) {
                System.out.println("訊息已經確認投遞成功");
                Thread.sleep(3000L);
            }
            System.out.println("驗證waitForConfirms是阻塞方法");
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            if (channel != null) {
                try {
                    channel.close();
                } catch (IOException | TimeoutException e) {
                    e.printStackTrace();
                }
            }
            if (connection != null) {
                try {
                    connection.close();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
        }
    }
}

其實，批量確認和序列確認的區別就是，批量確認是執行多個basicPublish後，執行一次waitForConfirms即可。

  • 第四種方式 - 非同步confirm方式
package fun.gosuncn.test1;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.ConfirmListener;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;

import java.io.IOException;

public class Sender {
    public static void main(String[] args) {
        ConnectionFactory factory = RabbitmqUtils.getConnectionFactory();
        Connection connection = null;
        Channel channel = null;
        try {
            connection = factory.newConnection();
            channel = connection.createChannel();
            channel.queueDeclare("direct-queue", false, false, false, null);
            channel.confirmSelect();
            channel.addConfirmListener(new ConfirmListener() {
                public void handleAck(long deliveryTag, boolean multiple) throws IOException {
                    System.out.println("ack:deliveryTag:" + deliveryTag + "，multiple:" + multiple);
                }

                public void handleNack(long deliveryTag, boolean multiple) throws IOException {
                    System.out.println("nack:deliveryTag:" + deliveryTag + "，multiple:" + multiple);
                }
            });
            channel.basicPublish("", "direct-queue", null, "hello world1".getBytes());
            channel.basicPublish("", "direct-queue", null, "hello world2".getBytes());
            channel.basicPublish("", "direct-queue", null, "hello world3".getBytes());
        } catch (Exception e) {
            e.printStackTrace();
        }
//        finally {
//            if (channel != null) {
//                try {
//                    channel.close();
//                } catch (IOException | TimeoutException e) {
//                    e.printStackTrace();
//                }
//            }
//            if (connection != null) {
//                try {
//                    connection.close();
//                } catch (IOException e) {
//                    e.printStackTrace();
//                }
//            }
//        }
    }
}

使用非同步確認的方式需要注意的是：一定不能關閉連線，否則可能會出現異常。

事務方式：42891毫秒

序列確認：40551毫秒

批量確認：688毫秒

非同步確認：318毫秒

綜上結果，可以看出：事務方式效率最慢，非同步確認方式效率最佳。

相關推薦

rabbitmq保證訊息投遞成功的對比分析

技術標籤：javarabbitmq生產者投遞訊息可靠性javarabbitmq pom.xml 引入rabbitmq所需jar包

springboot + rabbitmq傳送郵件(保證訊息100%投遞成功並被消費)

連結：https://www.jianshu.com/p/dca01aad6bc8 前言: RabbitMQ相關知識請參考: https://www.jianshu.com/p/cc3d2017e7b3

RabbitMQ如何保證訊息的可達性

一、RabbitMQ簡介 AMQP，即Advanced Message Queuing Protocol,一個提供統一訊息服務的應用層標準高階訊息佇列協議，是應用層協議的一個開放標準,為面向訊息的中介軟體設計。

RabbitMQ 之訊息的可靠性投遞

預設情況下，訊息傳送端傳送訊息給 RabbitMQ 後，RabbitMQ 是不會返回任何資訊的。

Rabbitmq如何保證訊息順序執行

1.為什麼要保證順序 訊息佇列中的若干訊息如果是對同一個資料進行操作，這些操作具有前後的關係，必須要按前後的順序執行，否則就會造成資料異常。舉例： 比如通過mysql binlog進行兩個資料庫的資料同步，由

RabbitMQ 如何保證訊息的可靠性

一條消費成功被消費經歷了生產者->MQ->消費者，因此在這三個步驟中都有可能造成訊息丟失。

【RabbitMQ】RabbitMQ 如何保證訊息的可靠傳輸

一、可靠傳輸 本篇文章主要講 RabbitMQ 如何保證訊息的可靠傳輸，所以在講RabbitMQ的實現之前，我們需要先來搞懂一個問題，就是什麼是訊息的可靠傳輸。

RabbitMq高階之如何保證訊息傳送可靠性

1.RabbitMq的傳送機制學過RabbitMq的同學們大概都知道了RabbitMq傳送機制引入了Exchange(交換機的概念),訊息傳送方,首先把訊息傳送到交換機這是第一個步驟,然後交換機在把訊息路由到不同的佇列中(Queue)這是第二個步

RocketMQ如何保證訊息的可靠性投遞？

要想保證訊息的可靠型投遞，無非保證如下3個階段的正常執行即可： 生產者將訊息成功投遞到broker

如何保證訊息的可靠性傳輸

從石杉碼農課程整理而來，後續會補圖 1、面試題 如何保證訊息的可靠性傳輸（如何處理訊息丟失的問題）？

如何保證訊息佇列的高可用？

RabbitMQ 的高可用性 RabbitMQ 是比較有代表性的，因為是基於主從（非分散式）做高可用性的，我們就以 RabbitMQ 為例子講解第一種 MQ 的高可用性怎麼實現。

Storm 系列（六）—— Storm 專案三種打包方式對比分析

一、簡介 在將 Storm Topology 提交到伺服器叢集執行時，需要先將專案進行打包。本文主要對比分析各種打包方式，並將打包過程中需要注意的事項進行說明。主要打包方式有以下三種：

RabbitMQ(三) -- 訊息與佇列進階

一：摘要概述 RabbitMQ(一) -- 初識RabbitMQ ：通過AMQP協議觸控RabbitMQ整體結構設計

RbbitMQ(五) -- 100%訊息投遞消費

一：前言概述 生產者生產訊息到消費者訊息消費，中間需要生產者將訊息傳送到交換器，再由交換器路由到佇列儲存，然後消費者進行訊息消費。在沒有任何設定情況下，中間可能存在以下幾種情況導致訊息丟失：

RabbitMQ 之訊息的可靠性消費

保障訊息的可靠性消費主要有以下兩個方面到內容 訊息在被消費端正確消費之前，不能刪除

C#呼叫RabbitMQ實現訊息佇列的示例程式碼

前言 我在剛接觸使用中介軟體的時候，發現，中介軟體的使用並不是最難的，反而是中介軟體的下載，安裝，配置才是最難的。

對比分析MySQL語句中的IN 和Exists

背景介紹 最近在寫SQL語句時，對選擇IN 還是Exists 猶豫不決，於是把兩種方法的SQL都寫出來對比一下執行效率，發現IN的查詢效率比Exists高了很多，於是想當然的認為IN的效率比Exists好，但本著尋根究底的原則，我想知

Oracle和MySQL的高可用方案對比分析

關於Oracle和MySQL的高可用方案，其實一直想要總結了，就會分為幾個系列來簡單說說。通過這樣的對比，會對兩種資料庫架構設計上的細節差異有一個基本的認識。Oracle有一套很成熟的解決方案。用我在OOW上的ppt來看，是

python Tensor和Array對比分析

如下所示： 區別 Array Tensor 型別 uint8，float32系列 {} 各型別相互轉換 uint8轉float64：image = image * (2. / 255.) - 1 float64轉uint8：image.astype(np.uint8)

Django bulk_create()、update()與資料庫事務的效率對比分析

下面以建立10000個物件為例進行測試： # 用for迴圈挨個建立，共花費37秒 for i in range(10000):