QsB28資訊網(wǎng)——每日最新資訊28at.com

1、消息丟失問題

RabbitMQ解決消息丟失的問題：

• RabbitMQ通過消息持久化和消息確認(rèn)機制來確保消息的可靠傳遞。生產(chǎn)者可以選擇將消息標(biāo)記為持久化，使得即使在消息隊列服務(wù)器故障后，消息也能被保存并傳遞給消費者。
• RabbitMQ還提供了多種消息確認(rèn)機制，如發(fā)布確認(rèn)（Publish Confirm）和事務(wù)機制（Transaction），生產(chǎn)者可以通過這些機制獲取消息是否成功被RabbitMQ接收和處理的確認(rèn)。

RocketMQ解決消息丟失的問題：

• RocketMQ通過持久化存儲和副本機制來保證消息的可靠傳遞。消息在發(fā)送前會被持久化存儲到磁盤上，即使在消息服務(wù)器故障時也能夠恢復(fù)消息。
• RocketMQ支持多副本機制，將消息復(fù)制到多個Broker節(jié)點上，即使其中一個Broker節(jié)點發(fā)生故障，仍然可以從其他副本節(jié)點讀取和傳遞消息。

Kafka解決消息丟失的問題：

• Kafka通過持久化存儲和副本機制來保證消息的可靠傳遞。消息在發(fā)送前被持久化存儲到磁盤上，即使在服務(wù)器重啟后也不會丟失。
• Kafka采用多副本機制，將消息復(fù)制到多個Broker節(jié)點上，即使其中一個Broker節(jié)點故障，仍然可以從其他副本節(jié)點讀取和傳遞消息。

2、消息積壓問題

RabbitMQ解決消息積壓的問題：

• RabbitMQ通過調(diào)整消費者的消費速率來控制消息積壓。可以使用QoS（Quality of Service）機制設(shè)置每個消費者的預(yù)取計數(shù)，限制每次從隊列中獲取的消息數(shù)量，以提升消費者的處理速度。
• RabbitMQ還支持消費者端的流量控制，通過設(shè)置basic.qos參數(shù)來提升消費者的處理速度，避免消息過多導(dǎo)致積壓。

RocketMQ解決消息積壓的問題：

• RocketMQ通過動態(tài)提升消費者的消費速率來控制消息積壓。可以根據(jù)系統(tǒng)的負(fù)載情況和消息隊列的堆積情況，動態(tài)調(diào)整消費者的并發(fā)消費線程數(shù)，以適應(yīng)消息的處理需求。
• RocketMQ還提供了消息拉取和推拉模式，消費者可以根據(jù)自身的處理能力主動拉取消息，避免消息積壓過多。

Kafka解決消息積壓的問題：

• Kafka通過分區(qū)和副本機制來實現(xiàn)消息的并行處理和負(fù)載均衡。可以根據(jù)消息的負(fù)載情況和消費者的處理能力，通過增加分區(qū)數(shù)量、調(diào)整副本分配策略等方式來提高系統(tǒng)的處理能力。
• Kafka還提供了消息清理（compaction）和數(shù)據(jù)保留策略，可以根據(jù)時間或者數(shù)據(jù)大小來自動刪除過期的消息，避免消息積壓過多。

3、消息重復(fù)消費問題

RabbitMQ：

• 冪等性處理：在消費者端實現(xiàn)冪等性邏輯，即無論消息被消費多少次，最終的結(jié)果應(yīng)該保持一致。這可以通過在消費端進(jìn)行唯一標(biāo)識的檢查或者記錄已經(jīng)處理過的消息來實現(xiàn)。
• 消息確認(rèn)機制：消費者在處理完消息后，發(fā)送確認(rèn)消息（ACK）給RabbitMQ，告知消息已經(jīng)成功處理。RabbitMQ根據(jù)接收到的確認(rèn)消息來判斷是否需要重新投遞消息給其他消費者。

RocketMQ：

• 使用消息唯一標(biāo)識符（Message ID）：在消息發(fā)送時，為每條消息附加一個唯一標(biāo)識符。消費者在處理消息時，可以通過判斷消息唯一標(biāo)識符來避免重復(fù)消費。可以將消息ID記錄在數(shù)據(jù)庫或緩存中，用于去重檢查。
• 消費者端去重處理：消費者在消費消息時，可以通過維護(hù)一個已消費消息的列表或緩存，來避免重復(fù)消費已經(jīng)處理過的消息。

Kafka：

• 冪等性處理：在消費者端實現(xiàn)冪等性邏輯，即多次消費同一條消息所產(chǎn)生的結(jié)果與單次消費的結(jié)果一致。這可以通過在業(yè)務(wù)邏輯中引入唯一標(biāo)識符或記錄已處理消息的狀態(tài)來實現(xiàn)。
• 消息確認(rèn)機制：消費者在處理完消息后，提交已消費的偏移量（Offset）給Kafka，Kafka會記錄已提交的偏移量，以便在消費者重新啟動時從正確的位置繼續(xù)消費。消費者可以定期提交偏移量，確保消息只被消費一次。

4、消息順序性

rabbitmq 的消息順序性主要依賴于以下幾個方面：

• 單個隊列：rabbitmq 保證了同一個隊列中的消息按照發(fā)布的順序進(jìn)入和出隊。

rokcetmq 的消息順序性主要依賴于以下幾個方面：

• 有序分區(qū)：rokcetmq 保證了同一個隊列（topic + queueId）中的消息按照發(fā)布的順序存儲和消費。

kafka 的消息順序性主要依賴于以下幾個方面：

• 有序分區(qū)：kafka 保證了同一個分區(qū)（topic + partition）中的消息按照發(fā)布的順序存儲和消費。

5、事務(wù)消息

RabbitMQ的事務(wù)消息：

• RabbitMQ支持事務(wù)消息的發(fā)送和確認(rèn)。在發(fā)送消息之前，可以通過調(diào)用"channel.txSelect()"來開啟事務(wù)，然后將要發(fā)送的消息發(fā)布到交換機中。如果事務(wù)成功提交，消息將被發(fā)送到隊列，否則事務(wù)會回滾，消息不會被發(fā)送。
• 在消費端，可以通過"channel.txSelect()"開啟事務(wù)，然后使用"basicAck"手動確認(rèn)消息的處理結(jié)果。如果事務(wù)成功提交，消費端會發(fā)送ACK確認(rèn)消息的處理；否則，事務(wù)回滾，消息將被重新投遞。

public class RabbitMQTransactionDemo {    private static final String QUEUE_NAME = "transaction_queue";    public static void main(String[] args) {        try {            // 創(chuàng)建連接工廠            ConnectionFactory factory = new ConnectionFactory();            factory.setHost("localhost");            // 創(chuàng)建連接            Connection connection = factory.newConnection();            // 創(chuàng)建信道            Channel channel = connection.createChannel();            // 聲明隊列            channel.queueDeclare(QUEUE_NAME, false, false, false, null);            try {                // 開啟事務(wù)                channel.txSelect();                // 發(fā)送消息                String message = "Hello, RabbitMQ!";                channel.basicPublish("", QUEUE_NAME, null, message.getBytes());                // 提交事務(wù)                channel.txCommit();            } catch (Exception e) {                // 事務(wù)回滾                channel.txRollback();                e.printStackTrace();            }            // 關(guān)閉信道和連接            channel.close();            connection.close();        } catch (Exception e) {            e.printStackTrace();        }    }}

RocketMQ的事務(wù)消息：

• RocketMQ提供了事務(wù)消息的機制，確保消息的可靠性和一致性。發(fā)送事務(wù)消息時，需要將消息發(fā)送到半消息隊列，然后執(zhí)行本地事務(wù)邏輯。事務(wù)執(zhí)行成功后，通過調(diào)用"TransactionStatus.CommitTransaction"提交事務(wù)消息；若事務(wù)執(zhí)行失敗，則通過調(diào)用"TransactionStatus.RollbackTransaction"回滾事務(wù)消息。事務(wù)消息的最終狀態(tài)由消息生產(chǎn)者根據(jù)事務(wù)執(zhí)行結(jié)果進(jìn)行確認(rèn)。

public class RocketMQTransactionDemo {    public static void main(String[] args) throws Exception {        // 創(chuàng)建事務(wù)消息生產(chǎn)者        TransactionMQProducer producer = new TransactionMQProducer("group_name");        producer.setNamesrvAddr("localhost:9876");                // 設(shè)置事務(wù)監(jiān)聽器        producer.setTransactionListener(new TransactionListener() {            @Override            public LocalTransactionState executeLocalTransaction(Message msg, Object arg) {                // 執(zhí)行本地事務(wù)邏輯，根據(jù)業(yè)務(wù)邏輯結(jié)果返回相應(yīng)的狀態(tài)                // 返回 LocalTransactionState.COMMIT_MESSAGE 表示事務(wù)提交                // 返回 LocalTransactionState.ROLLBACK_MESSAGE 表示事務(wù)回滾                // 返回 LocalTransactionState.UNKNOW 表示事務(wù)狀態(tài)未知            }            @Override            public LocalTransactionState checkLocalTransaction(MessageExt msg) {                // 根據(jù)消息的狀態(tài)，來判斷本地事務(wù)的最終狀態(tài)                // 返回 LocalTransactionState.COMMIT_MESSAGE 表示事務(wù)提交                // 返回 LocalTransactionState.ROLLBACK_MESSAGE 表示事務(wù)回滾                // 返回 LocalTransactionState.UNKNOW 表示事務(wù)狀態(tài)未知            }        });                // 啟動事務(wù)消息生產(chǎn)者        producer.start();        // 構(gòu)造消息        Message msg = new Message("topic_name", "tag_name", "Hello, RocketMQ!".getBytes());        // 發(fā)送事務(wù)消息        TransactionSendResult sendResult = producer.sendMessageInTransaction(msg, null);        System.out.println("Send Result: " + sendResult);        // 關(guān)閉事務(wù)消息生產(chǎn)者        producer.shutdown();    }}

Kafka的事務(wù)消息：

• Kafka引入了事務(wù)功能來確保消息的原子性和一致性。事務(wù)消息的發(fā)送和確認(rèn)在生產(chǎn)者端進(jìn)行。生產(chǎn)者可以通過初始化事務(wù)，將一系列的消息寫入事務(wù)，然后通過"commitTransaction()"提交事務(wù)，或者通過"abortTransaction()"中止事務(wù)。Kafka會保證在事務(wù)提交之前，寫入的所有消息不會被消費者可見，以保持事務(wù)的一致性。

public class KafkaTransactionDemo {    public static void main(String[] args) {        Properties props = new Properties();        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());        props.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, "transactional_id");        Producer<String, String> producer = new KafkaProducer<>(props);        // 初始化事務(wù)        producer.initTransactions();        try {            // 開啟事務(wù)            producer.beginTransaction();            // 發(fā)送消息            ProducerRecord<String, String> record = new ProducerRecord<>("topic_name", "Hello, Kafka!");            producer.send(record);            // 提交事務(wù)            producer.commitTransaction();        } catch (ProducerFencedException e) {            // 處理異常情況            producer.close();        } finally {            producer.close();        }    }}

6、ACK機制

RabbitMQ的ACK機制：

RabbitMQ使用ACK（消息確認(rèn)）機制來確保消息的可靠傳遞。消費者收到消息后，需要向RabbitMQ發(fā)送ACK來確認(rèn)消息的處理狀態(tài)。只有在收到ACK后，RabbitMQ才會將消息標(biāo)記為已成功傳遞，否則會將消息重新投遞給其他消費者或者保留在隊列中。QsB28資訊網(wǎng)——每日最新資訊28at.com

以下是RabbitMQ ACK的Java示例：QsB28資訊網(wǎng)——每日最新資訊28at.com

public class RabbitMQAckDemo {    public static void main(String[] args) throws Exception {        // 創(chuàng)建連接工廠        ConnectionFactory factory = new ConnectionFactory();        factory.setHost("localhost");        // 創(chuàng)建連接        Connection connection = factory.newConnection();        // 創(chuàng)建信道        Channel channel = connection.createChannel();        // 聲明隊列        String queueName = "queue_name";        channel.queueDeclare(queueName, false, false, false, null);        // 創(chuàng)建消費者        String consumerTag = "consumer_tag";        boolean autoAck = false; // 關(guān)閉自動ACK        // 消費消息        channel.basicConsume(queueName, autoAck, consumerTag, 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");                System.out.println("Received message: " + message);                try {                    // 模擬處理消息的業(yè)務(wù)邏輯                    processMessage(message);                    // 手動發(fā)送ACK確認(rèn)消息                    long deliveryTag = envelope.getDeliveryTag();                    channel.basicAck(deliveryTag, false);                } catch (Exception e) {                    // 處理消息異常，可以選擇重試或者記錄日志等操作                    System.out.println("Failed to process message: " + message);                    e.printStackTrace();                    // 手動發(fā)送NACK拒絕消息，并可選是否重新投遞                    long deliveryTag = envelope.getDeliveryTag();                    boolean requeue = true; // 重新投遞消息                    channel.basicNack(deliveryTag, false, requeue);                }            }        });    }    private static void processMessage(String message) {        // 模擬處理消息的業(yè)務(wù)邏輯    }}

RocketMQ的ACK機制：

RocketMQ的ACK機制由消費者控制，消費者從消息隊列中消費消息后，可以手動發(fā)送ACK確認(rèn)消息的處理狀態(tài)。只有在收到ACK后，RocketMQ才會將消息標(biāo)記為已成功消費，否則會將消息重新投遞給其他消費者。QsB28資訊網(wǎng)——每日最新資訊28at.com

以下是RocketMQ ACK的Java示例：

public class RocketMQAckDemo {    public static void main(String[] args) throws Exception {        // 創(chuàng)建消費者        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("group_name");        consumer.setNamesrvAddr("localhost:9876");        // 訂閱消息        consumer.subscribe("topic_name", "*");        // 注冊消息監(jiān)聽器        consumer.registerMessageListener((MessageListenerConcurrently) (msgs, context) -> {            for (MessageExt message : msgs) {                try {                    // 消費消息                    String msgBody = new String(message.getBody(), "UTF-8");                    System.out.println("Received message: " + msgBody);                    // 模擬處理消息的業(yè)務(wù)邏輯                    processMessage(msgBody);                    // 手動發(fā)送ACK確認(rèn)消息                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;                } catch (Exception e) {                    // 處理消息異常，可以選擇重試或者記錄日志等操作                    System.out.println("Failed to process message: " + new String(message.getBody()));                    e.printStackTrace();                    return ConsumeConcurrentlyStatus.RECONSUME_LATER;                }            }            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;        });        // 啟動消費者        consumer.start();    }    private static void processMessage(String message) {        // 模擬處理消息的業(yè)務(wù)邏輯    }}

Kafka的ACK機制：

Kafka的ACK機制用于控制生產(chǎn)者在發(fā)送消息后，需要等待多少個副本確認(rèn)才視為消息發(fā)送成功。這個機制可以通過設(shè)置acks參數(shù)來進(jìn)行配置。QsB28資訊網(wǎng)——每日最新資訊28at.com

在Kafka中，acks參數(shù)有三個可選值：

• acks=0：生產(chǎn)者在發(fā)送消息后不需要等待任何確認(rèn)，直接將消息發(fā)送給Kafka集群。這種方式具有最高的吞吐量，但是也存在數(shù)據(jù)丟失的風(fēng)險，因為生產(chǎn)者不會知道消息是否成功發(fā)送給任何副本。
• acks=1：生產(chǎn)者在發(fā)送消息后只需要等待首領(lǐng)副本（leader replica）確認(rèn)。一旦首領(lǐng)副本成功接收到消息，生產(chǎn)者就會收到確認(rèn)。這種方式提供了一定的可靠性，但是如果首領(lǐng)副本在接收消息后但在確認(rèn)之前發(fā)生故障，仍然可能會導(dǎo)致數(shù)據(jù)丟失。
• acks=all：生產(chǎn)者在發(fā)送消息后需要等待所有副本都確認(rèn)。只有當(dāng)所有副本都成功接收到消息后，生產(chǎn)者才會收到確認(rèn)。這是最安全的確認(rèn)機制，確保了消息不會丟失，但是需要更多的時間和資源。acks=-1與acks=all是等效的。

下面是一個使用Java編寫的Kafka生產(chǎn)者示例代碼：QsB28資訊網(wǎng)——每日最新資訊28at.com

public class KafkaProducerDemo {    public static void main(String[] args) {        // 配置Kafka生產(chǎn)者的參數(shù)        Properties props = new Properties();        props.put("bootstrap.servers", "localhost:9092"); // Kafka集群的地址和端口        props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer"); // 鍵的序列化器        props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer"); // 值的序列化器        props.put("acks", "all"); // 設(shè)置ACK機制為所有副本都確認(rèn)        // 創(chuàng)建生產(chǎn)者實例        KafkaProducer<String, String> producer = new KafkaProducer<>(props);        // 構(gòu)造消息        String topic = "my_topic";        String key = "my_key";        String value = "Hello, Kafka!";        // 創(chuàng)建消息記錄        ProducerRecord<String, String> record = new ProducerRecord<>(topic, key, value);        // 發(fā)送消息        producer.send(record, new Callback() {            @Override            public void onCompletion(RecordMetadata metadata, Exception exception) {                if (exception != null) {                    System.err.println("發(fā)送消息出現(xiàn)異常：" + exception.getMessage());                } else {                    System.out.println("消息發(fā)送成功！位于分區(qū) " + metadata.partition() + "，偏移量 " + metadata.offset());                }            }        });        // 關(guān)閉生產(chǎn)者        producer.close();    }}

本文鏈接：http://www.www897cc.com/showinfo-26-10561-0.htmlMQ黃金三劍客 Rabbit Rocket Kafka深入解密常見問題及功能對比指南

聲明：本網(wǎng)頁內(nèi)容旨在傳播知識，若有侵權(quán)等問題請及時與本網(wǎng)聯(lián)系，我們將在第一時間刪除處理。郵件：2376512515@qq.com

上一篇： Github的一個奇技淫巧，你學(xué)會了嗎？

下一篇： 圖解「正向代理」的原理 + 實踐應(yīng)用