Java Spring boot整合RabbitMQ如何實現B2B2C小程序電子商務

小編給大家分享一下Java Spring boot整合RabbitMQ如何實現B2B2C小程序電子商務，希望大家閱讀完這篇文章之后都有所收獲，下面讓我們一起去探討吧！

之前我們發送和接收到隊列中的消息，現在是時候在 RabbitMQ 中引入完整的消息傳遞模式了。

讓我們快速回顧一下之前了解的內容：

生產者（producer）：發送消息的程序

隊列（queue）：存儲消息的緩沖器

消費者（consumer）：接收消息的程序

RabbitMQ 消息模型的核心理念是：發布者（producer）不會直接發送任何消息給隊列。事實上，發布者（producer）甚至不知道消息是否已經被投遞到隊列。

發布者（producer）只需要把消息發送給一個交換器（exchange）。交換器非常簡單，它一邊從發布者方接收消息，一邊把消息推送到隊列。交換器必須知道如何處理它接收到的消息，是應該推送到指定的隊列還是多個隊列，或者是直接忽略消息。這些規則是通過交換器類型（exchange type）來定義的。

有幾個可供選擇的交換器類型：direct, topic, headers 和 fanout。我們在這里主要說明最后一個 ——fanout。fanout exchange 很簡單，你可能從名字上就能猜測出來，它把消息發送給它所知道的所有隊列。這正是我們所需要的。

先創建一個 fanout 類型的交換器，命名為”tut.fanout”

＠Profile({"tut3", "pub-sub"})
@Configuration
public class Tut3Config {
 /**
     * 定義一個 Exchange
     *
     * @return FanoutExchange
     */
     @Bean
    public FanoutExchange queue() {
        return new FanoutExchange("tut.fanout");
    }
    /**
     * 消費者一端的配置：queues、bindings
     */
    @Profile("receiver")
    private static class ReceiverConfig {
        @Bean 
        public Queue autoDeleteQueue1() {
            return new AnonymousQueue();
        }
        @Bean 
        public Queue autoDeleteQueue2() {
            return new AnonymousQueue();
        }
        @Bean 
        public Binding binding1(FanoutExchange fanout, Queue autoDeleteQueue1) {
            return BindingBuilder.bind(autoDeleteQueue1).to(fanout);
        }
        @Bean 
        public Binding binding2(FanoutExchange fanout, Queue autoDeleteQueue2) {
            return BindingBuilder.bind(autoDeleteQueue2).to(fanout);
        }
        @Bean 
        public Tut3Receiver receiver() {
            return new Tut3Receiver();
        }
    }
    
@Bean 
@Profile("sender")
    public Tut3Sender sender() {
        return new Tut3Sender();
    }
}

和之前的兩個教程一樣，我們定義了兩個 profiles（tut3、pub-sub）以保證我們能運行指定的示例。然后創建了一個消費者端的配置，并在其中定義了兩個 AnonymousQueue 和兩個 Binding 以便將相應的隊列和交換器綁定起來。

交換器列表

rabbitmqctl 能夠列出服務器上所有的交換器。這個列表中有一些叫做 amq.* 的交換器。這些都是默認創建的，不過這時候你還不需要使用他們。

$ sudo rabbitmqctl list_exchanges

Listing exchanges …

logs fanout

amq.direct direct

amq.topic topic

amq.fanout fanout

amq.headers headers

…done.

未命名交換器（Nameless exchange）

在之前，我們對交換一無所知，但依然能夠將消息發送到隊列。這是怎么回事？因為我們使用了默認的交換器，它是用空字符串（””）來標識的。

我們之前是這樣發送消息的：

template.convertAndSend(queue.getName(), message);

而具體源碼如下，可以看出之前我們使用的是默認的空字符定義的交換器

// RabbitTemplate
@Override
public void convertAndSend(String routingKey, final Object object) throws AmqpException {
convertAndSend(this.exchange, routingKey, object, (CorrelationData) null);
}
private volatile String exchange = DEFAULT_EXCHANGE;
/\*_ Alias for amq.direct default exchange. _/
private static final String DEFAULT_EXCHANGE = "";

臨時隊列（Temporary queues）

我們之前使用的是具有指定名稱的隊列（hello 和 work-queues）。能夠命名隊列對我們而言至關重要 —— 我們需要將工作進程指向同一個隊列。當我們想要在生產者和消費者之間共享隊列時，給隊列一個名字很重要。

但是我們的日志記錄器并不是這樣。我們希望記錄到所有的日志消息，而不僅僅是它們的一部分。我們也只對當前的消息感興趣，而對舊的消息不感興趣。為了解決這個問題，我們需要做兩件事情。

首先，每當我們連接到 RabbitMQ，我們需要一個新的空的隊列。我們可以創建一個具有隨機名稱的隊列，或者最好讓服務器為我們選擇一個隨機的隊列名。

其次，當與消費者（consumer）斷開連接的時候，這個隊列應當被立即刪除。

在 Spring AMQP 中，我們可以使用 AnonymousQueue 來作為臨時隊列。它是一個非持久化的、獨占的、可自動刪除的隊列。

@Bean
public Queue autoDeleteQueue1() {
    return new AnonymousQueue();
}
@Bean
public Queue autoDeleteQueue2() {
    return new AnonymousQueue();
}

此時我們的隊列名字看起來會像這樣：amq.gen-JzTY20BRgKO-HjmUJj0wLg

綁定（Bindings）

我們已經創建了一個 fanout exchange 和兩個隊列，現在我們需要告訴交換所將消息發送到我們的隊列。交換器和隊列之間的關系稱為綁定。在上面的 Tut3Config 中，您可以看到我們有兩個綁定，每個 AnonymousQueue 都有一個綁定。

@Bean
public Binding binding1(FanoutExchange fanout, Queue autoDeleteQueue1) {
    return BindingBuilder.bind(autoDeleteQueue1).to(fanout);
}

綁定列表

你可以使用 rabbitmqctl list_bindings 列出所有現存的綁定。

rabbitmqctl list_bindings

代碼整合

生產者

發出消息的生產者程序與以前的沒有多大區別。最重要的變化是我們現在要發布消息給我們的 fanout exchange，而不是默認的交換器。發送時我們需要提供一個 routingKey，但是對于 fanout exchange，這個值將被忽略。

public class Tut3Sender {
@Autowird
private AmqpTemplate template;
@Override
private FanoutExchange fanout;
 private int dots = 0;
 private int count = 0;
@Scheduled(fixedDelay = 1000, initialDelay = 500)
    public void send() {
        StringBuilder builder = new StringBuilder("Hello");
        if (dots++ == 3) {
            dots = 1;
        }
        for (int i = 0; i < dots; i++) {
            builder.append('.');
        }
        builder.append(Integer.toString(++count));
        String message = builder.toString();
        template.convertAndSend(fanout.getName(), "", message);
        System.out.println(" [x] Sent '" + message + "'");
    }
}

說明：

禁止發布到不存在的交換器
如果沒有任何隊列綁定到交換器，消息將丟失
消費者 了解springcloud架構可以加求求：三五三六二四七二五九
我們定義兩個消費者分別監聽兩個隊列

public class Tut3Receiver {
    @RabbitListener(queues = "#{autoDeleteQueue1.name}")
    public void receiver1(String in) throws InterruptedException {
        receive(in, 1);
    }
    @RabbitListener(queues = "#{autoDeleteQueue2.name}")
    public void receiver2(String in) throws InterruptedException {
        receive(in, 2);
    }
    private void receive(String in, int instance) throws InterruptedException {
        StopWatch watch = new StopWatch();
        watch.start();
        System.out.println("instance " + instance + " [x] Received '" + in + "'");
        doWork(in);
        watch.stop();
        System.out.println("instance " + instance + " [x] Done in "
                + watch.getTotalTimeSeconds() + "s");
    }
    private void doWork(String in) throws InterruptedException {
        for (char ch : in.toCharArray()) {
            if (ch == '.') {
                Thread.sleep(1000);
            }
        }
    }
}

運行

maven 編譯

mvn clean package -Dmaven.test.skip=true

運行（建議先運行消費者，再運行生產者）

java -jar target/rabbitmq-tutorial-0.0.1-SNAPSHOT.jar --spring.profiles.active=tut3,receiver  --tutorial.client.duration=60000

java -jar target/rabbitmq-tutorial-0.0.1-SNAPSHOT.jar --spring.profiles.active=tut3,sender  --tutorial.client.duration=60000

輸出

// Sender

Ready … running for 60000ms

[x] Sent ‘Hello.1’

[x] Sent ‘Hello…2’

[x] Sent ‘Hello…3’

// Receiver

Ready … running for 60000ms

instance 2 [x] Received ‘Hello.1’

instance 1 [x] Received ‘Hello.1’

instance 2 [x] Done in 1.002s

instance 1 [x] Done in 1.002s

instance 2 [x] Received ‘Hello…2’

instance 1 [x] Received ‘Hello…2’

instance 2 [x] Done in 2.003s

instance 1 [x] Done in 2.003s

instance 1 [x] Received ‘Hello…3’

instance 2 [x] Received ‘Hello…3’

instance 1 [x] Done in 3.011s

instance 2 [x] Done in 3.011s

看完了這篇文章，相信你對“Java Spring boot整合RabbitMQ如何實現B2B2C小程序電子商務”有了一定的了解，如果想了解更多相關知識，歡迎關注億速云行業資訊頻道，感謝各位的閱讀！