消費(fèi)者確認(rèn)機(jī)制

一、一次確認(rèn)一個(gè)消息

這里生產(chǎn)者一次性向rabbitmq發(fā)送一百條消息

	@GetMapping("/affair/affair")
    public String affair(){
        long begin=System.currentTimeMillis();
        for (int i = 0; i < 100; i++) {
            boolean b = Boolean.TRUE.equals(template.invoke(operations -> {
                template.convertAndSend("testQueue", "發(fā)布的消息");
                return template.waitForConfirms(1000);
            }));
            if(b) {
                System.out.println("發(fā)布成功");
            } else {
                System.out.println("發(fā)布失敗");
            }
        }
        long end=System.currentTimeMillis();
        return (end-begin)+"ms";
    }

然后消費(fèi)者一條一條的消費(fèi)，每次消費(fèi)時(shí)間模擬為0.5秒

 	@RabbitListener(queues = "testQueue", ackMode = "MANUAL")
    public void listen(Message msg, String str, Channel channel) throws IOException, InterruptedException {
        Thread.sleep(100);
        System.out.println("消息頭幀設(shè)置的內(nèi)容" + msg.getMessageProperties().getHeaders());
        System.out.println("消息體中明文發(fā)布的內(nèi)容" + str);
        channel.basicAck(msg.getMessageProperties().getDeliveryTag(), false);
    }

下面時(shí)消費(fèi)100條消息，消費(fèi)每條消息花時(shí)為0.5s的趨勢(shì)圖

二、一次確認(rèn)多個(gè)消息

yml文件中的配置其中concurrency和prefetch很重要

server:
  port: 8021
spring:
  application:
    name: rabbitmq-provider
  rabbitmq:
    host: 127.0.0.1
    port: 5672
    username: guest
    password: guest
    virtual-host: /
    publisher-returns: true #設(shè)置publisher-returns消息成功則會(huì)向發(fā)布者發(fā)送成功，失敗時(shí)不會(huì)將消息丟棄，而是返回給發(fā)布者，發(fā)布者根據(jù)需求處理
    publisher-confirm-type: correlated #開(kāi)啟發(fā)布確認(rèn)模式，具體有三個(gè)值，具體的可以去百度
    listener:
      simple:
        prefetch: 10 # 設(shè)置qos最大緩存數(shù)，預(yù)接受10個(gè)消息到緩存中
        concurrency: 10 # 處理線程數(shù)，可以理解為同一時(shí)間處理多少個(gè)消息
        max-concurrency: 20

發(fā)送代碼是相同的，以下是調(diào)整后的接受代碼

@RabbitListener(queues = "testQueue", ackMode = "MANUAL")
    public void listen(Message msg, String str) throws IOException, InterruptedException {
        Thread.sleep(500);
        System.out.println("消息頭幀設(shè)置的內(nèi)容" + msg.getMessageProperties().getHeaders());
        System.out.println("消息體中明文發(fā)布的內(nèi)容" + str);
        channel.basicAck(msg.getMessageProperties().getDeliveryTag(), false);
    }

下面時(shí)消費(fèi)100條消息，消費(fèi)每條消息花時(shí)為0.5s的趨勢(shì)圖，但每次同時(shí)有十條消息被消費(fèi)
文章來(lái)源地址http://www.zghlxwxcb.cn/news/detail-808593.html

到了這里，關(guān)于rabbitmq的qos和消費(fèi)者一次確認(rèn)多個(gè)消息的文章就介紹完了。如果您還想了解更多內(nèi)容，請(qǐng)?jiān)谟疑辖撬阉鱐OY模板網(wǎng)以前的文章或繼續(xù)瀏覽下面的相關(guān)文章，希望大家以后多多支持TOY模板網(wǎng)！