Java Spring boot 整合RabbitMQ（二）：工作隊列（Work queues）-B2B2C小程序電子商務

現在，我們將發送一些字符串，把這些字符串當作復雜的任務。我們并沒有一個真實的復雜任務，類似于圖片大小被調整或 pdf 文件被渲染，所以我們通過 sleep () 方法來模擬這種情況。我們在字符串中加上點號（.）來表示任務的復雜程度，一個點（.）將會耗時 1 秒鐘。比如 “Hello…” 就會耗時 3 秒鐘。

如果您尚未設置項目，請參閱第一個教程中的設置。我們將遵循與第一個教程相同的模式：創建一個包（tut2）并創建 Tut2Config、Tut2Receiver 和 Tut2Sender。

代碼整合

首先創建一個新的包（tut2），我們將在這里放置我們的三個類。在配置類 Tut2Config 中，我們設置了兩個配置文件 ——tut2 和 work-queues。我們利用 Spring 來將隊列 Queue 暴露為一個 bean。我們配置消費者，并定義兩個 bean 以對應于上圖中的工作進程 receiver1 和 receiver2。

配置類

@Profile({"tut2", "work-queues"})
@Configuration
public class Tut2Config {
    @Bean
    public Queue queue() {
        return new Queue("work-queues");
    }
    /**
     * 定義兩個消費者，并且給了他們不同的標識
     */
  @Profile ("receiver")
    private static class ReceiverConfig {
        @Bean
        public Tut2Receiver receiver1() {
            return new Tut2Receiver(1);
        }
        @Bean
        public Tut2Receiver receiver2() {
            return new Tut2Receiver(2);
        }
    }
   　@Profile("sender")
   　 @Bean
　　 public Tut2Sender sender() {
        return new Tut2Sender();
    }
}

生產者
我們簡單修改一下生產者的代碼，以添加點號（.）的方式來人為的增加該任務的時長，字符串中的每個點號（.）都會增加 1s 的耗時。

public class Tut2Sender {
 @Autowired
 private AmqpTemplate template;
 @Autowired   
 private Queue queue;
    int dots = 0;
    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(queue.getName(), message);
        System.out.println(" [x] Sent '" + message + "'");
    }
}

消費者
我們的消費者 Tut2Receiver 通過 doWork () 方法模擬了一個耗時的虛假任務，它需要為消息體中每一個點號（.）模擬 1 秒鐘的操作。并且我們為消費者增加了一個實例編號，以知道是哪個實例消費了消息和處理的時長。

@RebbitListener(queues = "work-queues")
public class Tut2Receiver {
    private int instance;
    public Tut2Receiver(int instance) {
        this.instance = instance;
    }
    @RabbitHandler
    public void receive(String in) throws InterruptedException {
        StopWatch watch = new StopWatch();
        watch.start();
        System.out.println("instance " + this.instance +
                " [x] Received '" + in + "'");
        doWork(in);
        watch.stop();
        System.out.println("instance " + this.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=tut2,sender --tutorial.client.duration=60000

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

輸出

// Sender

Ready … running for 10000ms

[x] Sent ‘Hello.1’

[x] Sent ‘Hello…2’

[x] Sent ‘Hello…3’

[x] Sent ‘Hello.4’

[x] Sent ‘Hello…5’

[x] Sent ‘Hello…6’

[x] Sent ‘Hello.7’

[x] Sent ‘Hello…8’

[x] Sent ‘Hello…9’

// Receiver

Ready … running for 10000ms

instance 1 [x] Received ‘Hello.1’

instance 2 [x] Received ‘Hello…2’

instance 1 [x] Done in 1.005s

instance 1 [x] Received ‘Hello…3’

instance 2 [x] Done in 2.007s

instance 2 [x] Received ‘Hello.4’

instance 2 [x] Done in 1.005s

instance 1 [x] Done in 3.01s

instance 1 [x] Received ‘Hello…5’

instance 2 [x] Received ‘Hello…6’

instance 1 [x] Done in 2.006s

instance 1 [x] Received ‘Hello.7’

instance 1 [x] Done in 1.002s

instance 1 [x] Received ‘Hello…9’

instance 2 [x] Done in 3.01s

instance 2 [x] Received ‘Hello…8’

prefetch

從消費者這端的輸出可以看出來，instance 1 得到的任務編號始終是奇數（Hello.1，Hello…3，Hello…5，Hello.7），而 instance 2 得到的任務編號始終是偶數。了解springcloud架構可以加求求：三五三六二四七二五九

如果感覺這次的輸出只是巧合，可以多試幾次或通過 --tutorial.client.duration= 調整時長得到更多的輸出，而結果肯定都是一樣的。

這里設計的問題就是之前在基礎概念里講到的調度策略的問題了。要實現公平調度（Fair dispatch）就是設置 prefetch 的值，實現方式有兩種。

全局設置

在 application.yml 中設置 spring.rabbitmq.listener.simple.prefetch=1 即可，這會影響到本 Spring Boot 應用中所有使用默認 SimpleRabbitListenerContainerFactory 的消費者。

網上很多人說改配置 pring.rabbitmq.listener.prefetc，實測已經無效，應該是版本的問題。我所使用的版本（RabbitMQ：3.7.4，Spring Boot: 2.0.1.RELEASE），除了 spring.rabbitmq.listener.simple.prefetch，還有一個 spring.rabbitmq.listener.direct.prefetch 可以配置。

改了配置后再運行，可以看到 instance 1 可以獲取到”Hello…6”、”Hello…12” 了。

Ready … running for 60000ms

instance 1 [x] Received ‘Hello.1’

instance 2 [x] Received ‘Hello…2’

instance 1 [x] Done in 1.004s

instance 1 [x] Received ‘Hello…3’

instance 2 [x] Done in 2.008s

instance 2 [x] Received ‘Hello.4’

instance 2 [x] Done in 1.004s

instance 2 [x] Received ‘Hello…5’

instance 1 [x] Done in 3.012s

instance 1 [x] Received ‘Hello…6’

instance 2 [x] Done in 2.007s

instance 2 [x] Received ‘Hello.7’

instance 2 [x] Done in 1.004s

instance 2 [x] Received ‘Hello…8’

instance 1 [x] Done in 3.011s

instance 1 [x] Received ‘Hello…9’

instance 2 [x] Done in 2.007s

instance 2 [x] Received ‘Hello.10’

instance 2 [x] Done in 1.006s

instance 2 [x] Received ‘Hello…11’

instance 1 [x] Done in 3.01s

instance 1 [x] Received ‘Hello…12’

特定消費者

上邊是改了全局的消費者，如果只針對特定的消費者的話，又怎么處理呢？

我們可以通過自定義 RabbitListenerContainerFactory 來實現。

@Bean
public RabbitListenerContainerFactory<SimpleMessageListenerContainer> prefetchOneRabbitListenerContainerFactory(ConnectionFactory rabbitConnectionFactory) {
    SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
    factory.setConnectionFactory(rabbitConnectionFactory);
    factory.setPrefetchCount(1);
    return factory;
}

然后在特定的消費者上指定 containerFactory

@RebbitListener(queues = "hello", containerFactory = "prefetchTenRabbitListenerContainerFactory")
public void receive(String in) {
    System.out.println(" [x] Received '" + in + "'")
}