消费端并发和限流设置

package com.java1234.consumer.service.impl;

import com.java1234.consumer.service.RabbitMqService;
import com.java1234.producer.config.RabbitMQConfig;
import com.rabbitmq.client.Channel;
import org.springframework.amqp.core.AmqpTemplate;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.support.AmqpHeaders;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.stereotype.Service;
import java.io.IOException;
import org.springframework.context.annotation.Bean;

@Service("rabbitmqService")
public class RabbitMqServiceImpl implements RabbitMqService {

    @Autowired
    private AmqpTemplate amqpTemplate;

    @Autowired
    CachingConnectionFactory cachingConnectionFactory;

    @Bean(name="limitContainerFactory")
    public SimpleRabbitListenerContainerFactory simpleRabbitListenerContainerFactory(){
        SimpleRabbitListenerContainerFactory factory=new SimpleRabbitListenerContainerFactory();
        factory.setConnectionFactory(cachingConnectionFactory);
        factory.setPrefetchCount(3);
        return factory;
    }

    @Override
    @RabbitListener(queues = {RabbitMQConfig.DIRECT_QUEUE},containerFactory = "limitContainerFactory")
    public void receiveMessage(String message, Channel channel, @Header(AmqpHeaders.DELIVERY_TAG) long deliveryTag) {
        try{
            System.out.println("接受到的mq消息："+message);
            //处理业务
            System.out.println("处理业务");
            System.out.println("deliveryTag="+deliveryTag);
            if(deliveryTag==5){
                channel.basicAck(deliveryTag,true);
            }
        }catch (Exception e){
            e.printStackTrace();
            try {
                channel.basicNack(deliveryTag,false,true);
//                channel.basicReject(deliveryTag,true);
//                Thread.sleep(1000);
            }catch (Exception e1){
                e1.printStackTrace();
            }

        }

    }


    @Override
//    @RabbitListener(queues = {RabbitMQConfig.DIRECT_QUEUE})
    public void receiveMessage2(String message) {
//        System.out.println("消费者1：接收到的mq消息："+message);
                System.out.println("队列1接收日志消息："+message);
    }


    @Override
    @RabbitListener(queues = {RabbitMQConfig.TOPIC_QUEUE2})
    public void receiveMessage3(String message) {
//        System.out.println("消费者2：接收到的mq消息："+message);
                System.out.println("队列2接收日志消息："+message);

    }

    @Override
    @RabbitListener(queues = {RabbitMQConfig.SUB_QUEUE1})
    public void receiveSubMessage1(String message) {
        System.out.println("订阅者1：接收到的mq消息："+message);
    }

    @Override
    @RabbitListener(queues = {RabbitMQConfig.SUB_QUEUE2})
    public void receiveSubMessage2(String message) {
        System.out.println("订阅者2：接收到的mq消息："+message);
    }

}