package com.iformall.mq; import com.iformall.domain.po.msg.*; import com.iformall.enums.*; import com.iformall.mapper.WxMsgRecordMapper; import com.iformall.service.msg.impl.*; import com.iformall.utils.JsonUtil; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.stereotype.Service; /** * 消息消费入口 */ @Service public class MqBaseConsumer { private final Logger log = LoggerFactory.getLogger(this.getClass()); @Autowired private WxMsgRecordMapper wxMsgRecordMapper; @Autowired private SendSmsServiceImpl sendSmsService; @Autowired private SendCallBackSmsServiceImpl sendCallBackSmsService; @Autowired private SendEmailServiceImpl sendEmailService; @Autowired private SendSmartAppMsgServiceImpl sendSmartAppMsgService; @Autowired private SendMpMsgServiceImpl sendMpMsgService; @Autowired private SendWeappUniformMsgServiceImpl sendWeappUniformMsgService; @Autowired private AfterAddScoreMsgServiceImpl afterAddScoreMsgService; @Autowired private AfterCarInOutMsgServiceImpl afterCarInOutMsgService; @Autowired private FmInsideOrderSuccessMsgServiceImpl fmInsideOrderSuccessMsgService; @Autowired private FmInsideCouponVerifyMsgServiceImpl fmInsideCouponVerifyMsgService; @Autowired private FmInsideOrderRefundMsgServiceImpl fmInsideOrderRefundMsgService; // @Autowired // private FmInsideOrderPushMsgServiceImpl fmInsideOrderPushMsgService; @Autowired private FmInsideOrderPushDeliveryMsgServiceImpl fmInsideOrderPushDeliveryMsgServiceImpl; @Autowired private FmInsideCLoginMsgServiceImpl fmInsideCLoginMsgService; @Autowired private MqBaseProducer mqBaseProducer; //@Autowired //private FmInsideNotifyPaySuccessMsgServiceImpl fmInsideNotifyPaySuccessMsgService; //@Autowired //UpdateCouponStockMsgServiceImpl updateCouponStockMsgService; @Autowired private FmInsideCashOutMsgServiceImpl fmInsideCashOutMsgService; @Autowired private FmInsideProductPushMsgServiceImpl fmInsideProductPushMsgServiceImpl; @Autowired private FmInsideBasicUserPushMsgServiceImpl fmInsideBasicUserPushMsgServiceImpl; @Autowired private FmInsideBasicUserTenantMsgServiceImpl FmInsideBasicUserTenantMsgService; public void doMessage(String message) { log.info("received message: {}", message); //因为数据库读写同步时间可能在毫秒,此处暂停1秒后再开始处理 try { Thread.sleep(1000); } catch (InterruptedException e) { } BaseMsg baseMsg = null; try { baseMsg = (BaseMsg)JsonUtil.readValue(message,BaseMsg.class); if(EnumMsgRecordType.SMS.getCode().equals(baseMsg.getMsgType())){ //短信 WxMsgRecord msg = (WxMsgRecord)JsonUtil.readValue(message,WxMsgRecord.class); sendSmsService.send(msg); } else if(EnumMsgRecordType.SMS_CALLBACK.getCode().equals(baseMsg.getMsgType())){ //业务短信 WxMsg msg = (WxMsg)JsonUtil.readValue(message,WxMsg.class); sendCallBackSmsService.send(msg); } else if(EnumMsgRecordType.EMAIL.getCode().equals(baseMsg.getMsgType())){ //邮件 MailMsg msg = (MailMsg)JsonUtil.readValue(message,MailMsg.class); sendEmailService.send(msg); // } else if(EnumMsgRecordType.SMART_APP.getCode().equals(baseMsg.getMsgType())){ } else if(EnumMsgRecordType.SMART_APP_TO.getCode().equals(baseMsg.getMsgType())){ //微信小程序-订阅消息 SmartAppMsg msg = (SmartAppMsg)JsonUtil.readValue(message,SmartAppMsg.class); sendSmartAppMsgService.send(msg); } else if(EnumMsgRecordType.PUBLIC.getCode().equals(baseMsg.getMsgType())) { //公众号-模板消息 MpAppMsg msg = (MpAppMsg)JsonUtil.readValue(message,MpAppMsg.class); sendMpMsgService.send(msg); } else if(EnumMsgRecordType.SMART_APP_UNIFORM.getCode().equals(baseMsg.getMsgType())) { //微信小程序-统一消息 AppUniformMsg msg = (AppUniformMsg)JsonUtil.readValue(message,AppUniformMsg.class); sendWeappUniformMsgService.send(msg); } else if(EnumMsgRecordType.AFTER_ADD_SCORE.getCode().equals(baseMsg.getMsgType())) { //成长值变更后 AfterAddScoreMsg msg = (AfterAddScoreMsg)JsonUtil.readValue(message,AfterAddScoreMsg.class); afterAddScoreMsgService.send(msg); } else if(EnumMsgRecordType.AFTER_CAR_IN_OR_OUT.getCode().equals(baseMsg.getMsgType())) { //车辆入场,出场后 AfterCarInOutMsg msg = (AfterCarInOutMsg)JsonUtil.readValue(message,AfterCarInOutMsg.class); afterCarInOutMsgService.send(msg); } else if(EnumMsgRecordType.INSIDE_ORDER_SUCCESS.getCode().equals(baseMsg.getMsgType())) { // 内部消息 - 下订单成功 FmInsideOrderSuccessMsg msg = (FmInsideOrderSuccessMsg)JsonUtil.readValue(message,FmInsideOrderSuccessMsg.class); fmInsideOrderSuccessMsgService.send(msg); } else if(EnumMsgRecordType.INSIDE_COUPON_VERIFY.getCode().equals(baseMsg.getMsgType())) { // 内部消息 - 券核销 FmInsideCouponVerifyMsg msg = (FmInsideCouponVerifyMsg)JsonUtil.readValue(message,FmInsideCouponVerifyMsg.class); fmInsideCouponVerifyMsgService.send(msg); } else if(EnumMsgRecordType.INSIDE_ORDER_REFUND.getCode().equals(baseMsg.getMsgType())){ //内部消息 - 退款申请成功 FmInsideOrderRefundMsg msg = (FmInsideOrderRefundMsg) JsonUtil.readValue(message,FmInsideOrderRefundMsg.class); fmInsideOrderRefundMsgService.send(msg); } else if(EnumMsgRecordType.INSIDE_C_LOGIN.getCode().equals(baseMsg.getMsgType())) { // 内部消息 - c端登录 FmInsideCLoginMsg msg = (FmInsideCLoginMsg)JsonUtil.readValue(message,FmInsideCLoginMsg.class); fmInsideCLoginMsgService.send(msg); } else if(EnumMsgRecordType.INSIDE_ORDER_PUSH.getCode().equals(baseMsg.getMsgType())) { // 内部消息 - 订单同步 // FmInsideOrderPushMsg msg = (FmInsideOrderPushMsg)JsonUtil.readValue(message,FmInsideOrderPushMsg.class); // fmInsideOrderPushMsgService.send(msg); } else if(EnumMsgRecordType.INSIDE_ORDER_PUSH_DELIVERY.getCode().equals(baseMsg.getMsgType())) { // 内部消息 - 2.0核销同步 FmInsideOrderPushMsg msg = (FmInsideOrderPushMsg)JsonUtil.readValue(message,FmInsideOrderPushMsg.class); fmInsideOrderPushDeliveryMsgServiceImpl.send(msg); } //else if(EnumMsgRecordType.INSIDE_NOTIFY_PAY_SUCCESS.getCode().equals(baseMsg.getMsgType())) { // 内部消息 - 微信支付通知 //FmInsideNotifyPaySuccessMsg msg = (FmInsideNotifyPaySuccessMsg)JsonUtil.readValue(message,FmInsideNotifyPaySuccessMsg.class); //fmInsideNotifyPaySuccessMsgService.send(msg); //} else if(EnumMsgRecordType.INSIDE_NOTIFY_REFUND_SUCCESS.getCode().equals(baseMsg.getMsgType())) { // 内部消息 - 微信退款通知 } else if(EnumMsgRecordType.CASH_OUT.getCode().equals(baseMsg.getMsgType())) { // 内部消息 - 商户提现通知 FmInsideCashOutMsg msg = (FmInsideCashOutMsg)JsonUtil.readValue(message,FmInsideCashOutMsg.class); fmInsideCashOutMsgService.send(msg); } //else if (EnumMsgRecordType.COUPON_STOCK.getCode().equals(baseMsg.getMsgType())) { // UpdateCouponStockMsg msg = (UpdateCouponStockMsg)JsonUtil.readValue(message,UpdateCouponStockMsg.class); // updateCouponStockMsgService.send(msg); //} //baseMsg.setMsgStatus(EnumMsgRecordStatus.CONSUME_SUCC.getCode()); //wxMsgRecordMapper.updateByBaseMsg(baseMsg); else if(EnumMsgRecordType.INSIDE_PRODUCT_PUSH.getCode().equals(baseMsg.getMsgType())) { // 内部消息 - 同步POI FmInsideProductPushMsg msg = (FmInsideProductPushMsg)JsonUtil.readValue(message,FmInsideProductPushMsg.class); fmInsideProductPushMsgServiceImpl.send(msg); } else if(EnumMsgRecordType.MALL_COO_USER_PUSH.getCode().equals(baseMsg.getMsgType())) { // 内部消息 - 同步会员 FmInsideBasicUserPushMsg msg = (FmInsideBasicUserPushMsg)JsonUtil.readValue(message,FmInsideBasicUserPushMsg.class); fmInsideBasicUserPushMsgServiceImpl.send(msg); } else if(EnumMsgRecordType.MEMBER_UPDATE_TENANT.getCode().equals(baseMsg.getMsgType())) { // 内部消息 - 集团会员广场标签 FmInsideBasicUserPushMsg msg = (FmInsideBasicUserPushMsg)JsonUtil.readValue(message,FmInsideBasicUserPushMsg.class); FmInsideBasicUserTenantMsgService.send(msg); } }catch (Throwable e){ log.error("consum received error: ", e); /** try { log.error("consum received error: ", e); baseMsg.setStatusMessage(e.getMessage()); baseMsg.setMsgStatus(EnumMsgRecordStatus.CONSUME_FAIL.getCode()); wxMsgRecordMapper.updateByBaseMsg(baseMsg); if(null != baseMsg){ if(baseMsg.getDelayTimeLevel() > 0){ log.info("消息重试,DelayTimeLevel: "+baseMsg.getDelayTimeLevel()); Thread.sleep(5000); baseMsg.setDelayTimeLevel(baseMsg.getDelayTimeLevel()-1); if(EnumMsgRecordType.INSIDE_NOTIFY_REFUND_SUCCESS.getCode().equals(baseMsg.getMsgType())) { // 内部消息 - 微信退款通知 FmInsideNotifyRefundSuccessMsg msg = (FmInsideNotifyRefundSuccessMsg)JsonUtil.readValue(message,FmInsideNotifyRefundSuccessMsg.class); mqBaseProducer.sendMessage(msg, EnumMsgMqTopic.DEFAULT.getCode(), EnumMsgMqTag.DEFAULT.getCode(), EnumMsgMqKey.DEFAULT.getCode()); } }else{ baseMsg.setStatusMessage("重试次数超过3次"); baseMsg.setMsgStatus(EnumMsgRecordStatus.CONSUME_FAIL.getCode()); wxMsgRecordMapper.updateByBaseMsg(baseMsg); } } }catch (Throwable e1){ log.error("更新db状态error: ", e1); }*/ } } }