优惠券活动创建

admin/src/main/resources/application-dev.yml

@@ -82,19 +82,19 @@ spring:
       datasource:
         db-admin:
           url: jdbc:mysql://121.40.98.15:3307/charge_admin?serverTimezone=Asia/Shanghai
-          username: kym
-          password: kx7#zFvS$$%uVDkZPk
+          username: root
+          password: KuaiyuMan/*-
           driver-class-name: com.mysql.cj.jdbc.Driver
         db-miniapp:
           url: jdbc:mysql://121.40.98.15:3307/charge_app?serverTimezone=Asia/Shanghai
-          username: kym
-          password: kx7#zFvS$$%uVDkZPk
+          username: root
+          password: KuaiyuMan/*-
           driver-class-name: com.mysql.cj.jdbc.Driver
   data:
     redis:
       port: 6380
       host: 121.40.98.15
-      password: kuaiyuman@3rt
+      password: KtXA^Zx!TZmLEy(@JjB@2(TVG0kdy5)&
       database: 0
       lettuce:
         pool:
@@ -108,6 +108,13 @@ spring:
     redis:
       # 缓存过期时间：7天
       time-to-live: 604800
+  rabbitmq:
+    host: 127.0.0.1
+    port: 5672
+    username: guest
+    password: guest
+    virtual-host: /
+    publisher-returns: true
 
 kym:
-  notify-email: skyline@kuaiyuman.cn
+  notify-email: skyline@kuaiyuman.cn

entity/src/main/java/com/kym/entity/admin/Activity.java

@@ -3,7 +3,7 @@ package com.kym.entity.admin;
 import com.baomidou.mybatisplus.annotation.TableName;
 import com.fasterxml.jackson.annotation.JsonFormat;
 import com.kym.entity.BaseEntity;
-import java.io.Serializable;
+
 import java.time.LocalDateTime;
 import lombok.Getter;
 import lombok.Setter;
@@ -29,6 +29,10 @@ public class Activity extends BaseEntity {
     public final static int APPLY_STATION_全部站点 = 0;
     public final static int APPLY_STATION_部分站点 = 1;
 
+    public final static int TARGET_USERS_全部用户 = 0;
+    public final static int TARGET_USERS_新用户 = 1;
+    public final static int TARGET_USERS_指定用户 = 2;
+
     public final static int STATUS_未开始 = 0;
     public final static int STATUS_进行中 = 1;
     public final static int STATUS_已结束 = 2;
@@ -71,7 +75,7 @@ public class Activity extends BaseEntity {
     private String discountType;
 
     /**
-     * 目标用户：0-全部，1-新用户，2-老用户
+     * 目标用户：0-全部，1-新用户，2-部分用户
      */
     private Integer targetUsers;
 

entity/src/main/java/com/kym/entity/admin/Coupon.java

@@ -25,6 +25,13 @@ public class Coupon extends BaseEntity {
 
     private static final long serialVersionUID = 1L;
 
+    public static final String COUPON_TYPE_折扣券 = "Discount";
+    public static final String COUPON_TYPE_满减券 = "FullDiscount";
+
+    public static final int STATUS_无效 = 0;
+    public static final int STATUS_有效 = 1;
+
+
     /**
      * 活动id
      */

entity/src/main/java/com/kym/entity/common/Queues.java

@@ -0,0 +1,14 @@
+package com.kym.entity.common;
+
+/**
+ * 消息队列
+ *
+ * @author skyline
+ */
+public class Queues {
+
+    /**
+     * 优惠券
+     */
+    public static final String QUEUE_MARKETING_COUPON = "MARKETING_COUPON";
+}

entity/src/main/java/com/kym/entity/miniapp/UserCoupon.java

@@ -4,6 +4,7 @@ import com.baomidou.mybatisplus.annotation.TableName;
 import com.kym.entity.BaseEntity;
 import lombok.Getter;
 import lombok.Setter;
+import lombok.experimental.Accessors;
 
 import java.time.LocalDateTime;
 
@@ -18,6 +19,7 @@ import java.time.LocalDateTime;
 @Getter
 @Setter
 @TableName("t_user_coupon")
+@Accessors(chain = true)
 public class UserCoupon extends BaseEntity {
 
     private static final long serialVersionUID = 1L;

miniapp/src/main/resources/application-dev.yml

@@ -82,13 +82,13 @@ spring:
       datasource:
         db-admin:
           url: jdbc:mysql://121.40.98.15:3307/charge_admin?tinyInt1isBit=false&serverTimezone=Asia/Shanghai
-          username: kym
-          password: kx7#zFvS$$%uVDkZPk
+          username: root
+          password: KuaiyuMan/*-
           driver-class-name: com.mysql.cj.jdbc.Driver
         db-miniapp:
           url: jdbc:mysql://121.40.98.15:3307/charge_app
-          username: kym
-          password: kx7#zFvS$$%uVDkZPk
+          username: root
+          password: KuaiyuMan/*-
           driver-class-name: com.mysql.cj.jdbc.Driver
   data:
     redis:

service/pom.xml

@@ -43,6 +43,12 @@
             <version>4.5.0</version>
         </dependency>
 
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter-amqp</artifactId>
+            <version>3.2.5</version>
+        </dependency>
+
     </dependencies>
 
 </project>

service/src/main/java/com/kym/service/admin/impl/ActivityServiceImpl.java

@@ -2,6 +2,7 @@ package com.kym.service.admin.impl;
 
 import com.baomidou.dynamic.datasource.annotation.DS;
 import com.baomidou.dynamic.datasource.toolkit.DynamicDataSourceContextHolder;
+import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
 import com.github.pagehelper.PageHelper;
 import com.github.yulichang.base.MPJBaseServiceImpl;
 import com.kym.common.utils.CommUtil;
@@ -11,11 +12,16 @@ import com.kym.entity.admin.delay.DelayActivity;
 import com.kym.entity.admin.queryParams.ActivityQueryParam;
 import com.kym.entity.admin.vo.ActivityVo;
 import com.kym.entity.common.PageBean;
+import com.kym.entity.miniapp.User;
+import com.kym.entity.miniapp.UserCoupon;
 import com.kym.entity.miniapp.UserRechargeRights;
 import com.kym.mapper.admin.ActivityMapper;
 import com.kym.service.admin.*;
 import com.kym.service.jobs.DelayService;
+import com.kym.service.miniapp.UserCouponService;
 import com.kym.service.miniapp.UserRechargeRightsService;
+import com.kym.service.miniapp.UserService;
+import com.kym.service.queue.producer.CouponMessageProducer;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.beans.BeanUtils;
 import org.springframework.context.annotation.Lazy;
@@ -25,8 +31,9 @@ import org.springframework.transaction.annotation.Transactional;
 
 import java.time.LocalDateTime;
 import java.time.LocalTime;
-import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Executors;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.stream.Collectors;
 
 import static com.kym.entity.admin.Activity.APPLY_STATION_部分站点;
 import static com.kym.entity.admin.Activity.DISCOUNT_TYPE_服务费折扣权益;
@@ -54,9 +61,15 @@ public class ActivityServiceImpl extends MPJBaseServiceImpl<ActivityMapper, Acti
     private final DelayService<DelayActivity> activityDelayService;
     private final BannerService bannerService;
 
+    private final UserService userService;
+
+    private final UserCouponService userCouponService;
+
+    private final CouponMessageProducer couponMessageProducer;
+
     public ActivityServiceImpl(ActivityStationService activityStationService, RechargeRightsService rechargeRightsService, CouponService couponService,
                                UserRechargeRightsService userRechargeRightsService, StationService stationService,
-                               @Lazy DelayService<DelayActivity> activityDelayService, BannerService bannerService) {
+                               @Lazy DelayService<DelayActivity> activityDelayService, BannerService bannerService, UserService userService, UserCouponService userCouponService, CouponMessageProducer couponMessageProducer) {
         this.activityStationService = activityStationService;
         this.rechargeRightsService = rechargeRightsService;
         this.couponService = couponService;
@@ -64,6 +77,9 @@ public class ActivityServiceImpl extends MPJBaseServiceImpl<ActivityMapper, Acti
         this.stationService = stationService;
         this.activityDelayService = activityDelayService;
         this.bannerService = bannerService;
+        this.userService = userService;
+        this.userCouponService = userCouponService;
+        this.couponMessageProducer = couponMessageProducer;
     }
 
     /**
@@ -81,7 +97,7 @@ public class ActivityServiceImpl extends MPJBaseServiceImpl<ActivityMapper, Acti
         activity.setId(IDGenerator.INS().nextId());
         save(activity);
 
-        // 新增activityStation
+        // 新增activityStation 活动-站点关联
         if (activityVo.getApplyStation() == APPLY_STATION_部分站点 && !CommUtil.isEmptyOrNull(activityVo.getStationIds())) {
             var activityStationList = activityVo.getStationIds().stream().map(stationId -> new ActivityStation().setActivityId(activity.getId()).setStationId(stationId)).toList();
             activityStationService.saveBatch(activityStationList);
@@ -98,11 +114,12 @@ public class ActivityServiceImpl extends MPJBaseServiceImpl<ActivityMapper, Acti
 
         // 新增优惠券活动
         if (activityVo.getDiscountType().equals(Activity.DISCOUNT_TYPE_优惠券) && !CommUtil.isEmptyOrNull(activityVo.getCouponList())) {
-            var couponList = activityVo.getCouponList().stream().peek(coupon -> {
+            List<Coupon> couponList = activityVo.getCouponList().stream().peek(coupon -> {
                 BeanUtils.copyProperties(activity, coupon, "id");
                 coupon.setActivityId(activity.getId());
-            }).toList();
+            }).collect(Collectors.toList());
             couponService.saveBatch(couponList);
+            handleUserCoupons(activity, activityVo);
         }
 
         // 加入活动延迟启停队列
@@ -111,6 +128,54 @@ public class ActivityServiceImpl extends MPJBaseServiceImpl<ActivityMapper, Acti
         activityDelayService.addToDelayQueue(new DelayActivity(activity.getId(), activity.getName(), activity.getEndTime(), DelayActivity.TYPE_结束));
     }
 
+
+    /**
+     * 处理优惠券用户绑定
+     *
+     * @param activity
+     * @param activityVo
+     */
+    private void handleUserCoupons(Activity activity, ActivityVo activityVo) {
+        List<UserCoupon> userCouponList = new ArrayList<>();
+        var couponList = activityVo.getCouponList();
+
+        var targetUsersType = activityVo.getTargetUsers();
+        switch (targetUsersType) {
+            case Activity.TARGET_USERS_全部用户:
+                userService.listObjs(new QueryWrapper<User>().lambda().select(User::getId))
+                        .forEach(useId -> couponList.forEach(coupon -> addUserCoupon(userCouponList, coupon, activity.getId(), Long.valueOf(useId.toString()))));
+                break;
+            case Activity.TARGET_USERS_指定用户:
+                activityVo.getTargetUserIds().forEach(useId -> couponList.forEach(coupon -> addUserCoupon(userCouponList, coupon, activity.getId(), useId)));
+                break;
+        }
+        userCouponService.saveBatch(userCouponList);
+        userCouponList.forEach(couponMessageProducer::sendMessage);
+    }
+
+    /**
+     * 添加用户优惠券
+     *
+     * @param userCouponList
+     * @param coupon
+     * @param activityId
+     * @param userId
+     */
+    private void addUserCoupon(List<UserCoupon> userCouponList, Coupon coupon, Long activityId, Long userId) {
+        var userCoupon = new UserCoupon()
+                .setCouponId(coupon.getId())
+                .setActivityId(activityId)
+                .setUserId(userId)
+                .setStartTime(coupon.getStartTime())
+                .setEndTime(coupon.getEndTime())
+                .setCouponType(coupon.getCouponType())
+                .setDiscount(coupon.getDiscount())
+                .setMinServiceMoney(coupon.getMinServiceMoney())
+                .setAllowStacke(coupon.getAllowStacke())
+                .setStatus(Coupon.STATUS_有效);
+        userCouponList.add(userCoupon);
+    }
+
     /**
      * 主活动列表
      *

service/src/main/java/com/kym/service/miniapp/impl/UserServiceImpl.java

@@ -32,6 +32,7 @@ import lombok.SneakyThrows;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.beans.BeanUtils;
+import org.springframework.context.annotation.Lazy;
 import org.springframework.stereotype.Service;
 import org.springframework.transaction.annotation.Transactional;
 
@@ -65,7 +66,7 @@ public class UserServiceImpl extends MPJBaseServiceImpl<UserMapper, User> implem
 
     public UserServiceImpl(WxConfig wxConfig, AccountService accountService, RefundLogService refundLogService,
                            CarsService carsService, UserRechargeRightsService userRechargeRightsService,
-                           RechargeRightsService rechargeRightsService, ActivityService activityService, BannerService bannerService) {
+                           RechargeRightsService rechargeRightsService, @Lazy ActivityService activityService, BannerService bannerService) {
         this.wxConfig = wxConfig;
         this.accountService = accountService;
         this.refundLogService = refundLogService;

service/src/main/java/com/kym/service/queue/MessageQueueService.java

@@ -0,0 +1,8 @@
+package com.kym.service.queue;
+
+/**
+ * 消息队列服务
+ */
+public interface MessageQueueService {
+    void sendMessage(String exchange, String routingKey, Object msg);
+}

service/src/main/java/com/kym/service/queue/RabbitMQConfig.java

@@ -0,0 +1,44 @@
+package com.kym.service.queue;
+
+import jakarta.annotation.Resource;
+import org.springframework.amqp.core.AcknowledgeMode;
+import org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory;
+import org.springframework.amqp.rabbit.connection.ConnectionFactory;
+import org.springframework.amqp.rabbit.core.RabbitAdmin;
+import org.springframework.amqp.rabbit.listener.RabbitListenerContainerFactory;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+
+@Configuration
+public class RabbitMQConfig {
+
+    //配置连接工厂
+    @Resource
+    private ConnectionFactory connectionFactory;
+
+    @Bean
+    public RabbitAdmin rabbitAdmin() {
+        //需要传入
+        RabbitAdmin rabbitAdmin = new RabbitAdmin(connectionFactory);
+        rabbitAdmin.setAutoStartup(true);
+        return rabbitAdmin;
+    }
+
+
+    /**
+     * 设置成手动ack
+     * https://blog.csdn.net/m912595719/article/details/83787486
+     * @param connectionFactory
+     * @return
+     */
+    @Bean
+    public RabbitListenerContainerFactory<?> rabbitListenerContainerFactory(ConnectionFactory connectionFactory) {
+        SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
+        factory.setConnectionFactory(connectionFactory);
+        // factory.setMessageConverter(new Jackson2JsonMessageConverter());
+        factory.setAcknowledgeMode(AcknowledgeMode.MANUAL);
+        return factory;
+    }
+
+
+}

service/src/main/java/com/kym/service/queue/callback/ConfirmCallbackService.java

@@ -0,0 +1,21 @@
+package com.kym.service.queue.callback;
+
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.amqp.rabbit.connection.CorrelationData;
+import org.springframework.amqp.rabbit.core.RabbitTemplate;
+import org.springframework.stereotype.Component;
+
+@Slf4j
+@Component
+public class ConfirmCallbackService implements RabbitTemplate.ConfirmCallback {
+
+    @Override
+    public void confirm(CorrelationData correlationData, boolean ack, String cause) {
+
+        if (!ack) {
+            log.error("消息发送异常!");
+        } else {
+            log.info("发送者已经收到确认，correlationData={} ,ack={}, cause={}", correlationData.getId(), ack, cause);
+        }
+    }
+}

service/src/main/java/com/kym/service/queue/callback/ReturnCallbackService.java

@@ -0,0 +1,17 @@
+package com.kym.service.queue.callback;
+
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.amqp.core.ReturnedMessage;
+import org.springframework.amqp.rabbit.core.RabbitTemplate;
+import org.springframework.stereotype.Component;
+
+
+@Slf4j
+@Component
+public class ReturnCallbackService implements RabbitTemplate.ReturnsCallback {
+
+    @Override
+    public void returnedMessage(ReturnedMessage returnedMessage) {
+        log.info("returnedMessage ===> replyCode={} ,replyText={} ,exchange={} ,routingKey={}", returnedMessage.getReplyCode(), returnedMessage.getReplyText(), returnedMessage.getExchange(), returnedMessage.getExchange());
+    }
+}

service/src/main/java/com/kym/service/queue/consumer/ReceiverMessage.java

@@ -0,0 +1,46 @@
+package com.kym.service.queue.consumer;
+
+import com.kym.entity.common.Queues;
+import com.rabbitmq.client.Channel;
+import lombok.extern.slf4j.Slf4j;
+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 java.io.IOException;
+
+/**
+ * @author 公众号：程序员小富
+ * @description 消息消费
+ * @date 2020/6/29 16:31
+ */
+@Slf4j
+@Component
+
+public class ReceiverMessage {
+
+    @RabbitListener(queues = Queues.QUEUE_MARKETING_COUPON)
+    @RabbitHandler
+    public void messageHandler(String msg, Channel channel, Message message) throws IOException {
+
+        try {
+            log.info("消费者 1 号收到：{}", msg);
+
+            //TODO 具体业务
+
+            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
+
+        } catch (Exception e) {
+            if (message.getMessageProperties().getRedelivered()) {
+                log.error("消息已重复处理失败,拒绝再次接收...");
+                // 拒绝消息
+                channel.basicReject(message.getMessageProperties().getDeliveryTag(), false);
+            } else {
+                log.error("消息即将再次返回队列处理...");
+                channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, true);
+            }
+        }
+    }
+
+}

service/src/main/java/com/kym/service/queue/impl/MessageQueueServiceImpl.java

@@ -0,0 +1,53 @@
+package com.kym.service.queue.impl;
+
+import com.kym.service.queue.MessageQueueService;
+import com.kym.service.queue.callback.ConfirmCallbackService;
+import com.kym.service.queue.callback.ReturnCallbackService;
+import org.springframework.amqp.core.MessageDeliveryMode;
+import org.springframework.amqp.rabbit.connection.CorrelationData;
+import org.springframework.amqp.rabbit.core.RabbitTemplate;
+import org.springframework.stereotype.Service;
+
+import java.util.UUID;
+
+/**
+ * 消息队列服务
+ *
+ * @author skyline
+ */
+@Service
+public class MessageQueueServiceImpl implements MessageQueueService {
+
+    private final RabbitTemplate rabbitTemplate;
+    private final ConfirmCallbackService confirmCallbackService;
+
+    private final ReturnCallbackService returnCallbackService;
+
+
+    public MessageQueueServiceImpl(RabbitTemplate rabbitTemplate, ConfirmCallbackService confirmCallbackService, ReturnCallbackService returnCallbackService) {
+        this.rabbitTemplate = rabbitTemplate;
+        this.confirmCallbackService = confirmCallbackService;
+        this.returnCallbackService = returnCallbackService;
+    }
+
+    @Override
+    public void sendMessage(String exchange, String routingKey, Object msg) {
+
+        //  确保消息发送失败后可以重新返回到队列中, 注意：yml需要配置 publisher-returns: true
+        rabbitTemplate.setMandatory(true);
+        // 消费者确认收到消息后，手动ack回执回调处理
+        rabbitTemplate.setConfirmCallback(confirmCallbackService);
+
+        // 消息投递到队列失败回调处理
+        rabbitTemplate.setReturnsCallback(returnCallbackService);
+        // 发送消息
+        rabbitTemplate.convertAndSend(exchange, routingKey, msg,
+                message -> {
+                    message.getMessageProperties().setDeliveryMode(MessageDeliveryMode.PERSISTENT);
+                    return message;
+                },
+                new CorrelationData(UUID.randomUUID().toString()));
+    }
+
+
+}

service/src/main/java/com/kym/service/queue/producer/CouponMessageProducer.java

@@ -0,0 +1,56 @@
+package com.kym.service.queue.producer;
+
+import com.kym.entity.common.Queues;
+import com.kym.service.queue.MessageQueueService;
+import org.springframework.amqp.AmqpException;
+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.beans.factory.annotation.Qualifier;
+import org.springframework.context.annotation.Bean;
+import org.springframework.scheduling.annotation.Async;
+import org.springframework.stereotype.Component;
+
+/**
+ * 优惠券消息生产者
+ *
+ * @author skyline
+ */
+@Component
+public class CouponMessageProducer {
+
+    private static final String EXCHANGE = "kym";
+    private static final String ROUTING_KEY = "coupon";
+    private final MessageQueueService messageQueueService;
+
+    public CouponMessageProducer(MessageQueueService messageQueueService) {
+        this.messageQueueService = messageQueueService;
+    }
+
+    @Async
+    public void  sendMessage(Object message) {
+        try {
+            messageQueueService.sendMessage(EXCHANGE, ROUTING_KEY, message);
+        } catch (AmqpException e) {
+            e.printStackTrace();
+        }
+    }
+
+    @Bean(name = "queue")
+    protected Queue queue() {
+        return new Queue(Queues.QUEUE_MARKETING_COUPON, true, false, false);
+    }
+
+    @Bean(name = "exchange")
+    protected DirectExchange exchange() {
+        return new DirectExchange(EXCHANGE, true, false);
+    }
+
+    @Bean
+    private Binding binding(@Qualifier("exchange") DirectExchange exchange, @Qualifier("queue") Queue queue) {
+        return BindingBuilder.bind(queue).to(exchange).with(ROUTING_KEY);
+    }
+
+
+}