Spring rabbitmq 任务队列并发答案

【问题标题】：Spring rabbitmq task queue concurrencySpring rabbitmq 任务队列并发
【发布时间】：2019-12-23 14:43:06
【问题描述】：

我有一个使用代理发出 http 请求的任务队列。代理限制为 10 个并发线程/连接。我无权访问代理的日志。

我正在使用以下代码，这是在名为 ntContainer#1-1 和 container1 的两个线程上发出请求。这会导致许多请求由于使用过多的代理连接而出错。

监听器是只使用 1 个默认线程和额外的容器线程，还是在 spring/rabbitmq 的幕后发生了更多事情？另外我该如何进一步调试？

@Configuration
public class RabbitMQConfig {

    public final static String EXCHANGE_NAME = "my-tx";
    public final static String MY_PRODUCT_ROUTING_KEY = "my-product-routing-key";
    public final static String MY_PRODUCT_QUEUE = "my-product";

    @Bean
    public TopicExchange topicExchange() {
        return new TopicExchange(EXCHANGE_NAME);
    }

    @Bean
    public Queue myProductQueue() {
        return new Queue(MY_PRODUCT_QUEUE);
    }

    @Bean
    Binding myProductBinding() {
        return BindingBuilder.bind(myProductQueue()).to(topicExchange()).with(MY_PRODUCT_ROUTING_KEY);
    }

    @Bean
    SimpleMessageListenerContainer container(ConnectionFactory connectionFactory, MessageListenerAdapter messageListenerAdapter) {
        SimpleMessageListenerContainer container = new SimpleMessageListenerContainer();
        container.setConnectionFactory(connectionFactory);
        container.setQueueNames(MY_PRODUCT_QUEUE);
        container.setMessageListener(messageListenerAdapter);
        container.setPrefetchCount(1);
        container.setConcurrentConsumers(1);
        return container;
    }

    @Bean
    MessageListenerAdapter messageListenerAdapter(MyListener myListener) {
        return new MessageListenerAdapter(myListener, "process");
    }
}

// 监听器

@RabbitListener(queues = RabbitMQConfig.MY_PRODUCT_QUEUE)
public void process(final Message message) {
    // something like this
    Jsoup.connect(message.getUrl()).proxy().execute()
}

【问题讨论】：

我的猜测是监听是 Rabbit MQ 的默认设置，它是一个大小为 5 的固定线程池。我猜 Spring 不会改变线程模型。为了调试，我会从我的监听器打印 Thread.currentThread()。
有了这些属性，只有一个线程会调用监听器。这是一个 Spring 线程，而不是客户端线程之一。引导默认日志配置会截断线程名称，因此打印完整的线程名称会有所帮助。

标签： spring-boot rabbitmq spring-rabbit

【解决方案1】：

哎呀；我正在看手机上的问题；我跳过了容器豆；我认为容器 bean 是容器工厂而不是容器。

您有 2 个侦听器容器 -

@RabbitListener(queues = RabbitMQConfig.MY_PRODUCT_QUEUE)
public void process(final Message message) {
    // something like this
    Jsoup.connect(message.getUrl()).proxy().execute()
}

框架会自动为那个监听器创建一个容器（它会检测注解）并且你已经明确声明了另一个容器@Bean。

代理限制为 10 个并发线程/连接。

即使有 2 个容器，您也只能获得 2 个线程，而不是 10 个。

【讨论】：