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

成都創(chuàng)新互聯(lián)是一家專業(yè)提供東興企業(yè)網(wǎng)站建設,專注與網(wǎng)站制作、網(wǎng)站設計、H5建站、小程序制作等業(yè)務。10年已為東興眾多企業(yè)、政府機構等服務。創(chuàng)新互聯(lián)專業(yè)網(wǎng)站設計公司優(yōu)惠進行中。
如果您尚未設置項目,請參閱第一個教程中的設置。我們將遵循與第一個教程相同的模式:創(chuàng)建一個包(tut2)并創(chuàng)建 Tut2Config、Tut2Receiver 和 Tut2Sender。
代碼整合
首先創(chuàng)建一個新的包(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();
    }
}生產(chǎn)者
我們簡單修改一下生產(chǎn)者的代碼,以添加點號(.)的方式來人為的增加該任務的時長,字符串中的每個點號(.)都會增加 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 秒鐘的操作。并且我們?yōu)橄M者增加了一個實例編號,以知道是哪個實例消費了消息和處理的時長。
@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 得到的任務編號始終是奇數(shù)(Hello.1,Hello…3,Hello…5,Hello.7),而 instance 2 得到的任務編號始終是偶數(shù)。了解springcloud架構可以加求求:三五三六二四七二五九
如果感覺這次的輸出只是巧合,可以多試幾次或通過 --tutorial.client.duration= 調整時長得到更多的輸出,而結果肯定都是一樣的。
這里設計的問題就是之前在基礎概念里講到的調度策略的問題了。要實現(xiàn)公平調度(Fair dispatch)就是設置 prefetch 的值,實現(xiàn)方式有兩種。
全局設置
在 application.yml 中設置 spring.rabbitmq.listener.simple.prefetch=1 即可,這會影響到本 Spring Boot 應用中所有使用默認 SimpleRabbitListenerContainerFactory 的消費者。
網(wǎng)上很多人說改配置 pring.rabbitmq.listener.prefetc,實測已經(jīng)無效,應該是版本的問題。我所使用的版本(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 來實現(xiàn)。
@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 + "'")
}            
                本文題目:JavaSpringboot整合RabbitMQ(二):工作隊列(Workqueues)-B2B2C小程序電子商務
                
                文章鏈接:http://www.chinadenli.net/article30/gpidpo.html
            
成都網(wǎng)站建設公司_創(chuàng)新互聯(lián),為您提供網(wǎng)站收錄、商城網(wǎng)站、企業(yè)建站、外貿建站、品牌網(wǎng)站制作、定制開發(fā)
聲明:本網(wǎng)站發(fā)布的內容(圖片、視頻和文字)以用戶投稿、用戶轉載內容為主,如果涉及侵權請盡快告知,我們將會在第一時間刪除。文章觀點不代表本網(wǎng)站立場,如需處理請聯(lián)系客服。電話:028-86922220;郵箱:631063699@qq.com。內容未經(jīng)允許不得轉載,或轉載時需注明來源: 創(chuàng)新互聯(lián)
