rabbitmq 在Springboot项目中的运用

Posted 2023-02-23 皓洲

rabbitmq 在Springboot项目中的运用

rabbitMQ的依赖

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

rabbitMQ的下载

• 第一步：安装Erlang OTP，由于RabbitMQ是用Erlang编写的，所以在安装RabbitMQ之前要先安装Erlang 下载地址：http://www.erlang.org/downloads 下载最新版本即可，例如OTP 20.0 Windows 64-bit Binary File, 下载完成后解压，双击otpwin6420.0.exe 一路next即可，安装成功后需要配置环境变量，可以新建一个变量，例如ERLANGHOME=D:erl9.0，最后将环境变量追加到Path中去Path中追加：%ERLANGHOMEbin%;
• 第二步：安装RabbitMQ，在官网上下载最新的RabbitMQhttps://www.rabbitmq.com/download.html 双击rabbitmq-server-3.6.10 一路next即可，安装成功后同样需要配置环境变量，将D:RabbitMQabbitmq_server-3.6.10sbin追加到Path中
• 第三步：安装RabbitMQ Management 插件，执行命令：rabbitmq-plugins enable rabbitmq_management 如果幸运的话一步成功，不知道最新版本会不会好安装，如果在安装中报错，要根据具体错误来解决

rabbitmq-service start 开始服务

config文件

package com.example.rabbitmqdemo.config;

import org.springframework.amqp.core.*;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

/**
 * @author ZhzGod
 * @date 2021/9/2 14:54
 * @introduction
 */
@Configuration
public class RabbitMqConfig 

    //1.申明注册fanout模式的交换机
    @Bean
    public DirectExchange directExchange() 
        return new DirectExchange("direct_order_exchange",true,false);
    
    //2.声明队列 sms.direct.queue
    @Bean
    public Queue smsQueue()
        return new Queue("sms.direct.queue",true);
    
    @Bean
    public Queue duanxinQueue()
        return new Queue("duanxin.direct.queue",true);
    
    @Bean
    public Queue emailQueue()
        return new Queue("email.direct.queue",true);
    
    //3.完成绑定关系（队列和交换机完成绑定关系）
    @Bean
    public Binding smsBingding()
        return BindingBuilder.bind((smsQueue())).to(directExchange()).with("sms");
    
    @Bean
    public Binding duanxinBingding()
        return BindingBuilder.bind((duanxinQueue())).to(directExchange()).with("duanxin");
    
    @Bean
    public Binding emailBingding()
        return BindingBuilder.bind((emailQueue())).to(directExchange()).with("email");
    

生产者

package com.example.rabbitmqdemo.service;

import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.util.UUID;

/**
 * @author ZhzGod
 * @date 2021/9/2 14:47
 * @introduction
 */
@Service
public class OrderService 

    @Autowired
    private RabbitTemplate rabbitTemplate;


    public void makeOrder(String userId,String productId,int num)

        //1：根据商品id检查库存是否充足
        //2：保存订单
        String orderId = UUID.randomUUID().toString();
        //1：通过MQ来完成消息的分表
        //参数1：交换机。 参数2：路由key/queue队列名称。 参数3：消息内容
        String exchangeName = "direct_order_exchange";
        String routingKey = "";
//        rabbitTemplate.convertAndSend(exchangeName,routingKey,orderId);
        rabbitTemplate.convertAndSend(exchangeName,"sms",orderId);
        rabbitTemplate.convertAndSend(exchangeName,"duanxin",orderId);
    

消费者

package com.example.rabbitmqdemo.service;

import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Service;

/**
 * @author ZhzGod
 * @date 2021/10/22 10:41
 * @introduction
 */
@RabbitListener(queues = "sms.direct.queue")
@Service
public class FanoutNotifyConsumer 
    @RabbitHandler
    public void receiveMessage(String message)
        System.out.println("sms fanout---接收到了订单信息是->"+message);
    

测试

package com.example.rabbitmqdemo;

import com.example.rabbitmqdemo.service.OrderService;
import org.junit.jupiter.api.Test;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;

@SpringBootTest
class RabbitMqDemoApplicationTests 

    @Autowired
    private OrderService orderService;
    @Test
    void contextLoads() 
        orderService.makeOrder("user1","good1",10);
    

结果

利用mq发送延迟任务

路由key枚举类中添加信息

/**
     * appsflyer延迟消息队列
     */
    CIF_ADMIN_APPSFLYER_RETRY("cifAdminAppsflyer"),
    CIF_ADMIN_APPSFLYER_RETRY_LISTENER("cif.admin.appsflyer.listener");



    CifMQRoutingKeyEnum(String routingKey) 
        this.routingKey = routingKey;
    

    public String getRoutingKey() 
        return routingKey;
    

    public void setRoutingKey(String routingKey) 
        this.routingKey = routingKey;
    

    private String routingKey;

队列名枚举类添加信息

/**
     * appsflyer延迟消息队列
     */
    CIF_ADMIN_APPSFLYER_RETRY("cifAdminAppsflyer"),
    CIF_ADMIN_APPSFLYER_RETRY_LISTENER("cif.admin.appsflyer.listener");

CifMQQueueNameEnum(String queueName) 
        this.queueName = queueName;
    

    public String getQueueName() 
        return queueName;
    

    private String queueName;

在MQconfig中添加配置信息

public final static String EXCHANGE_NAME = "flex";

//延时队列配置
    @Bean
    Queue queue15() 
        Map<String, Object> argsMap = Maps.newHashMap();
        argsMap.put("x-dead-letter-exchange", RabbitMQConfig.EXCHANGE_NAME); //真正的交换机
        argsMap.put("x-dead-letter-routing-key", CifMQRoutingKeyEnum.CIF_ADMIN_APPSFLYER_RETRY_LISTENER.getRoutingKey()); //真正的路由键
        return new Queue(CifMQQueueNameEnum.CIF_ADMIN_APPSFLYER_RETRY.getQueueName(), true, false, false, argsMap);
    

    //appsflyer的延迟消息队列
    @Bean
    Queue queue16() 
        return new Queue(CifMQQueueNameEnum.CIF_ADMIN_APPSFLYER_RETRY_LISTENER.getQueueName());
    

发送延迟的mq消息

@Autowired
RabbitTemplate rabbitTemplate;

rabbitTemplate.convertAndSend(RabbitMQConfig.EXCHANGE_NAME, CifMQRoutingKeyEnum.CIF_ADMIN_APPSFLYER_RETRY.getRoutingKey(), JsonUtil.allToJson(appsFlyerRetryDTO), msg -> 
                    // 延迟发送，毫秒
                    msg.getMessageProperties().setExpiration("60000");
                    return msg;
                );

监听延迟mq消息

@RabbitListener(queues = RabbitMQConfig.CIF_ADMIN_APPSFLYER_RETRY_LISTENER)

原理：建立了两个消息队列，发送延迟消息时，将消息发送给消息队列1，消息会隔一段时间传送到消息队列2，这时候我们只需要监听消息队列2就可以了。