新增定时取消订单

carbon-h5/carbon-h5-service/pom.xml

@@ -140,6 +140,12 @@
             <artifactId>redisson</artifactId>
             <version>3.5.0</version>
         </dependency>
+
+        <!--消息中间件-->
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter-amqp</artifactId>
+        </dependency>
     </dependencies>
 
 

carbon-h5/carbon-h5-service/src/main/java/com/hcloud/microserver/h5/config/mq/MQConstant.java

@@ -0,0 +1,23 @@
+package com.hcloud.microserver.h5.config.mq;
+
+/**
+ * Rabbit消息队列相关常量
+ */
+public final class MQConstant {
+
+    private MQConstant(){}
+
+    //exchange name
+    public static final String DEFAULT_EXCHANGE = "ZyChange";
+
+    //TTL QUEUE
+    public static final String DEFAULT_DEAD_LETTER_QUEUE_NAME = "zy.dead.letter.queue";
+
+    //DLX repeat QUEUE 死信转发队列
+    public static final String DEFAULT_REPEAT_TRADE_QUEUE_NAME = "zy.repeat.trade.queue";
+
+    //Hello 测试消息队列名称
+    public static final String HELLO_QUEUE_NAME = "HELLO";
+
+}
+

carbon-h5/carbon-h5-service/src/main/java/com/hcloud/microserver/h5/config/mq/MessageService.java

@@ -0,0 +1,54 @@
+package com.hcloud.microserver.h5.config.mq;
+
+import com.alibaba.fastjson.JSON;
+import com.hcloud.microserver.h5.config.mq.service.DLXMessage;
+import com.hcloud.microserver.h5.config.mq.service.impl.IMessageService;
+import org.springframework.amqp.AmqpException;
+import org.springframework.amqp.core.Message;
+import org.springframework.amqp.core.MessagePostProcessor;
+import org.springframework.amqp.rabbit.core.RabbitTemplate;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Service;
+
+/**
+ * 消息队列服务接口实现
+ */
+@Service("messageService")
+public class MessageService implements IMessageService {
+
+    @Autowired
+    private RabbitTemplate rabbitTemplate;
+
+    /**
+     * 发送消息到队列
+     *
+     * @param queueName 队列名称
+     * @param message   消息内容
+     */
+    @Override
+    public void send(String queueName, String message) {
+        rabbitTemplate.convertAndSend(MQConstant.DEFAULT_EXCHANGE, queueName, message);
+    }
+
+    /**
+     * 延迟发送消息到队列
+     *
+     * @param queueName 队列名称
+     * @param message   消息内容
+     * @param times     延迟时间 单位毫秒
+     */
+    @Override
+    public void send(String queueName, String message, long times) {
+        //消息发送到死信队列上，当消息超时时，会发生到转发队列上，转发队列根据下面封装的queueName，把消息转发的指定队列上
+        //发送前，把消息进行封装，转发时应转发到指定 queueName 队列上
+        DLXMessage dlxMessage = new DLXMessage(MQConstant.DEFAULT_EXCHANGE, queueName, message, times);
+        MessagePostProcessor processor = new MessagePostProcessor() {
+            @Override
+            public Message postProcessMessage(Message message) throws AmqpException {
+                message.getMessageProperties().setExpiration(times + "");
+                return message;
+            }
+        };
+        rabbitTemplate.convertAndSend(MQConstant.DEFAULT_EXCHANGE, MQConstant.DEFAULT_DEAD_LETTER_QUEUE_NAME, JSON.toJSONString(dlxMessage), processor);
+    }
+}

carbon-h5/carbon-h5-service/src/main/java/com/hcloud/microserver/h5/config/mq/QueueConfiguration.java

@@ -0,0 +1,66 @@
+package com.hcloud.microserver.h5.config.mq;
+
+import org.springframework.amqp.core.Binding;
+import org.springframework.amqp.core.BindingBuilder;
+import org.springframework.amqp.core.DirectExchange;
+import org.springframework.amqp.core.Queue;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+
+import java.util.HashMap;
+import java.util.Map;
+
+/**
+ * 队列配置，所有配置@Bean的队列名称，由系统启动时创建队列，并绑定到Exchane上
+ */
+@Configuration
+public class QueueConfiguration {
+
+    //信道配置
+    @Bean
+    public DirectExchange defaultExchange() {
+        return new DirectExchange(MQConstant.DEFAULT_EXCHANGE, true, false);
+    }
+
+    /*********************    业务队列定义与绑定 hello 测试    *****************/
+    @Bean
+    public Queue queue() {
+        Queue queue = new Queue(MQConstant.HELLO_QUEUE_NAME,true);
+        return queue;
+    }
+
+    @Bean
+    public Binding binding() {
+        //队列绑定到exchange上，再绑定好路由键
+        return BindingBuilder.bind(queue()).to(defaultExchange()).with(MQConstant.HELLO_QUEUE_NAME);
+    }
+    /*********************    业务队列定义与绑定 hello 测试    *****************/
+
+    //下面是延迟队列的配置
+    //转发队列
+    @Bean
+    public Queue repeatTradeQueue() {
+        Queue queue = new Queue(MQConstant.DEFAULT_REPEAT_TRADE_QUEUE_NAME,true,false,false);
+        return queue;
+    }
+    //绑定转发队列
+    @Bean
+    public Binding  drepeatTradeBinding() {
+        return BindingBuilder.bind(repeatTradeQueue()).to(defaultExchange()).with(MQConstant.DEFAULT_REPEAT_TRADE_QUEUE_NAME);
+    }
+
+    //死信队列  -- 消息在死信队列上堆积，消息超时时，会把消息转发到转发队列，转发队列根据消息内容再把转发到指定的队列上
+    @Bean
+    public Queue deadLetterQueue() {
+        Map<String, Object> arguments = new HashMap<>();
+        arguments.put("x-dead-letter-exchange", MQConstant.DEFAULT_EXCHANGE);
+        arguments.put("x-dead-letter-routing-key", MQConstant.DEFAULT_REPEAT_TRADE_QUEUE_NAME);
+        Queue queue = new Queue(MQConstant.DEFAULT_DEAD_LETTER_QUEUE_NAME,true,false,false,arguments);
+        return queue;
+    }
+    //绑定死信队列
+    @Bean
+    public Binding  deadLetterBinding() {
+        return BindingBuilder.bind(deadLetterQueue()).to(defaultExchange()).with(MQConstant.DEFAULT_DEAD_LETTER_QUEUE_NAME);
+    }
+}

carbon-h5/carbon-h5-service/src/main/java/com/hcloud/microserver/h5/config/mq/RabbitMQConfiguration.java

@@ -0,0 +1,54 @@
+package com.hcloud.microserver.h5.config.mq;
+
+import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
+import org.springframework.amqp.rabbit.connection.ConnectionFactory;
+import org.springframework.amqp.rabbit.core.RabbitTemplate;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.beans.factory.config.ConfigurableBeanFactory;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.context.annotation.Scope;
+
+import java.util.logging.Logger;
+
+/**
+ * 读取application.properties中的连接配置
+ */
+@Configuration
+public class RabbitMQConfiguration {
+
+    private static Logger logger = Logger.getLogger("RabbitMQConfiguration");
+
+    @Value("${spring.rabbitmq.host}")
+    public String host;
+
+    @Value("${spring.rabbitmq.port}")
+    public int port;
+
+    @Value("${spring.rabbitmq.username}")
+    public String username;
+
+    @Value("${spring.rabbitmq.password}")
+    public String password;
+
+    @Value("${spring.rabbitmq.virtual-host}")
+    public String virtualHost;
+
+    @Bean
+    public ConnectionFactory connectionFactory() {
+        CachingConnectionFactory connectionFactory = new CachingConnectionFactory(host, port);
+        connectionFactory.setUsername(username);
+        connectionFactory.setPassword(password);
+        connectionFactory.setVirtualHost(virtualHost);
+        connectionFactory.setPublisherConfirms(true);
+        logger.info("Create ConnectionFactory bean ..");
+        return connectionFactory;
+    }
+
+    @Bean
+    @Scope(ConfigurableBeanFactory.SCOPE_PROTOTYPE)
+    public RabbitTemplate rabbitTemplate() {
+        RabbitTemplate template = new RabbitTemplate(connectionFactory());
+        return template;
+    }
+}

carbon-h5/carbon-h5-service/src/main/java/com/hcloud/microserver/h5/config/mq/SysInitLoad.java

@@ -0,0 +1,27 @@
+package com.hcloud.microserver.h5.config.mq;
+
+import com.hcloud.microserver.h5.config.mq.service.impl.IMessageService;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.ApplicationArguments;
+import org.springframework.boot.ApplicationRunner;
+import org.springframework.stereotype.Component;
+
+/**
+ * 启动启动时执行
+ */
+@Component
+public class SysInitLoad implements ApplicationRunner {
+
+    @Autowired
+    private IMessageService messageService;
+
+    @Override
+    public void run(ApplicationArguments args) throws Exception {
+        System.out.println("发送时间:" + System.currentTimeMillis());
+        String message = "测试延迟消息";
+//        messageService.send(MQConstant.HELLO_QUEUE_NAME, message, 6000);
+//
+//        message = "测试普通消息";
+//        messageService.send(MQConstant.HELLO_QUEUE_NAME, message);
+    }
+}

carbon-h5/carbon-h5-service/src/main/java/com/hcloud/microserver/h5/config/mq/listener/ReceiverDelayMessage.java

@@ -0,0 +1,28 @@
+package com.hcloud.microserver.h5.config.mq.listener;
+
+import com.alibaba.fastjson.JSON;
+import com.hcloud.microserver.h5.config.mq.MQConstant;
+import com.hcloud.microserver.h5.config.mq.service.DLXMessage;
+import com.hcloud.microserver.h5.config.mq.service.impl.IMessageService;
+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 = MQConstant.DEFAULT_REPEAT_TRADE_QUEUE_NAME)
+public class ReceiverDelayMessage {
+
+    @Autowired
+    private IMessageService messageService;
+
+    @RabbitHandler
+    public void process(String content) {
+        //此时，才把消息发送到指定队列，而实现延迟功能
+        DLXMessage message = JSON.parseObject(content, DLXMessage.class);
+        messageService.send(message.getQueueName(), message.getContent());
+    }
+
+}

carbon-h5/carbon-h5-service/src/main/java/com/hcloud/microserver/h5/config/mq/listener/ReceiverMessage.java

@@ -0,0 +1,39 @@
+package com.hcloud.microserver.h5.config.mq.listener;
+
+import com.hcloud.microserver.facade.carbon.forms.GoodsOrderInfoForm;
+import com.hcloud.microserver.h5.config.mq.MQConstant;
+import com.hcloud.microserver.h5.service.OrderDetailInfoService;
+import com.hcloud.microserver.h5.service.OrderInfoService;
+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;
+
+//监听hello队列，有消息时进行消费
+@Component
+@RabbitListener(queues = MQConstant.HELLO_QUEUE_NAME)
+public class ReceiverMessage {
+
+    @Autowired
+    private OrderInfoService orderInfoService;
+    @Autowired
+    private OrderDetailInfoService orderDetailInfoService;
+
+    @RabbitHandler
+    public void process(String content) {
+        System.out.println("接受时间:" + System.currentTimeMillis());
+        System.out.println("接受消息:" + content);
+        String[] split = content.split("-");
+        if(split.length>1){
+            String orderId = split[1];
+            GoodsOrderInfoForm orderInfoBO = orderDetailInfoService.info(orderId);
+
+            Integer orderStatus = orderInfoBO.getOrderStatus();
+            if(orderStatus.intValue() ==1){
+                orderInfoService.rollbackOrderInfo(orderId);
+                System.out.println("取消订单:" + content);
+            }
+
+        }
+    }
+}

carbon-h5/carbon-h5-service/src/main/java/com/hcloud/microserver/h5/config/mq/service/DLXMessage.java

@@ -0,0 +1,72 @@
+package com.hcloud.microserver.h5.config.mq.service;
+
+import java.io.Serializable;
+
+/**
+ * rabbit 死信消息载体
+ */
+public class DLXMessage implements Serializable {
+
+    private static final long serialVersionUID = 9956432152000L;
+    private String exchange;
+    private String queueName;
+    private String content;
+    private long times;
+
+    public DLXMessage() {
+        super();
+    }
+
+    public DLXMessage(String queueName, String content, long times) {
+        super();
+        this.queueName = queueName;
+        this.content = content;
+        this.times = times;
+    }
+
+    public DLXMessage(String exchange, String queueName, String content, long times) {
+        super();
+        this.exchange = exchange;
+        this.queueName = queueName;
+        this.content = content;
+        this.times = times;
+    }
+
+
+    public static long getSerialVersionUID() {
+        return serialVersionUID;
+    }
+
+    public String getExchange() {
+        return exchange;
+    }
+
+    public void setExchange(String exchange) {
+        this.exchange = exchange;
+    }
+
+    public String getQueueName() {
+        return queueName;
+    }
+
+    public void setQueueName(String queueName) {
+        this.queueName = queueName;
+    }
+
+    public String getContent() {
+        return content;
+    }
+
+    public void setContent(String content) {
+        this.content = content;
+    }
+
+    public long getTimes() {
+        return times;
+    }
+
+    public void setTimes(long times) {
+        this.times = times;
+    }
+}
+

carbon-h5/carbon-h5-service/src/main/java/com/hcloud/microserver/h5/config/mq/service/impl/IMessageService.java

@@ -0,0 +1,23 @@
+package com.hcloud.microserver.h5.config.mq.service.impl;
+
+public interface IMessageService {
+
+    /**
+     * 发送消息到队列
+     * @param queueName 队列名称
+     * @param message 消息内容
+     */
+    public void send(String queueName, String message);
+
+
+    /**
+     * 延迟发送消息到队列
+     * @param queueName 队列名称
+     * @param message 消息内容
+     * @param times 延迟时间 单位毫秒
+     */
+    public void send(String queueName, String message, long times);
+}
+
+
+

carbon-h5/carbon-h5-service/src/main/java/com/hcloud/microserver/h5/controller/web/OrderManagerController.java

@@ -2,6 +2,7 @@ package com.hcloud.microserver.h5.controller.web;
 
 import com.github.pagehelper.PageInfo;
 import com.google.common.collect.Lists;
+import com.hcloud.microserver.facade.carbon.entity.GoodsOrderInfo;
 import com.hcloud.microserver.facade.carbon.entity.from.CreateOrderFrom;
 import com.hcloud.microserver.h5.bo.CredentialInfoBO;
 import com.hcloud.microserver.h5.bo.OrderInfoBO;
@@ -68,8 +69,9 @@ public class OrderManagerController extends CarbonBaseController {
     @AuthCarbonValidate
     public ResponseBase createOrder(@RequestBody CreateOrderFrom createOrderFrom) {
 
-        orderInfoService.createOrder(createOrderFrom);
-        return resultVOSuccess();
+        GoodsOrderInfo goodsOrderInfo = orderInfoService.createOrder(createOrderFrom);
+
+        return resultVOSuccess(goodsOrderInfo);
     }
 
 
@@ -164,8 +166,6 @@ public class OrderManagerController extends CarbonBaseController {
 
 
 
-    //商品秒杀核心业务逻辑的处理-redisson的分布式锁
-
     public ResultVO saveOrderInfoAndPayOrderInfo(GoodsOrderInfoForm goodsOrderInfoForm) {
         int i = convertOrderInfoForm(goodsOrderInfoForm);
         if (i == 1) {

carbon-h5/carbon-h5-service/src/main/java/com/hcloud/microserver/h5/service/OrderInfoService.java

@@ -2,6 +2,7 @@ package com.hcloud.microserver.h5.service;
 
 import com.github.pagehelper.PageInfo;
 import com.hcloud.microserver.commoncore.base.BaseServie;
+import com.hcloud.microserver.facade.carbon.entity.GoodsOrderInfo;
 import com.hcloud.microserver.facade.carbon.entity.from.CreateOrderFrom;
 import com.hcloud.microserver.facade.carbon.forms.GoodsOrderInfoForm;
 
@@ -75,5 +76,5 @@ public interface OrderInfoService extends BaseServie<GoodsOrderInfoForm,String>
     GoodsOrderInfoForm saveOrderInfoForOffline(GoodsOrderInfoForm goodsOrderInfoForm);
 
 
-    void createOrder(CreateOrderFrom createOrderFrom);
+    GoodsOrderInfo createOrder(CreateOrderFrom createOrderFrom);
 }

carbon-h5/carbon-h5-service/src/main/java/com/hcloud/microserver/h5/service/impl/OrderInfoServiceImpl.java

@@ -8,6 +8,8 @@ import com.hcloud.microserver.facade.carbon.entity.GoodsOrderDetail;
 import com.hcloud.microserver.facade.carbon.entity.from.CreateOrderDetailFrom;
 import com.hcloud.microserver.facade.carbon.entity.from.CreateOrderFrom;
 import com.hcloud.microserver.h5.bo.OrderInfoBO;
+import com.hcloud.microserver.h5.config.mq.MQConstant;
+import com.hcloud.microserver.h5.config.mq.service.impl.IMessageService;
 import com.hcloud.microserver.h5.dao.GoodsInfoMapper;
 import com.hcloud.microserver.h5.dao.GoodsOrderDetailMapper;
 import com.hcloud.microserver.h5.dao.GoodsOrderInfoMapper;
@@ -499,10 +501,12 @@ public class OrderInfoServiceImpl implements OrderInfoService {
 
     @Autowired
     private GoodsOrderDetailMapper goodsOrderDetailMapper;
+    @Autowired
+    private IMessageService messageService;
 
     @Override
     @Transactional
-    public void createOrder(CreateOrderFrom createOrderFrom) {
+    public GoodsOrderInfo createOrder(CreateOrderFrom createOrderFrom) {
         //获取登陆人信息
         CustomerInfoForm customerInfo = customerService.getCurrentMember();
         if (customerInfo == null) {
@@ -555,6 +559,11 @@ public class OrderInfoServiceImpl implements OrderInfoService {
         //新增小定单
         goodsOrderDetailMapper.batchSaveOrderDetailList(dataList);
 
+        String message = "order-" + goodsOrderInfo.getGuid();
+        //延时消息队列
+        messageService.send(MQConstant.HELLO_QUEUE_NAME, message, 60000);
+        return goodsOrderInfo;
+
     }
 
     private void addGoodsOrderInfo(CustomerInfoForm customerInfo, String customerId, BigDecimal sumTotal, BigDecimal carbonTotal, String orderId, GoodsOrderInfo goodsOrderInfo) {

sql/表结构变化语句/更新sql.sql

@@ -0,0 +1,2 @@
+##订单详情表新增
+ALTER TABLE t_goods_order_detail ADD COLUMN `carbon_val` decimal(10,2) DEFAULT NULL COMMENT '碳汇量' AFTER `amount`;