spring-boot接入RabbitMQ的详细步骤

RabbitMQ是目前使用最为广泛的开源消息中间件之一。它实现了AMQP协议，它支持集群化部署，可以动态伸缩。使用RabbitMQ可以将复杂的系统解耦，可以做瞬时高峰的削峰处理。那么在spring-boot中我们该如何集成RabbitMQ呢？

    org.springframework.boot
    spring-boot-starter-amqp

spring:
  rabbitmq:
    host: xxxx
    port: xxxx
    username: xxxx
    password: xxxx
    listener:
      type: simple
      simple:
        acknowledge-mode: manual #采用手动应答
        prefetch: 1 #限制每次发送一条数据。
queues:
	testQueueName: testQueueName
	testQueueExchangeName: testQueueExchangeName
	testExchangeEnabled: true

@Configuration
public class RabbitMQConfig {
    @Value("${queues.testQueueName}")
    private String testQueueName;

    @Value("${queues.testQueueExchangeName}")
    private String testQueueExchangeName;

    @Bean(name = "TestInfo")
    public Queue TestMessage() {
        return new Queue(testQueueName);
    }

    
    @Bean
    FanoutExchange testFanoutExchange() {
        return new FanoutExchange(testQueueExchangeName);
    }
    
    @Bean
    Binding bindingExchangeTestInfo(@Qualifier("TestInfo") Queue TestMessage, @Qualifier("testFanoutExchange") FanoutExchange fanoutExchange) {
        return BindingBuilder.bind(TestMessage).to(fanoutExchange);
    }
}

@Slf4j
@Service
public class RabbitMQSendServiceImpl implements RabbitMQSendService {
	
	@Autowired
    RabbitTemplate rabbitmqTemplate;
    
    
    public void sendToMQ(String msg) {
        // Send MQ
        try {
            rabbitmqTemplate.convertAndSend("testQueueExchangeName", "", msg);
        } catch (Exception ex) {
            ex.printStackTrace();
            System.out.println(ex);
        }
    }
}

@Component
public class RabbitMQMsgListener {
	
	@Bean("batchTestListenerContainerFactory")
    public SimpleRabbitListenerContainerFactory batchBumImpListenerContainerFactory(ConnectionFactory connectionFactory) {
        SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
        factory.setConnectionFactory(connectionFactory);
        //设置批量
        factory.setBatchListener(true);
        factory.setConsumerBatchEnabled(true);//设置BatchMessageListener生效
        factory.setBatchSize(100);//设置监听器一次批量处理的消息数量
        return factory;
    }

	// 单条消息接收处理
	@RabbitListener(queues = "${queues.testQueueName}", autoStartup = "${queues.testExchangeEnabled}")
    public void process(Message message, Channel channel) throws IOException {
        String msg = new String(message.getBody(), "UTF-8");
        // TODO 单条消息处理
        // 处理完手动ACK
        channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
    }
    
	// 批量接收处理
    @RabbitListener(queues = "${queues.testQueueName}", autoStartup = "${queues.testExchangeEnabled}", containerFactory = "batchTestListenerContainerFactory")
    public void onMessageBatch(List messages, Channel channel) throws IOException {
        try {
            for (Message message : messages) {
                String msg = new String(message.getBody(), "UTF-8");
                // TODO 单条消息处理
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}