当前位置:   article > 正文

RabbitMQ之消费者批量消费_rabbitmq工具客户端 怎么一次性消费

rabbitmq工具客户端 怎么一次性消费

为什么要用消费端批量消费?
在一些业务场景下,我们希望使用 Consumer 批量消费消息,提高消费速度。可以通过对 SimpleRabbitListenerContainerFactory 进行配置实现批量消费能力

==========================>配置类
@Configuration
public class ConsumerConfiguration {
    @Resource
    ConnectionFactory connectionFactory;
    @Resource
    SimpleRabbitListenerContainerFactoryConfigurer configurer;

    /**
     * 配置一个批量消费的 SimpleRabbitListenerContainerFactory
     */
    @Bean(name = "consumer10BatchContainerFactory")
    public SimpleRabbitListenerContainerFactory consumer10BatchContainerFactory() {
        SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
        configurer.configure(factory, connectionFactory);
        // 这里是重点 配置消费者的监听器是批量消费消息的类型
        factory.setBatchListener(true);

        // 一批十个
        factory.setBatchSize(1000);
        // 等待时间 毫秒 , 这里其实是单个消息的等待时间 指的是单个消息的等待时间
        // 也就是说极端情况下,你会等待 BatchSize * ReceiveTimeout 的时间才会收到消息
        factory.setReceiveTimeout(10 * 1000L);
        factory.setConsumerBatchEnabled(true);

        return factory;
    }
}
====================》生产者
@Component
public class Producer10 {

    @Resource
    RabbitTemplate rabbitTemplate;

    public void sendSingle(String id, String routingKey) {
        Message10 message = new Message10();
        message.setId(id);
        rabbitTemplate.convertAndSend(Message10.EXCHANGE, routingKey, message);
    }
}
================================》消费者
@RabbitListener(queues = Message10.QUEUE, containerFactory = "consumer10BatchContainerFactory")
@Component
@Slf4j
public class Consumer10 {
    /**
     * 批量消费
     *
     * @param message 一批消息
     */
    @RabbitHandler
    public void onMessage(List<Message10> message) {
        log.info("[{}][Consumer10 批量][线程编号:{}][消息个数:{}][消息内容:{}]"
                , LocalDateTime.now()
                , Thread.currentThread().getId()
                , message.size()
                , message);
    }

    /**
     * 单个消费
     *
     * @param message 一个消息
     */
    @RabbitHandler
    public void onMessage(Message10 message) {
        log.info("[{}][Consumer10 单个][线程编号:{}][消息内容:{}]"
                , LocalDateTime.now()
                , Thread.currentThread().getId()
                , message);
    }
}
==================================》测试类
@Test
    void sendSingle() throws InterruptedException {
        // 假设 一秒一个,发送 1000 个,观察消费者的情况
        for (int i = 0; i < 15; i++) {
            TimeUnit.SECONDS.sleep(1);
            String id = UUID.randomUUID().toString();
            producer10.sendSingle(id, Message10.ROUTING_KEY);
            if (i == 9) {
                log.info("[{}][test producer10 sendSingle] 发送成功10个", LocalDateTime.now());
            }
        }
        log.info("[{}][test producer10 sendSingle] 发送成功", LocalDateTime.now());

        TimeUnit.SECONDS.sleep(20);
    }
}
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24
  • 25
  • 26
  • 27
  • 28
  • 29
  • 30
  • 31
  • 32
  • 33
  • 34
  • 35
  • 36
  • 37
  • 38
  • 39
  • 40
  • 41
  • 42
  • 43
  • 44
  • 45
  • 46
  • 47
  • 48
  • 49
  • 50
  • 51
  • 52
  • 53
  • 54
  • 55
  • 56
  • 57
  • 58
  • 59
  • 60
  • 61
  • 62
  • 63
  • 64
  • 65
  • 66
  • 67
  • 68
  • 69
  • 70
  • 71
  • 72
  • 73
  • 74
  • 75
  • 76
  • 77
  • 78
  • 79
  • 80
  • 81
  • 82
  • 83
  • 84
  • 85
  • 86
  • 87
  • 88
  • 89
  • 90

以上的是RabbitMQ之消费者批量消费实现的代码 若不了解rabbitmq的基本使用 建议先看看我前面对应的文章 文章链接:点我—>let’s go
若需完整代码 可识别二维码后 给您发代码。
在这里插入图片描述

声明:本文内容由网友自发贡献,不代表【wpsshop博客】立场,版权归原作者所有,本站不承担相应法律责任。如您发现有侵权的内容,请联系我们。转载请注明出处:https://www.wpsshop.cn/w/weixin_40725706/article/detail/680481
推荐阅读
相关标签
  

闽ICP备14008679号