RabbitMQ模式，RabbitMQ和springboot整合，RabbitMQ全鏈路消息不丟失解決

一，RabbitMQ簡介

　　RabbitMQ是實現了高級消息隊列協議（AMQP）的開源消息代理軟件（亦稱面向消息的中間件）。RabbitMQ服務器是用Erlang語言編寫的，而集群和故障轉移是構建在開放電信平台框架上的。所有主要的編程語言均有與代理接口通訊的客戶端庫。

　　優點：跨語言，高並發。

二，實際項目中作用

（一）任務異步處理

場景：用戶下單后，根據訂單信息進行減庫存以及增加積分。

傳統：必須要整個流程走完。

如今：下完單，用戶操作就可以結束了，不必等積分增加完。

（二） 應用程序解耦

場景：下單減庫存。

傳統模式的缺點：當前是基於同步調用方式，訂單服務與庫存服務處於耦合狀態，假設庫存服務狀態異常，則訂單服務在調用庫存服務時，也會出現失敗。

如今：訂單服務：將用戶訂單信息寫入訂單數據庫，並向消息隊列發送訂單消息，成功發送之后，向用戶返回下單成功。

庫存服務：監聽消息隊列，發現消息隊列中存在訂單消息，則取回，根據訂單信息進行庫存操作

此時兩個服務之前不存在耦合關系，兩個服務之間基於消息隊列完成消息傳遞。

（三）流量削峰

場景：秒殺或搶購活動，

傳統：很有可能因為出現瞬時流量暴增，導致后端服務無法承受這些流量，最終導致服務宕機。

如今：架設消息隊列，對流量進行限流計數。可以控制前端請求數量。

三，幾種常見模式案例

1.工作對列模式：一個producer生產者

work queues兩個消費端，兩個消費端共同消費同一個隊列中的消息。

應用場景：對於 任務過重或任務較多情況使用工作隊列可以提高任務處理的速度。

測試：

1、使用入門程序，啟動多個消費者。

2、生產者發送多個消息。

生產者：

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 QuickProducer {
    public static final String QUEUE="helloworld";
    public static  void main(String[] args){
        Connection connection =null;
        Channel channel = null;

        try{

            //聲明連接工廠
            ConnectionFactory connectionFactory = new ConnectionFactory();
            //設置ip
            connectionFactory.setHost("127.0.0.1");
            //設置端口
            connectionFactory.setPort(5672);
            //設置用戶名
            connectionFactory.setUsername("guest");
            //設置密碼
            connectionFactory.setPassword("guest");
            //設置虛擬主機
            //虛擬主機：當安裝好了一個rabbitMQ服務之后，內部可以去設置多個虛擬的rabbitmq，對於虛擬主機的設置是通過名稱進行划分的。且相互獨立
            connectionFactory.setVirtualHost("/");

            //聲明連接對象與通道對象
            connection = connectionFactory.newConnection();
            channel = connection.createChannel();

            //聲明隊列
            /**
             * String queue, :隊列名稱
             * boolean durable, :隊列是否持久化。當為true，隊列信息會被保存到磁盤
             * boolean exclusive, :是否獨占此連接，當為true，則此連接中只有這一個隊列，當為false，可以存在多個
             * boolean autoDelete,:是否自動刪除,當為true，此隊列一旦沒有消費者監聽，則自動從rabbitmq中刪除，當為false，則不會
             * Map<String, Object> arguments:隊列的屬性設置
             */
            channel.queueDeclare(QUEUE,true,false,false,null);

            //發送消息
            /**
             * String exchange, ：交換機
             * String routingKey, :路由key
             * BasicProperties props,：消息的屬性設置
             * byte[] body：被發送的消息
             */
            String message ="第三條消息";
            channel.basicPublish("",QUEUE,null,message.getBytes());

        }catch (Exception e){
            e.printStackTrace();
        }finally {
            try {
                channel.close();
            } catch (IOException e) {
                e.printStackTrace();
            } catch (TimeoutException e) {
                e.printStackTrace();
            }
            try {
                connection.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }
}

 

消費者1：

import com.rabbitmq.client.*;

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

public class QuickConsumer {
    public static final String QUEUE="helloworld";

    public static void main(String[] args) throws IOException, TimeoutException {

        Connection connection = null;
        Channel channel = null;

        //聲明連接工廠
        ConnectionFactory connectionFactory = new ConnectionFactory();
        connectionFactory.setHost("127.0.0.1");
        connectionFactory.setPort(5672);
        connectionFactory.setUsername("guest");
        connectionFactory.setPassword("guest");
        connectionFactory.setVirtualHost("/");

        //聲明連接對象與通道對象
        connection = connectionFactory.newConnection();
        channel = connection.createChannel();



        Consumer consumer = new DefaultConsumer(channel){
            @Override
            /**
             * String consumerTag, : 消費者標識
             * Envelope envelope,：信封
             * AMQP.BasicProperties properties,:默認屬性值
             * byte[] body：被消費的數據
             */
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                String value = new String(body);
                System.out.println("消費者一號"+value);
            }
        };

        //設置監聽
        /**
         * String queue,:被監聽的隊列名稱
         * boolean autoAck,:是否開啟自動應答。當為true， 。。當為false，需要開發人員手動的編碼去通知rabbitmq
         * Consumer callback:獲取消息的回調方法
         */
        channel.basicConsume(QUEUE,true,consumer);
    }
}

 

消費者2：

public class QuickConsumer2 {
    public static final String QUEUE="helloworld";

    public static void main(String[] args) throws IOException, TimeoutException {

        Connection connection = null;
        Channel channel = null;

        //聲明連接工廠
        ConnectionFactory connectionFactory = new ConnectionFactory();
        connectionFactory.setHost("127.0.0.1");
        connectionFactory.setPort(5672);
        connectionFactory.setUsername("guest");
        connectionFactory.setPassword("guest");
        connectionFactory.setVirtualHost("/");

        //聲明連接對象與通道對象
        connection = connectionFactory.newConnection();
        channel = connection.createChannel();



        Consumer consumer = new DefaultConsumer(channel){
            @Override
            /**
             * String consumerTag, : 消費者標識
             * Envelope envelope,：信封
             * AMQP.BasicProperties properties,:默認屬性值
             * byte[] body：被消費的數據
             */
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                String value = new String(body);
                System.out.println("消費者二號"+value);
            }
        };

        //設置監聽
        /**
         * String queue,:被監聽的隊列名稱
         * boolean autoAck,:是否開啟自動應答。當為true， 。。當為false，需要開發人員手動的編碼去通知rabbitmq
         * Consumer callback:獲取消息的回調方法
         */
        channel.basicConsume(QUEUE,true,consumer);
    }
}

結果：

默認輪詢策略

消費者一號第一條消息

消費者二號第二條消息

消費者一號第三條消息

2.發布、訂閱模式

Publish/subscribe

生產：

import com.rabbitmq.client.BuiltinExchangeType;
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 PublishProducer {

    //隊列名稱
    public static final String QUEUE_INFORM_EMAIL ="queue_inform_email";
    public static final String QUEUE_INFORM_SMS ="queue_inform_sms";

    //交換機名稱
    private static final String EXCHANGE_FANOUT_INFORM="exchange_fanout_inform";

    public static void main(String[] args) {

        Connection connection = null;
        Channel channel = null;

        try{

            //聲明連接工廠
            ConnectionFactory connectionFactory = new ConnectionFactory();
            connectionFactory.setHost("127.0.0.1");
            connectionFactory.setPort(5672);
            connectionFactory.setUsername("guest");
            connectionFactory.setPassword("guest");
            connectionFactory.setVirtualHost("/");

            //聲明連接對象與通道對象
            connection = connectionFactory.newConnection();
            channel = connection.createChannel();

            //聲明隊列
            channel.queueDeclare(QUEUE_INFORM_EMAIL,true,false,false,null);
            channel.queueDeclare(QUEUE_INFORM_SMS,true,false,false,null);

            //聲明交換機
            /**
             * name:交換機名稱
             * type: 交換機的類型  當工作模式為發布/訂閱模式，交換機的類型：fanount
             */
            channel.exchangeDeclare(EXCHANGE_FANOUT_INFORM, BuiltinExchangeType.FANOUT);

            //隊列綁定交換機
            channel.queueBind(QUEUE_INFORM_EMAIL,EXCHANGE_FANOUT_INFORM,"");
            channel.queueBind(QUEUE_INFORM_SMS,EXCHANGE_FANOUT_INFORM,"");

            //生產者會將消息發送到交換機，交換機會按照一定的規則將消息轉發給一個或多個隊列

            //發送消息
            channel.basicPublish(EXCHANGE_FANOUT_INFORM,"",null,"publish message".getBytes());

        }catch (Exception e){
            e.printStackTrace();
        }finally {
            try {
                channel.close();
            } catch (IOException e) {
                e.printStackTrace();
            } catch (TimeoutException e) {
                e.printStackTrace();
            }
            try {
                connection.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }
}

 

消費：

import com.rabbitmq.client.*;

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

public class PublishConsumer {

    //隊列名稱
    public static final String QUEUE_INFORM_EMAIL ="queue_inform_email";
    public static final String QUEUE_INFORM_SMS ="queue_inform_sms";

    public static void main(String[] args) throws IOException, TimeoutException {

        Connection connection = null;
        Channel channel = null;

        //聲明連接工廠
        ConnectionFactory connectionFactory = new ConnectionFactory();
        connectionFactory.setHost("127.0.0.1");
        connectionFactory.setPort(5672);
        connectionFactory.setUsername("guest");
        connectionFactory.setPassword("guest");
        connectionFactory.setVirtualHost("/");

        //聲明連接對象與通道對象
        connection = connectionFactory.newConnection();
        channel = connection.createChannel();


        //設置監聽
        Consumer consumer = new DefaultConsumer(channel){
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                String value = new String(body);
                System.out.println(value);
            }
        };
        channel.basicConsume( QUEUE_INFORM_EMAIL,true,consumer);

    }
}

 

結果：

當

channel.basicConsume( QUEUE_INFORM_EMAIL,true,consumer);

 

當

channel.basicConsume( QUEUE_INFORM_SMS,true,consumer);

 

 

總結

發布訂閱模式：

1、每個消費者監聽自己的隊列。

2、生產者將消息發給broker，由交換機將消息轉發到綁定此交換機的每個隊列，每個綁定交換機的隊列都將接收到消息

3. Routing路由模式

路由模式：

1、每個消費者監聽自己的隊列，並且設置routingkey。

2、生產者將消息發給交換機，由交換機根據routingkey來轉發消息到指定的隊列。

生產者：

package com.itheima.producer;

import com.rabbitmq.client.BuiltinExchangeType;
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 RoutingProducer {

    //隊列名稱
    private static final String QUEUE_ROUTING_EMAIL = "queue_routing_email";
    private static final String QUEUE_ROUTING_SMS = "queue_routing_sms";

    //交換機名稱
    private static final String EXCHANGE_ROUTING_INFORM="exchange_routing_inform";

    public static void main(String[] args) {

        Connection connection = null;
        Channel channel = null;

        try{

            //聲明連接工廠
            //聲明連接工廠
            ConnectionFactory connectionFactory = new ConnectionFactory();
            connectionFactory.setHost("127.0.0.1");
            connectionFactory.setPort(5672);
            connectionFactory.setUsername("guest");
            connectionFactory.setPassword("guest");
            connectionFactory.setVirtualHost("/");

            //聲明連接對象與通道對象
            connection = connectionFactory.newConnection();
            channel = connection.createChannel();

            //聲明隊列
            channel.queueDeclare(QUEUE_ROUTING_EMAIL,true,false,false,null);
            channel.queueDeclare(QUEUE_ROUTING_SMS,true,false,false,null);

            //聲明交換機
            //當使用路由模式，交換機類型必須為direct
            channel.exchangeDeclare(EXCHANGE_ROUTING_INFORM, BuiltinExchangeType.DIRECT);

            //隊列綁定交換機
            channel.queueBind(QUEUE_ROUTING_EMAIL,EXCHANGE_ROUTING_INFORM,"info");
            channel.queueBind(QUEUE_ROUTING_SMS,EXCHANGE_ROUTING_INFORM,"info");

            //發送消息
            channel.basicPublish(EXCHANGE_ROUTING_INFORM,"info",null,"info message".getBytes());

        }catch (Exception e){
            e.printStackTrace();
        }finally {
            try {
                channel.close();
            } catch (IOException e) {
                e.printStackTrace();
            } catch (TimeoutException e) {
                e.printStackTrace();
            }
            try {
                connection.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }
}

 

消費者：

import com.rabbitmq.client.*;

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

public class RoutingConsumer {

    //隊列名稱
    private static final String QUEUE_ROUTING_EMAIL = "queue_routing_email";
    private static final String QUEUE_ROUTING_SMS = "queue_routing_sms";

    public static void main(String[] args) throws IOException, TimeoutException {

        Connection connection = null;
        Channel channel = null;

        //聲明連接工廠
        ConnectionFactory connectionFactory = new ConnectionFactory();
        connectionFactory.setHost("127.0.0.1");
        connectionFactory.setPort(5672);
        connectionFactory.setUsername("guest");
        connectionFactory.setPassword("guest");
        connectionFactory.setVirtualHost("/");

        //聲明連接對象與通道對象
        connection = connectionFactory.newConnection();
        channel = connection.createChannel();


        //設置監聽
        Consumer consumer = new DefaultConsumer(channel){
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                String value = new String(body);
                System.out.println(value);
            }
        };
        channel.basicConsume(QUEUE_ROUTING_EMAIL,true,consumer);

    }
}

 

結果：

info message
info message

 

4.通配符模式

特點：

1、每個消費者監聽自己的隊列，並且設置帶統配符的routingkey。

2、生產者將消息發給broker，由交換機根據routingkey來轉發消息到指定的隊列

生產者：

import com.rabbitmq.client.BuiltinExchangeType;
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 TopicProducer {

    //隊列名稱
    public static final String QUEUE_EMAIL="queue_topic_email";
    public static final String QUEUE_SMS="queue_topic_sms";

    //交換機名稱
    public static final String EXCHANGE_TOPIC_INFORM="exchange_topic_inform";

    //路由key
    public static final String ROUTING_EMAIL="inform.#.email.#"; // inform.email
    public static final String ROUTING_SMS="inform.#.sms.#";//

    public static void main(String[] args) {

        Connection connection = null;

        Channel channel = null;

        try{

            //聲明連接工廠
            //聲明連接工廠
            ConnectionFactory connectionFactory = new ConnectionFactory();
            connectionFactory.setHost("127.0.0.1");
            connectionFactory.setPort(5672);
            connectionFactory.setUsername("guest");
            connectionFactory.setPassword("guest");
            connectionFactory.setVirtualHost("/");

            //聲明連接對象與通道對象
            connection = connectionFactory.newConnection();
            channel = connection.createChannel();

            //聲明隊列
            channel.queueDeclare(QUEUE_EMAIL,true,false,false,null);
            channel.queueDeclare(QUEUE_SMS,true,false,false,null);

            //聲明交換機
            //當使用通配符模式，交換機類型必須為topic
            channel.exchangeDeclare(EXCHANGE_TOPIC_INFORM, BuiltinExchangeType.TOPIC);

            //隊列綁定交換機
            channel.queueBind(QUEUE_EMAIL,EXCHANGE_TOPIC_INFORM,ROUTING_EMAIL);
            channel.queueBind(QUEUE_SMS,EXCHANGE_TOPIC_INFORM,ROUTING_SMS);

            //發送消息
            //發送消息到郵件隊列
            //channel.basicPublish(EXCHANGE_TOPIC_INFORM,"inform.email",null,"email message".getBytes());

            //發送消息到短信隊列
            //channel.basicPublish(EXCHANGE_TOPIC_INFORM,"inform.sms",null,"sms message".getBytes());

            //發送消息到郵件與短信隊列
            channel.basicPublish(EXCHANGE_TOPIC_INFORM,"inform.sms.email",null,"sms email message".getBytes());
        }catch(Exception e)

        {
            e.printStackTrace();
        }finally{
            try {
                channel.close();
            } catch (IOException e) {
                e.printStackTrace();
            } catch (TimeoutException e) {
                e.printStackTrace();
            }
            try {
                connection.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }
}

 

消費者：

import com.rabbitmq.client.*;

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

public class TopicConsumer {

    //隊列名稱
    public static final String QUEUE_EMAIL="queue_topic_email";
    public static final String QUEUE_SMS="queue_topic_sms";

    public static void main(String[] args) throws IOException, TimeoutException {

        Connection connection = null;
        Channel channel = null;

        //聲明連接工廠
        ConnectionFactory connectionFactory = new ConnectionFactory();
        connectionFactory.setHost("127.0.0.1");
        connectionFactory.setPort(5672);
        connectionFactory.setUsername("guest");
        connectionFactory.setPassword("guest");
        connectionFactory.setVirtualHost("/");

        //聲明連接對象與通道對象
        connection = connectionFactory.newConnection();
        channel = connection.createChannel();


        //設置監聽
        Consumer consumer = new DefaultConsumer(channel){
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                String value = new String(body);
                System.out.println(value);
            }
        };
        channel.basicConsume(QUEUE_SMS,true,consumer);

    }
}

 

結果：

sms email message

總結幾種模式：

四，springboot整合RabbitMQ

1.消費者config

import org.springframework.amqp.core.*;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
public class RabbitMQConfig {

    public static final String QUEUE_INFORM_EMAIL = "queue_inform_email";
    public static final String QUEUE_INFORM_SMS = "queue_inform_sms";

    public static final String EXCHANGE_TOPICS_INFORM="exchange_topics_inform";


    //聲明隊列
    @Bean(QUEUE_INFORM_EMAIL)
    public Queue QUEUE_INFORM_EMAIL(){
        Queue queue = new Queue(QUEUE_INFORM_EMAIL);
        return queue;
    }

    @Bean(QUEUE_INFORM_SMS)
    public Queue QUEUE_INFORM_SMS(){
        Queue queue = new Queue(QUEUE_INFORM_SMS);
        return queue;
    }

    //聲明交換機
    @Bean(EXCHANGE_TOPICS_INFORM)
    public Exchange EXCHANGE_TOPICS_INFORM(){
        return ExchangeBuilder.topicExchange(EXCHANGE_TOPICS_INFORM).durable(true).build();
    }

    //隊列綁定交換機
    @Bean
    public Binding BIND_QUEUE_INFORM_EMAIL(@Qualifier(QUEUE_INFORM_EMAIL)Queue queue,@Qualifier(EXCHANGE_TOPICS_INFORM)Exchange exchange){
        return BindingBuilder.bind(queue).to(exchange).with(QUEUE_INFORM_EMAIL).noargs();
    }

    @Bean
    public Binding BIND_QUEUE_INFORM_SMS(@Qualifier(QUEUE_INFORM_SMS)Queue queue,@Qualifier(EXCHANGE_TOPICS_INFORM)Exchange exchange){
        return BindingBuilder.bind(queue).to(exchange).with(QUEUE_INFORM_SMS).noargs();
    }

}

消費者config

import org.springframework.amqp.core.Queue;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
public class RabbitMQConfig {

    public static final String QUEUE_INFORM_EMAIL = "queue_inform_email";
    public static final String QUEUE_INFORM_SMS = "queue_inform_sms";


    //聲明隊列
    @Bean(QUEUE_INFORM_EMAIL)
    public Queue QUEUE_INFORM_EMAIL(){
        Queue queue = new Queue(QUEUE_INFORM_EMAIL);
        return queue;
    }

    @Bean(QUEUE_INFORM_SMS)
    public Queue QUEUE_INFORM_SMS(){
        Queue queue = new Queue(QUEUE_INFORM_SMS);
        return queue;
    }

}

 

監聽消費者：

import com.itheima.springboot.consumer.config.RabbitMQConfig;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

@Component
public class MyMessageConsumer {

    //設置監聽方法 ， 每一個監聽方法可以去設置監聽一個或多個隊列
    @RabbitListener(queues = {RabbitMQConfig.QUEUE_INFORM_EMAIL})
    public void receivceMessage(String message){
        System.out.println(message);
    }
}

 

五，全鏈路消息不丟失

TPS,QPS是指標。

1.生產者宕機：

情形一：Q：當訂單服務發送的消息成功到達了rabbitmq並保存在內存中，但是此消息在沒有被消費者拿走之前，rabbitmq宕機了此時應該怎么辦？

A：此時可以考慮進行持久化操作，此處需要將==隊列與消息==都進行持久化操作，將這兩部分信息寫入磁盤。

情形二; Q：當rabbitmq未將消息寫入到磁盤時，消息隊列宕機了，怎么辦？

A：此時需要保證生產者發送的消息，rabbitmq會一定成功的進行隊列與消息持久化，否則進行重發。

 

​ RabbitMQ針對生產者投遞數據丟失,已經提供了兩種解決機制, 分別是 ==重量級的事務機制== 與 ==輕量級的confirm機制==.

選擇confirm機制：（事務同步效率低）：

confirm模式需要基於channel進行設置, 一旦某條消息被投遞到隊列之后,消息隊列就會發送一個確認信息給生產者。

情形三：高並發下生產者

　　1.阿里sql，對比mysql性能提高15%，對比nosql數據可以保存在磁盤不用考慮在內存空間的大小。生產者實例收到一個消息ack之后，就從kv存儲中刪除這條臨時消息；收到一個消息nack之后，就從kv存儲提取這條消息然后重新投遞一次即可；也可以自己對kv存儲里的消息做監控，如果超過一定時長沒收到ack，就主動重發消息。（生產者有個記錄表，每個一段時間掃描此表）。

　　2.預抓取總數100-300，根據壓力測試臨界值，閾值向下減100左右來設定，不可過大不可過小（堆積，效率低）。

2.消費者宕機

情形一     

消費者成功接收到消息之后，可以自動的向rabbitMQ返回一個應答信息，通知rabbitMQ此條消息已經被成功的接收，當rabbitMQ接收到這條消息之后，則會將此條消息刪除。這叫做自動ACK（自動應答）。

可以將自動應答改為手動應答，另外提高效率還可以批量ACK，集中返回此時當消費者接收到消息，不會馬上自動向消息隊列發送應答消息，而是需要開發人員手動編碼發送應答消息， 從而保證消息隊列不會自動刪除這條消息，而是等到消費者返回ACK確認消息才會進行刪除。（開發人員要做的就是當且只有庫存服務執行完畢並調用成功物流系統之后才會向消息隊列對返回一條確認消息,當消息隊列接收到這條消息之后才刪除此消息），現某個庫存服務(消費者)宕機了. 那么就會將這個訂單消息發送給其他的庫存服務實例(消費者)去使用這條消息.

情形二；

A消費者宕機后，另一台消費者B機器發貨，A重啟，又發之前沒發的貨，發了兩遍。

解決方案：避免重復發貨，發貨時判斷狀態碼（冪等性）。

六。冪等性

1冪等性：

​ 對於相同的一次請求操作，不管操作多少次，得到的結果都是相同的。

2場景:

​ 在開啟消費者手動ack的情況下, 庫存服務的消費者已經接收到了消息,並調用物流系統並成功更改狀態進行發送, 但是在ack返回消息的時候, 這個消費者服務可能突然宕機, 因為消費者ack未返回, 所以會導致這個消息一直在MQ中, 當消費者重新啟動就會又接收這個消息進行發貨, 這樣的話,就會導致相同的貨品發貨了兩次!!!

3解決方案:

​ 1）狀態判斷：在消費者接收消息執行之前, 先根據消息的標識信息查詢DB,判斷狀態, 如果為已發貨/待發貨狀態的話,則直接ack返回MQ,刪除這條消息

​ 2）樂觀鎖

七，有序性

對於上述流程因為有多個消費者共同監聽了同一個隊列，每一個隊列都會獲取到不同消息，但是消費者的執行時間不盡相同，因此會導致數據的無序性。

​ 將上述流程改造為只有一個消費者監聽此隊列，或這些消息只會推送給一個消費者即可。