处理Spring-Boot-RabbitMQ中的连接

问题描述 投票：1回答：2

嗨，我正在开发Spring-boot-RabbitMQ版本1.6。我在开发应用程序时几乎没有查询。阅读文档并浏览其他堆栈溢出问题，但我无法清楚地知道一些事情（可能是因为我的记忆力不好）。如果有人回答我的问题会很棒。

1）目前我有4个生产者和4个消费者。生产者可能会产生数百万个消息或事件，因此对生产者和消费者使用单一连接将阻止消费者消费消息。所以我想到的是创建单独的连接生产者和消费者，这样既不会阻止，也会提高性能。我对这种方法是否正确？

2）我正在使用CachingConnectionFactory来使用SimpleRabbitListenerContainerFactory.While调用这个工厂是否会为我们返回新连接？所以如果我们使用CachingConnectionFactory，我们真的需要为Producer和consumer编写一个单独的连接工厂。请在下面找到我的

1）配置类

@Configuration
@EnableRabbit
public class RabbitMqConfiguration{

@Autowired
private CachingConnectionFactory cachingConnectionFactory;

@Value("${concurrent.consumers}")
public int concurrent_consumers;

@Value("${max.concurrent.consumers}")
public int max_concurrent_consumers;

 @Bean
    public SimpleRabbitListenerContainerFactory rabbitListenerContainerFactory() {
        SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
        factory.setConnectionFactory(cachingConnectionFactory);
        factory.setConcurrentConsumers(concurrent_consumers);
        factory.setMaxConcurrentConsumers(max_concurrent_consumers);
        factory.setMessageConverter(jsonMessageConverter());
        return factory;
    }

@Bean
public MessageConverter jsonMessageConverter()
{
    final Jackson2JsonMessageConverter converter = new Jackson2JsonMessageConverter();
    return converter;
}

}

2）制片人类

@Configuration
public class TaskProducerConfiguration extends RabbitMqConfiguration {

@Value("${queue1}")
public String queue1;

@Value("${queue2}")
public String queue2;

@Value("${queue3}")
public String queue1;

@Value("${queue4}")
public String queue2;

@Value("${spring.rabbit.exchange}")
public String exchange;

@Autowired
private CachingConnectionFactory cachingConnectionFactory;

@Primary
@Bean
public RabbitTemplate getQueue1Template()
{
    RabbitTemplate template = new RabbitTemplate(cachingConnectionFactory);
    template.setRoutingKey(this.queue1);
    template.setMessageConverter(jsonMessageConverter());
    return template;
}

@Bean
public RabbitTemplate getQueue2Template()
{
    RabbitTemplate template = new RabbitTemplate(cachingConnectionFactory);
    template.setRoutingKey(this.queue2);
    template.setMessageConverter(jsonMessageConverter());
    return template;
}

@Bean
public RabbitTemplate getQueue3Template()
{
    RabbitTemplate template = new RabbitTemplate(cachingConnectionFactory);
    template.setRoutingKey(this.queue3);
    template.setMessageConverter(jsonMessageConverter());
    return template;
}

@Bean
public RabbitTemplate getQueue4Template()
{
    RabbitTemplate template = new RabbitTemplate(cachingConnectionFactory);
    template.setRoutingKey(this.queue4);
    template.setMessageConverter(jsonMessageConverter());
    return template;
}
@Bean(name="queue1Bean")
public Queue queue1()
{
    return new Queue(this.queue1);
}

@Bean(name="queue2Bean")
public Queue queue2()
{
    return new Queue(this.queue2);
}

@Bean(name="queue3Bean")
public Queue queue3()
{
    return new Queue(this.queue3);
}

@Bean(name="queue4Bean")
public Queue queue4()
{
    return new Queue(this.queue4);
}

@Bean
TopicExchange exchange() {
    return new TopicExchange(exchange);
}

@Bean
List<Binding> bindings(Queue queue1Bean,Queue queue2Bean,Queue queue3Bean,Queue queue4Bean, TopicExchange exchange) {
    List<Binding> bindingList = new ArrayList<Binding>();
    bindingList.add(BindingBuilder.bind(queue1Bean).to(exchange).with(this.queue1));
    bindingList.add(BindingBuilder.bind(queue2Bean).to(exchange).with(this.queue2));
    bindingList.add(BindingBuilder.bind(queue3Bean).to(exchange).with(this.queue3));
    bindingList.add(BindingBuilder.bind(queue4Bean).to(exchange).with(this.queue4));
    return bindingList;
}

}

3）接收器类（仅共享一个接收器类，其中3个接收器类是除了队列名和路由键之外的一个）。

@Component
public class Queue1Receiver {

@Autowired
private TaskProducer taskProducer;

@Value("${queue1}")
public String queue1;

@RabbitListener(id="queue1",containerFactory="rabbitListenerContainerFactory",queues = "#{queue1Bean}")
public void handleQueue1Message(TaskMessage taskMessage,@Header(AmqpHeaders.CONSUMER_QUEUE) String queue)
{
    System.out.println("Queue::"+queue);
    System.out.println("CustomerId: " + taskMessage.getCustomerID());
    if(taskMessage.isHasQueue2()){
        taskProducer.sendQueue2Message(taskMessage);
    }
    if(taskMessage.isHasQueue3()){
        taskProducer.sendQueue3Message(taskMessage);
    }
    if(taskMessage.isHasQueue4()){
        taskProducer.sendQueue4Message(taskMessage);
    }
}

@Bean
public Queue queue1Bean() {
    // This queue has the following properties:
    // name: my_durable,durable: true,exclusive: false,auto_delete: false
    return new Queue(queue1, true, false, false);
}

}

你的帮助应该是值得的。

注意：向下选民请在投票前注册您的评论，以便将来我可以避免错误。

根据Gary Russell的评论编辑：1）RabbitMqConfiguration

@Configuration
@EnableRabbit
public class RabbitMqConfiguration{

@Value("${concurrent.consumers}")
public int concurrent_consumers;

@Value("${max.concurrent.consumers}")
public int max_concurrent_consumers;

 @Bean
    public SimpleRabbitListenerContainerFactory rabbitListenerContainerFactory() {
        SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
        factory.setConnectionFactory(connectionFactory());
        factory.setConcurrentConsumers(concurrent_consumers);
        factory.setMaxConcurrentConsumers(max_concurrent_consumers);
        factory.setMessageConverter(jsonMessageConverter());
        return factory;
    }

@Bean
public CachingConnectionFactory connectionFactory()
{
    CachingConnectionFactory connectionFactory = new CachingConnectionFactory("localhost");
    connectionFactory.setUsername("guest");
    connectionFactory.setPassword("guest");
    connectionFactory.setCacheMode(CacheMode.CONNECTION);
    return connectionFactory;
}


@Bean
public MessageConverter jsonMessageConverter()
{
    final Jackson2JsonMessageConverter converter = new Jackson2JsonMessageConverter();
    return converter;
}


}
spring-amqp spring-rabbitmq spring-rabbit
2个回答
3
投票

对生产者和消费者使用单一连接将阻止消费者使用消息

是什么让你相信？单个连接通常没问题。如果您确实需要单独的连接，请将连接工厂cacheMode更改为CONNECTION。

0
投票

您可以在相同的情况下使用连接池，保持池大小合适可以解决问题。如上面的回答所示，生产者和消费者都使用相同的连接，因此池可能会帮助您。

最新问题
© www.soinside.com 2019 - 2024. All rights reserved.