<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-amqp</artifactId>
</dependency>

spring.rabbitmq.host=localhost
spring.rabbitmq.port=5672
spring.rabbitmq.username=guest
spring.rabbitmq.password=guest
spring.rabbitmq.virtual-host=/

# 隊(duì)列名稱(chēng)
spring.rabbitmq.listener.simple.queue-name=myQueue

# 最大并發(fā)消費(fèi)者數(shù)量
spring.rabbitmq.listener.simple.concurrency=5

# 最小數(shù)量
spring.rabbitmq.listener.simple.min-concurrency=1

# 最大數(shù)量
spring.rabbitmq.listener.simple.max-concurrency=10

# 批量處理消息的大小
spring.rabbitmq.listener.simple.batch-size=50

spring:
  rabbitmq:
    host: localhost
    listener:
      simple:
        batch-size: 50
        concurrency: 5
        max-concurrency: 10
        min-concurrency: 1
        queue-name: myQueue
    password: guest
    port: 5672
    username: guest
    virtual-host: /

@Component
public class MyListener {
    
    @RabbitListener(queues = "myQueue", containerFactory = "myFactory")
    public void handleMessage(List<MyMessage> messages, Channel channel, @Header(AmqpHeaders.DELIVERY_TAG) long deliveryTag)
            throws IOException {
        try {
            // 處理消息
            System.out.println("Received " + messages.size() + " messages");
            for (Message message : messages) {
           		// 處理消息
            	System.out.println("Received message: " + new String(message.getBody()));
        	}
        	channel.basicAck(messages.get(messages.size() - 1).getMessageProperties().getDeliveryTag(), true);
        } finally {
            // 手動(dòng)確認(rèn)消息
            channel.basicAck(deliveryTag, true);
        }
    }
}

@Configuration
public class RabbitMQConfig {

//    @Bean
//    public SimpleRabbitListenerContainerFactory myFactory(ConnectionFactory connectionFactory) {
//        SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
//        factory.setConnectionFactory(connectionFactory);
//        factory.setConcurrentConsumers(1);
//        factory.setMaxConcurrentConsumers(10);
//        factory.setBatchListener(true);
//        factory.setBatchSize(50);
//        return factory;
//    }

	@Bean
    public SimpleRabbitListenerContainerFactory myFactory(
            ConnectionFactory connectionFactory,
            PlatformTransactionManager transactionManager,
            MessageConverter messageConverter) {
        
        SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
        
        // 并發(fā)消費(fèi)者數(shù)，默認(rèn)為 1
        factory.setConcurrentConsumers(5);
        
        // 最大并發(fā)消費(fèi)者數(shù)，默認(rèn)為 1
        factory.setMaxConcurrentConsumers(10);
        
        // 拒絕未確認(rèn)的消息并重新將它們放回隊(duì)列，默認(rèn)為 true
        factory.setDefaultRequeueRejected(false);
        
        // 容器啟動(dòng)時(shí)是否自動(dòng)啟動(dòng)，默認(rèn)為 true
        factory.setAutoStartup(true);
        
        // 消息確認(rèn)模式，默認(rèn)為 AUTO
        factory.setAcknowledgeMode(AcknowledgeMode.MANUAL);
        
        // 每個(gè)消費(fèi)者在一次請(qǐng)求中預(yù)獲取的消息數(shù)，默認(rèn)為 1
        factory.setPrefetchCount(5);
        
        // 從隊(duì)列中接收消息的超時(shí)時(shí)間，默認(rèn)為 0，表示沒(méi)有超時(shí)限制
        factory.setReceiveTimeout(1000);
        
        // 與容器一起使用的事務(wù)管理器。默認(rèn)情況下，容器不會(huì)使用事務(wù)
        factory.setTransactionManager(transactionManager);
        
        // 消息轉(zhuǎn)換器，用于將接收到的消息轉(zhuǎn)換為 Java 對(duì)象或?qū)?Java 對(duì)象轉(zhuǎn)換為消息
        factory.setMessageConverter(messageConverter);
        
        // 用于異步消息處理的線程池。默認(rèn)情況下，容器使用一個(gè)簡(jiǎn)單的 SimpleAsyncTaskExecutor
        factory.setTaskExecutor(new SimpleAsyncTaskExecutor());
        
        // 在關(guān)閉容器時(shí)等待活動(dòng)線程終止的時(shí)間，默認(rèn)為 5000 毫秒
        factory.setShutdownTimeout(10000);
        
        // 重試失敗的消息之前等待的時(shí)間，默認(rèn)為 5000 毫秒
        factory.setRecoveryInterval(5000);
        
        // 如果消息處理器嘗試監(jiān)聽(tīng)不存在的隊(duì)列，是否拋出異常。默認(rèn)為 true
        factory.setMissingQueuesFatal(false);
        
        // 監(jiān)聽(tīng)器容器連接工廠
        factory.setConnectionFactory(connectionFactory);

        return factory;
    }
}

@Component
public class MySender {

    @Autowired
    private RabbitTemplate rabbitTemplate;

    public void sendMessage(String message) {
        for (int i = 0; i < 100; i++) {
            rabbitTemplate.convertAndSend("myQueue", "message：" + i);
        }
    }
}