mq配置、springboot+rabbitmq实现生产消费

1）引用包

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

2）配置

spring:
  rabbitmq:
    host: 192.168.221.129
    port: 5673
    username: test
    password: test
    virtual-host: /
//  配置ack方式，我这配置成手动，方便容错重试
    listener:
      simple:
        acknowledge-mode: manual

3）配置主题交换机设置主题

 4）配置多个查询并绑定实现但生产多消费

 5）代码

@SneakyThrows
    @Override
    public void run(ApplicationArguments args) {
        rabbitProducerTopic("topic测试");
    }
 
 
    /**
     * 生产消息
     * @param msg 消息
     */
    private void rabbitProducerTopic(String msg){
        rabbitTemplate.convertAndSend("test.topic","test",msg);
    }
 
    /**
     * 消费消息1
     * @param msg
     * @param channel
     * @param message
     */
    @RabbitHandler
    @RabbitListener(queuesToDeclare = @Queue("topic.que"))
    private void rabbitTopic(String msg, Channel channel, Message message){
        System.out.println("test路由消费者topic："+msg);
        ackMethod(channel,message);
    }
 
    /**
     * 消费消息2
     * @param msg
     * @param channel
     * @param message
     */
    @RabbitHandler
    @RabbitListener(queuesToDeclare = @Queue("topic.queall"))
    private void rabbitTopicNoRout(String msg, Channel channel, Message message){
        System.out.println("无路由消费者topic："+msg);
        ackMethod(channel,message);
    }
 
    /**
     * 确认消息
     * @param channel
     * @param message
     */
    @SneakyThrows
    private void ackMethod(Channel channel, Message message){
        channel.basicAck(message.getMessageProperties().getDeliveryTag(),true);
    }
 
    /**
     * 异常消息重试
     * @param channel
     * @param message
     */
    @SneakyThrows
    private void nackMethod(Channel channel, Message message){
        channel.basicNack(message.getMessageProperties().getDeliveryTag(),false,// false 继续消费，true消息丢失
                true);
        // 异常阻塞十秒
        Thread.sleep(10000);
    }

5）其他

* AMQP四种不同的Exchange