提交盲票购买订单

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

@@ -111,7 +111,8 @@ mq:
     service-url: http://pulsar-rkrxw2wx8zeo.tdmq-pulsar.ap-sh.public.tencenttdmq.com:8080
     auth-token: eyJrZXlJZCI6InB1bHNhci1ya3J4dzJ3eDh6ZW8iLCJhbGciOiJIUzI1NiJ9.eyJzdWIiOiJwdWxzYXItcmtyeHcyd3g4emVvX21wLXNlcnZlci10ZXN0In0.12cemXSqXGn9beDFnB-uUe477G6pP9bNbDUA4fIRsgk
     topic-ticket-generate: pulsar-rkrxw2wx8zeo/test-share/topic-ticket-generate
-    consumer-topics: pulsar-rkrxw2wx8zeo/test-share/topic-ticket-generate
+    topic-ticket-pay: pulsar-rkrxw2wx8zeo/test-share/topic-ticket-pay
+    consumer-topics: pulsar-rkrxw2wx8zeo/test-share/topic-ticket-generate,pulsar-rkrxw2wx8zeo/test-share/topic-ticket-pay
 
 # 小程序
 miniprogram:

mp-admin/src/main/resources/application-test.yml

@@ -116,7 +116,8 @@ mq:
   service-url: http://pulsar-rkrxw2wx8zeo.tdmq-pulsar.ap-sh.public.tencenttdmq.com:8080
   auth-token: eyJrZXlJZCI6InB1bHNhci1ya3J4dzJ3eDh6ZW8iLCJhbGciOiJIUzI1NiJ9.eyJzdWIiOiJwdWxzYXItcmtyeHcyd3g4emVvX21wLXNlcnZlci10ZXN0In0.12cemXSqXGn9beDFnB-uUe477G6pP9bNbDUA4fIRsgk
   topic-ticket-generate: pulsar-rkrxw2wx8zeo/test-share/topic-ticket-generate
-  consumer-topics: pulsar-rkrxw2wx8zeo/test-share/topic-ticket-generate
+  topic-ticket-pay: pulsar-rkrxw2wx8zeo/test-share/topic-ticket-pay
+  consumer-topics: pulsar-rkrxw2wx8zeo/test-share/topic-ticket-generate,pulsar-rkrxw2wx8zeo/test-share/topic-ticket-pay
 
 
 # 小程序

mp-common/src/main/java/com/qs/mp/common/enums/MqTopicType.java

@@ -9,7 +9,8 @@ import com.baomidou.mybatisplus.annotation.IEnum;
  */
 public enum MqTopicType implements IEnum<String> {
 
-  ticket_generate("1", "盲票生成");
+  ticket_generate("1", "盲票生成"),
+  ticket_pay("2", "盲票付款"),;
 
 
   private final String value;

mp-common/src/main/java/com/qs/mp/common/pulsar/PulsarClientService.java

@@ -42,6 +42,9 @@ public class PulsarClientService {
   @Value("${mq.topic-ticket-generate}")
   private String topicTicketGenerate;
 
+  @Value("${mq.topic-ticket-pay}")
+  private String topicTicketPay;
+
   @Value("${mq.consumer-conn}")
   private boolean consumerConn;
 
@@ -108,7 +111,9 @@ public class PulsarClientService {
     logger.info("start producer mq data:" + data);
     String topic = "";
     if (mqTopicType.getValue() == MqTopicType.ticket_generate.getValue()) {
-      topic = topicTicketGenerate; //
+      topic = topicTicketGenerate; // 盲票生成
+    } else if (mqTopicType.getValue() == MqTopicType.ticket_pay.getValue()) {
+      topic = topicTicketPay; // 盲票付款成功
     }
     Producer<byte[]> producer = null;
     if (Producers.containsKey(mqTopicType.getValue())) {

mp-service/src/main/java/com/qs/mp/mq/impl/PulsarConsumerImpl.java

@@ -9,6 +9,8 @@ import com.qs.mp.common.enums.MqTopicType;
 import com.qs.mp.common.enums.TicketBoxStatusEnum;
 import com.qs.mp.common.pulsar.PulsarConsumer;
 import com.qs.mp.common.utils.LogUtil;
+import com.qs.mp.user.domain.UserTicketOrder;
+import com.qs.mp.user.service.IUserTicketOrderService;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.beans.factory.annotation.Autowired;
@@ -29,26 +31,37 @@ public class PulsarConsumerImpl implements PulsarConsumer {
   @Autowired
   private ITicketBoxService ticketBoxService;
 
+  @Autowired
+  private IUserTicketOrderService userTicketOrderService;
+
   @Override
   @Async("threadPoolTaskExecutor")
   public void wsConsumer(String topicType, String mqData) {
     logger.info(topicType + "  >>>>>>>>>>>>>>>>>>>:" + mqData);
-    //监听商户充值消息，后续业务处理
+    //监听盲票生成消息，后续业务处理
     if (MqTopicType.ticket_generate.getValue().equals(topicType)) {
-      String boxId = mqData;
-      TicketBox ticketBox = ticketBoxService.getById(boxId);
-      if (ticketBox.getStatus() != TicketBoxStatusEnum.WAIT ) {
-        LogUtil.error(logger, "收到盲票生成消息，票组状态不是待出票，忽略消息。boxId=" + boxId);
-        return;
-      }
-      // 先更新状态，防并发。如果最后生成失败了，暂时人工重新触发
-      boolean rst = ticketBoxService.update(new LambdaUpdateWrapper<TicketBox>().set(TicketBox::getStatus, TicketBoxStatusEnum.DOING)
-          .eq(TicketBox::getBoxId, boxId).eq(TicketBox::getStatus, TicketBoxStatusEnum.WAIT));
-      if (!rst) {
-        LogUtil.error(logger, "收到盲票生成消息，更新票组状态为出票中失败。boxId=" + boxId);
-        return;
-      }
-      ticketBoxService.generateTicket(boxId);
+      processTicketGenerateMsg(mqData);
+    } else if (MqTopicType.ticket_pay.getValue().equals(topicType)) {
+      String orderId = mqData;
+      UserTicketOrder ticketOrder = userTicketOrderService.getById(orderId);
+
+    }
+  }
+
+  private void processTicketGenerateMsg(String mqData) {
+    String boxId = mqData;
+    TicketBox ticketBox = ticketBoxService.getById(boxId);
+    if (ticketBox.getStatus() != TicketBoxStatusEnum.WAIT ) {
+      LogUtil.error(logger, "收到盲票生成消息，票组状态不是待出票，忽略消息。boxId=" + boxId);
+      return;
+    }
+    // 先更新状态，防并发。如果最后生成失败了，暂时人工重新触发
+    boolean rst = ticketBoxService.update(new LambdaUpdateWrapper<TicketBox>().set(TicketBox::getStatus, TicketBoxStatusEnum.DOING)
+        .eq(TicketBox::getBoxId, boxId).eq(TicketBox::getStatus, TicketBoxStatusEnum.WAIT));
+    if (!rst) {
+      LogUtil.error(logger, "收到盲票生成消息，更新票组状态为出票中失败。boxId=" + boxId);
+      return;
     }
+    ticketBoxService.generateTicket(boxId);
   }
 }

mp-service/src/main/java/com/qs/mp/user/service/impl/UserTicketOrderServiceImpl.java

@@ -11,12 +11,14 @@ import com.qs.mp.admin.service.ITicketBoxService;
 import com.qs.mp.admin.service.ITicketService;
 import com.qs.mp.channel.domain.ChannelUserRel;
 import com.qs.mp.channel.service.IChannelUserRelService;
+import com.qs.mp.common.enums.MqTopicType;
 import com.qs.mp.common.enums.TicketBoxStatusEnum;
 import com.qs.mp.common.enums.TicketStatusEnum;
 import com.qs.mp.common.enums.TicketTypeEnum;
 import com.qs.mp.common.enums.UserCouponStatusEnum;
 import com.qs.mp.common.enums.UserTicketOrderStatusEnum;
 import com.qs.mp.common.exception.ServiceException;
+import com.qs.mp.common.pulsar.PulsarClientService;
 import com.qs.mp.common.utils.LogUtil;
 import com.qs.mp.common.utils.RSAUtil;
 import com.qs.mp.common.utils.StringUtils;
@@ -34,6 +36,8 @@ import com.qs.mp.user.service.IUserTicketOrderService;
 import java.util.Arrays;
 import java.util.Date;
 import java.util.List;
+import org.apache.pulsar.client.api.PulsarClient;
+import org.apache.pulsar.client.api.PulsarClientException;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.beans.factory.annotation.Autowired;
@@ -70,6 +74,9 @@ public class UserTicketOrderServiceImpl extends
   @Autowired
   private ITicketService ticketService;
 
+  @Autowired
+  private PulsarClientService pulsarClientService;
+
   @Autowired
   private BizIdGenerator bizIdGenerator;
 
@@ -245,6 +252,7 @@ public class UserTicketOrderServiceImpl extends
   }
 
   @Override
+  @Transactional
   public boolean paySuccess(PayOrder payOrder) {
     UserTicketOrder ticketOrder = getById(payOrder.getBizId());
     if (null == ticketOrder || ticketOrder.getStatus() != UserTicketOrderStatusEnum.NOT_PAY) {
@@ -279,6 +287,13 @@ public class UserTicketOrderServiceImpl extends
     ticketOrder.setStatus(UserTicketOrderStatusEnum.FINISHED);
     boolean updateRst = updateById(ticketOrder);
     Assert.isTrue(updateRst, "支付回调用户购票订单处理，更新订单状态失败，orderId:" + ticketOrder.getOrderId());
+
+    try {
+      pulsarClientService.producer(MqTopicType.ticket_pay, ticketOrder.getOrderId());
+    } catch (PulsarClientException e) {
+      LogUtil.error(logger, e, "盲票支付成功，发送异步消息失败. {0}", JSONObject.toJSONString(ticketOrder));
+    }
+
     return true;
   }
 }