Mall整合RabbitMQ

Mall整合RabbitMQ Mall整合RabbitMQ

http://localhost:15672 配置mall用户 配置/mall的虚拟机 配置mall用户绑定/mall虚拟机！

1、依赖 1.1、pom依赖

    4.0.0


    
        org.springframework.boot
        spring-boot-starter-parent
        2.1.3.RELEASE
         
    


    com.wnx.mall.tiny
    mall-tiny-08
    0.0.1-SNAPSHOT
    mall-tiny-08

    Demo project for Spring Boot
    
        1.8
    

    
        
        
            org.springframework.boot
            spring-boot-starter-web
        
        
            org.springframework.boot
            spring-boot-starter-actuator
        
        
            org.springframework.boot
            spring-boot-starter-aop
        
        
            org.springframework.boot
            spring-boot-starter-test
            test
        
        
        
            com.github.pagehelper
            pagehelper-spring-boot-starter
            1.2.10
        
        
        
            com.alibaba
            druid-spring-boot-starter
            1.1.10
        
        
        
            org.mybatis.generator
            mybatis-generator-core
            1.3.3
        
        
        
            mysql
            mysql-connector-java
            8.0.15
        
        
        
            io.springfox
            springfox-swagger2
            2.7.0
        
        
            io.springfox
            springfox-swagger-ui
            2.7.0
        
        
        
            org.springframework.boot
            spring-boot-starter-data-redis
        
        
        
            org.springframework.boot
            spring-boot-starter-security
        
        
        
            cn.hutool
            hutool-all
            4.5.7
        
        
        
            io.jsonwebtoken
            jjwt
            0.9.0
        
        
        
            org.springframework.boot
            spring-boot-starter-data-elasticsearch
        
        
        
            org.springframework.boot
            spring-boot-starter-data-mongodb
        
        
        
            org.springframework.boot
            spring-boot-starter-amqp
        
        
        
            org.projectlombok
            lombok
            true
        
    

    
        
            
                org.springframework.boot
                spring-boot-maven-plugin
            
        
    

1.2、application.yaml配置

server:
  port: 8080

spring:
  datasource:
    url: jdbc:mysql://localhost:3306/mall?useUnicode=true&characterEncoding=utf-8&serverTimezone=Asia/Shanghai
    username: root
    password: root
  redis:
    host: localhost # Redis服务器地址
    database: 0 # Redis数据库索引（默认为0）
    port: 6379 # Redis服务器连接端口
    password: # Redis服务器连接密码（默认为空）
    jedis:
      pool:
        max-active: 8 # 连接池最大连接数（使用负值表示没有限制）
        max-wait: -1ms # 连接池最大阻塞等待时间（使用负值表示没有限制）
        max-idle: 8 # 连接池中的最大空闲连接
        min-idle: 0 # 连接池中的最小空闲连接
    timeout: 3000ms # 连接超时时间（毫秒）
  data:
    elasticsearch:
      repositories:
        enabled: true
      cluster-nodes: 127.0.0.1:9300 # es的连接地址及端口号
      cluster-name: elasticsearch # es集群的名称
    mongodb:
      host: localhost # mongodb的连接地址
      port: 27017 # mongodb的连接端口号
      database: mall-port # mongodb的连接的数据库
  rabbitmq:
    host: localhost # rabbitmq的连接地址
    port: 5672 # rabbitmq的连接端口号
    virtual-host: /mall # rabbitmq的虚拟host
    username: mall # rabbitmq的用户名
    password: mall # rabbitmq的密码
    publisher-/confirm/is: true #如果对异步消息需要回调必须设置为true

mybatis:
  mapper-locations:
    - classpath:mapper/*.xml
    - classpath*:commapper
@Configuration
public class RabbitMqConfig {

    
    @Bean
    DirectExchange orderDirect() {
        return (DirectExchange) ExchangeBuilder
                .directExchange(QueueEnum.QUEUE_ORDER_CANCEL.getExchange())
                .durable(true)
                .build();
    }

    
    @Bean
    DirectExchange orderTtlDirect() {
        return (DirectExchange) ExchangeBuilder
                .directExchange(QueueEnum.QUEUE_TTL_ORDER_CANCEL.getExchange())
                .durable(true)
                .build();
    }

    
    @Bean
    public Queue orderQueue() {
        return new Queue(QueueEnum.QUEUE_ORDER_CANCEL.getName());
    }

    
    @Bean
    public Queue orderTtlQueue() {
        return QueueBuilder
                .durable(QueueEnum.QUEUE_TTL_ORDER_CANCEL.getName())
                .withArgument("x-dead-letter-exchange", QueueEnum.QUEUE_ORDER_CANCEL.getExchange())//到期后转发的交换机
                .withArgument("x-dead-letter-routing-key", QueueEnum.QUEUE_ORDER_CANCEL.getRouteKey())//到期后转发的路由键
                .build();
    }

    
    @Bean
    Binding orderBinding(DirectExchange orderDirect,Queue orderQueue){
        return BindingBuilder
                .bind(orderQueue)
                .to(orderDirect)
                .with(QueueEnum.QUEUE_ORDER_CANCEL.getRouteKey());
    }

    
    @Bean
    Binding orderTtlBinding(DirectExchange orderTtlDirect,Queue orderTtlQueue){
        return BindingBuilder
                .bind(orderTtlQueue)
                .to(orderTtlDirect)
                .with(QueueEnum.QUEUE_TTL_ORDER_CANCEL.getRouteKey());
    }

}

2.2、组件

• 取消订单消息的发出者

package com.wnx.mall.tiny.component;


import com.wnx.mall.tiny.dto.QueueEnum;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.AmqpException;
import org.springframework.amqp.core.AmqpTemplate;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessagePostProcessor;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;


@Component
public class CancelOrderSender {
    private static Logger LOGGER =LoggerFactory.getLogger(CancelOrderSender.class);
    @Autowired
    private AmqpTemplate amqpTemplate;

    public void sendMessage(Long orderId,final long delayTimes){
        //给延迟队列发送消息
        amqpTemplate.convertAndSend(QueueEnum.QUEUE_TTL_ORDER_CANCEL.getExchange(), QueueEnum.QUEUE_TTL_ORDER_CANCEL.getRouteKey(), orderId, new MessagePostProcessor() {
            @Override
            public Message postProcessMessage(Message message) throws AmqpException {
                //给消息设置延迟毫秒值
                message.getMessageProperties().setExpiration(String.valueOf(delayTimes));
                return message;
            }
        });
        LOGGER.info("send delay message orderId:{}",orderId);
    }
}

• 取消订单消息的处理者

package com.wnx.mall.tiny.component;

import com.wnx.mall.tiny.service.OmsPortalOrderService;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;


@Component
@RabbitListener(queues = "mall.order.cancel")
public class CancelOrderReceiver {
    private static Logger LOGGER =LoggerFactory.getLogger(CancelOrderReceiver.class);
    @Autowired
    private OmsPortalOrderService portalOrderService;
    @RabbitHandler
    public void handle(Long orderId){
        LOGGER.info("receive delay message orderId:{}",orderId);
        portalOrderService.cancelOrder(orderId);
    }
}

3、业务 3.1、Service

package com.wnx.mall.tiny.service;

import com.wnx.mall.tiny.common.api.CommonResult;
import com.wnx.mall.tiny.dto.OrderParam;
import org.springframework.transaction.annotation.Transactional;


public interface OmsPortalOrderService {
    
    @Transactional
    CommonResult generateOrder(OrderParam orderParam);

    
    @Transactional
    void cancelOrder(Long orderId);
}

3.2、ServiceImpl

package com.wnx.mall.tiny.service.impl;

import com.wnx.mall.tiny.common.api.CommonResult;
import com.wnx.mall.tiny.component.CancelOrderReceiver;
import com.wnx.mall.tiny.component.CancelOrderSender;
import com.wnx.mall.tiny.dto.OrderParam;
import com.wnx.mall.tiny.service.OmsPortalOrderService;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;

@Service
public class OmsPortalOrderServiceImpl implements OmsPortalOrderService {
    private static Logger LOGGER = LoggerFactory.getLogger(OmsPortalOrderServiceImpl.class);
    @Resource
    private CancelOrderSender orderSender;
    @Override
    public CommonResult generateOrder(OrderParam orderParam) {
        //todo 执行一系类下单 *** 作，具体参考mall项目
        LOGGER.info("process generateOrder");
        //下单完成后开启一个延迟消息，用于当用户没有付款时取消订单（orderId应该在下单后生成！）
        sendDelayMessageCancelOrder(11L);
        return CommonResult.success(null,"下单成功！");
    }

    @Override
    public void cancelOrder(Long orderId) {
        //todo:取消一系类取消订单 *** 作，具体参考mll项目
        LOGGER.info("process cancelOrder orderId:{}",orderId);

    }

    private void sendDelayMessageCancelOrder(Long orderId){
        //获取订单超时时间，假设为60分钟（测试用的30秒）
        long delayTimes = 30 *1000;
        //发送延迟消息
        orderSender.sendMessage(orderId,delayTimes);


    }
}

3.3、Controller

package com.wnx.mall.tiny.controller;

import com.wnx.mall.tiny.dto.OrderParam;
import com.wnx.mall.tiny.service.OmsPortalOrderService;
import io.swagger.annotations.Api;
import io.swagger.annotations.ApiOperation;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Controller;
import org.springframework.web.bind.annotation.RequestBody;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestMethod;
import org.springframework.web.bind.annotation.ResponseBody;


@Controller
@Api(tags = "OmsPortalOrderController", description = "订单管理")
@RequestMapping("/order")
public class OmsPortalOrderController {
    @Autowired
    private OmsPortalOrderService portalOrderService;

    @ApiOperation("根据购物车信息生成订单")
    @RequestMapping(value = "/generateOrder", method = RequestMethod.POST)
    @ResponseBody
    public Object generateOrder(@RequestBody OrderParam orderParam) {
        return portalOrderService.generateOrder(orderParam);
    }
}

4、测试

http://localhost:8080/swagger-ui.html 30秒后，取消订单！=》下单完成后开启一个延迟消息，用于当用户没有付款时取消订单（orderId应该在下单后生成！）