Springboot2.x整合rabbitmq

一、spring-AMQP

Spring 框架的AMQP消息解决方案，提供模板化的发送和接收消息的抽象层，提供基于消息驱动的 POJO的消息监听等，提供不依赖于任何特定的AMQP代理实现或客户端库通用的抽象， 最终用户代码将很容易实现更易替换、添加和删除AMQP，因为它可以只针对抽象层来开发

总之就是提高我们的框架整合消息队列的效率,SpringBoot为更方便开发RabbitMQ推出了starter

当有新的AMQP协议的消息队列产品出现，根据这个协议进行消息队列的替换，虽然不能保证百分之百，但是提供了一个很好的接口

二、依赖

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

三、配置文件

spring:
  rabbitmq:
    #集群配置，不能同时与单机一起配置
    addresses: 127.0.0.1:5672,127.0.0.1:5673,127.0.0.1:5674
    #单机配置
    host: 127.0.0.1
    port: 5672
    virtual-host: /dev
    username: guest
    password: guest

四、配置类

package com.lcy.cloud.video.config;

import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.Exchange;
import org.springframework.amqp.core.ExchangeBuilder;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.core.QueueBuilder;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import com.lcy.cloud.video.constant.RabbitmqConstant;

@Configuration
public class RabbitMqConfig {
    /**
     * 交换机对象
     **/
    @Bean
    public Exchange orderExchange() {
        return ExchangeBuilder.topicExchange(RabbitmqConstant.ORDER_EXCHANGE_NAME).durable(true).build();
    }

    /**
     * 队列对象
     **/
    @Bean
    public Queue orderQueue() {
        return QueueBuilder.durable(RabbitmqConstant.ORDER_QUEUE_NAME).build();
    }

    /**
     * 交换机和队列绑定关系
     * @param queue    队列
     * @param exchange 交换机
     **/
    @Bean
    public Binding orderBinding(Queue queue, Exchange exchange) {
        return BindingBuilder.bind(queue).to(exchange).with("order.#").noargs();
    }
}

五、消息

六、生产者

package com.lcy.cloud.video;

import org.junit.jupiter.api.Test;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;

import com.lcy.cloud.video.constant.RabbitmqConstant;

@SpringBootTest(classes = VideoApplication.class)
public class OrderProduct {
    @Autowired
    private RabbitTemplate rabbitTemplate;

    @Test
    public void order() {
        rabbitTemplate.convertAndSend(RabbitmqConstant.ORDER_EXCHANGE_NAME, "order.new", "new order");
    }
}

七、消费者

package com.lcy.cloud.video.listener;

import java.io.IOException;

import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

import com.lcy.cloud.video.constant.RabbitmqConstant;

@Component
@RabbitListener(queues = RabbitmqConstant.ORDER_QUEUE_NAME)
public class OrderListener {
    @RabbitHandler
    public void releaseCouponRecord(String msg, Message message) throws IOException {
        long msgTag = message.getMessageProperties().getDeliveryTag();
        System.out.println("msgTag=" + msgTag);
        System.out.println("message=" + message);
        System.out.println("监听到消息：消息内容:" + new String(message.getBody()));
    }
}

八、消息可靠性投递

九、发送者到broker（交换机）

#旧版，确认消息发送成功，通过实现ConfirmCallBack接口，消息发送到交换器Exchange后触发回调
spring.rabbitmq.publisher-confirms=true
#新版，NONE值是禁用发布确认模式，是默认值，CORRELATED值是发布消息成功到交换器后会触发回调方法
spring.rabbitmq.publisher-confirm-type: correlated

@Test
void testConfirmCallback() {
    rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback() {
        /**
         *
         * @param correlationData 配置
         * @param ack 交换机是否收到消息，true是成功，false是失败
         * @param cause 失败的原因
         */
        @Override
        public void confirm(CorrelationData correlationData, boolean ack, String cause) {
            System.out.println("confirm=====>");
            System.out.println("confirm==== ack=" + ack);
            System.out.println("confirm==== cause=" + cause);
            if (ack) {
                System.out.println("消息发送成功");
            } else {
                System.out.println("消息发送失败");
            }
        }
    });
    rabbitTemplate.convertAndSend(RabbitmqConstant.ORDER_EXCHANGE_NAME, "order.new", "new order");
}

十、broker到queue队列

第一步：在配置文件开启全局的

#一定要开启
spring.rabbitmq.publisher-returns: true
#为true,则交换机处理消息到路由失败，则会返回给生产者
spring.rabbitmq.template.mandatory=true

也可以通过一下的方式不设置全局

rabbitTemplate.setMandatory(true);

@Test
void testReturnCallback() {
    rabbitTemplate.setReturnCallback(new RabbitTemplate.ReturnCallback() {
        @Override
        public void returnedMessage(Message message, int i, String s, String s1, String s2) {
            System.out.println("message" + new String(message.getBody()));
            System.out.println("i" + message);
            System.out.println("s" + s);
            System.out.println("s1" + s1);
            System.out.println("s2" + s2);
        }
    });
    rabbitTemplate.convertAndSend(RabbitmqConstant.ORDER_EXCHANGE_NAME, "order.new", "new order return message");
}