diff --git a/mallinkMQConsumer/src/main/java/com/iformall/mq/impl/RocketMqConsumer.java b/mallinkMQConsumer/src/main/java/com/iformall/mq/impl/RocketMqConsumer.java index 4d4e35442..a100c8dd6 100644 --- a/mallinkMQConsumer/src/main/java/com/iformall/mq/impl/RocketMqConsumer.java +++ b/mallinkMQConsumer/src/main/java/com/iformall/mq/impl/RocketMqConsumer.java @@ -1,31 +1,44 @@ package com.iformall.mq; +import com.alibaba.druid.support.json.JSONUtils; +import com.iformall.domain.po.WxMsgRecord; +import com.iformall.enums.EnumMsgRecordStatus; +import com.iformall.mapper.WxMsgRecordMapper; import lombok.extern.slf4j.Slf4j; -import org.apache.rocketmq.common.message.MessageExt; -import org.rocketmq.starter.annotation.RocketMQListener; -import org.rocketmq.starter.annotation.RocketMQMessage; +import org.apache.rocketmq.spring.annotation.RocketMQMessageListener; +import org.apache.rocketmq.spring.core.RocketMQListener; import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import org.springframework.beans.factory.annotation.Autowired; import org.springframework.stereotype.Service; @Slf4j @Service -@RocketMQListener(topic = "topic-1") -public class MqConsumer { +@RocketMQMessageListener(topic = "topic-1",consumerGroup = "DefaultConsumer") +public class MqConsumer implements RocketMQListener{ private final Logger log = LoggerFactory.getLogger(this.getClass()); + @Autowired + private WxMsgRecordMapper wxMsgRecordMapper; - @RocketMQMessage(messageClass = MessageExt.class, tag = "tag-1") - public void onMessage(MessageExt message) { + public void onMessage(String message) { + WxMsgRecord wxMsgRecord = null; try { log.info("received message: {}", message); + wxMsgRecord = (WxMsgRecord)JSONUtils.parse(message); -// System.out.println("------"+message.getProperties().get("PROPERTY_RECONSUME_TIME")); -// System.out.println(1/0); -// return ConsumeConcurrentlyStatus.CONSUME_SUCCESS; + wxMsgRecord.setStatus(EnumMsgRecordStatus.CONSUME_SUCC.getCode()); + wxMsgRecordMapper.update(wxMsgRecord); }catch (Throwable e){ e.printStackTrace(); + if(null != wxMsgRecord){ + wxMsgRecord.setStatusMessage(e.getMessage()); + wxMsgRecord.setStatus(EnumMsgRecordStatus.CONSUME_FAIL.getCode()); + wxMsgRecordMapper.update(wxMsgRecord); + } } -// return ConsumeConcurrentlyStatus.RECONSUME_LATER; } + + + } diff --git a/mallinkService/src/main/java/com/iformall/domain/po/WxMsgRecord.java b/mallinkService/src/main/java/com/iformall/domain/po/WxMsgRecord.java index 908bcd293..e019cc5d5 100644 --- a/mallinkService/src/main/java/com/iformall/domain/po/WxMsgRecord.java +++ b/mallinkService/src/main/java/com/iformall/domain/po/WxMsgRecord.java @@ -2,189 +2,225 @@ package com.iformall.domain.po; import javax.persistence.Id; import javax.persistence.Table; +import javax.persistence.Transient; +import java.io.Serializable; import java.util.Date; -import java.util.Map; +import java.util.List; @Table(name = "wx_msg_record") -public class WxMsgRecord extends BaseMsg { +public class WxMsgRecord implements Serializable { private static final long serialVersionUID = 18957895653478L; @Id protected Long id; + + @Transient + protected List ids; + @Transient + protected String sortColumns; + + public Long getId() { + return id; + } + + public void setId(Long id) { + this.id = id; + } + + public String getSortColumns() { + return sortColumns; + } + + public List getIds() { + return ids; + } + + public void setIds(List ids) { + this.ids = ids; + } + + //uuid + private String uuid; //租户id private String tenantId; //域1sys 2a端 3c端 4b端 - private Integer domain; - //msg 消息类型1短信 2回调短信 邮件 3微信小程序模板 4微信公众号模板 5系统通 + private String group; + //type + private Integer type; + //msg private String msg; //发送时间 private String sendTime; //model_id private Long modelId; //from,手机号邮件地址等 - private String sender; + private String from; //发送人id - private Long senderUserId; + private Long fromUserId; //发送人姓名 - private String senderUserName; + private String fromUserName; //接收人,手机号邮件地址等 - private String receiver; + private String to; //接收人用户id - private Long receiverUserId; + private Long toUserId; //接收人姓名 - private String receiverUserName; + private String toUserName; + //消费结果 + private Integer status; + //消费结果信息 + private String statusMessage; //签名 private String signature; //模板类型 private Integer modelType; - //消息json - private String msgJson; + //动态内容(map结构的json数据) + private String dynamicContent; private Date createtime; private Date updatetime; - //动态替换内容 - private Map dynamicContentMap; - private String dynamicContent; - - - public String getMsgJson() { - return msgJson; + public String getDynamicContent() { + return dynamicContent; } - public void setMsgJson(String msgJson) { - this.msgJson = msgJson; + public void setDynamicContent(String dynamicContent) { + this.dynamicContent = dynamicContent; } - public Map getDynamicContentMap() { - return dynamicContentMap; + public Integer getModelType() { + return modelType; } - - public void setDynamicContentMap(Map dynamicContentMap) { - this.dynamicContentMap = dynamicContentMap; + public void setModelType(Integer modelType) { + this.modelType = modelType; } - public String getDynamicContent() { - return dynamicContent; + public String getSignature() { + return signature; } - public void setDynamicContent(String dynamicContent) { - this.dynamicContent = dynamicContent; + public void setSignature(String signature) { + this.signature = signature; } - public Long getId() { - return id; + public String getUuid() { + return uuid; } - public void setId(Long id) { - this.id = id; + public void setUuid(String uuid) { + this.uuid = uuid; } - public Integer getDomain() { - return domain; + public String getTenantId() { + return tenantId; } - public void setDomain(Integer domain) { - this.domain = domain; + public void setTenantId(String tenantId) { + this.tenantId = tenantId; } - public String getSender() { - return sender; + public String getGroup() { + return group; } - public void setSender(String sender) { - this.sender = sender; + public void setGroup(String group) { + this.group = group; } - public Long getSenderUserId() { - return senderUserId; + public Integer getType() { + return type; } - public void setSenderUserId(Long senderUserId) { - this.senderUserId = senderUserId; + public void setType(Integer type) { + this.type = type; } - public String getSenderUserName() { - return senderUserName; + public String getMsg() { + return msg; } - public void setSenderUserName(String senderUserName) { - this.senderUserName = senderUserName; + public void setMsg(String msg) { + this.msg = msg; } - public String getReceiver() { - return receiver; + public String getSendTime() { + return sendTime; } - public void setReceiver(String receiver) { - this.receiver = receiver; + public void setSendTime(String sendTime) { + this.sendTime = sendTime; } - public Long getReceiverUserId() { - return receiverUserId; + public Long getModelId() { + return modelId; } - public void setReceiverUserId(Long receiverUserId) { - this.receiverUserId = receiverUserId; + public void setModelId(Long modelId) { + this.modelId = modelId; } - public String getReceiverUserName() { - return receiverUserName; + public String getFrom() { + return from; } - public void setReceiverUserName(String receiverUserName) { - this.receiverUserName = receiverUserName; + public void setFrom(String from) { + this.from = from; } - public Integer getModelType() { - return modelType; + public Long getFromUserId() { + return fromUserId; } - public void setModelType(Integer modelType) { - this.modelType = modelType; + public void setFromUserId(Long fromUserId) { + this.fromUserId = fromUserId; } - public String getSignature() { - return signature; + public String getFromUserName() { + return fromUserName; } - public void setSignature(String signature) { - this.signature = signature; + public void setFromUserName(String fromUserName) { + this.fromUserName = fromUserName; } + public String getTo() { + return to; + } - public String getTenantId() { - return tenantId; + public void setTo(String to) { + this.to = to; } - public void setTenantId(String tenantId) { - this.tenantId = tenantId; + public Long getToUserId() { + return toUserId; } + public void setToUserId(Long toUserId) { + this.toUserId = toUserId; + } - public String getMsg() { - return msg; + public String getToUserName() { + return toUserName; } - public void setMsg(String msg) { - this.msg = msg; + public void setToUserName(String toUserName) { + this.toUserName = toUserName; } - public String getSendTime() { - return sendTime; + public Integer getStatus() { + return status; } - public void setSendTime(String sendTime) { - this.sendTime = sendTime; + public void setStatus(Integer status) { + this.status = status; } - public Long getModelId() { - return modelId; + public String getStatusMessage() { + return statusMessage; } - public void setModelId(Long modelId) { - this.modelId = modelId; + public void setStatusMessage(String statusMessage) { + this.statusMessage = statusMessage; } public Date getCreatetime() { @@ -203,5 +239,62 @@ public class WxMsgRecord extends BaseMsg { this.updatetime = updatetime; } - + public static enum Field + { + Id_ASC("`id` ASC"),Id_DESC("`id` DESC") + ,TenantId_ASC("`tenantId` ASC"),TenantId_DESC("`tenantId` DESC") + ,Name_ASC("`name` ASC"),Name_DESC("`name` DESC") + ,Available_ASC("`available` ASC"),Available_DESC("`available` DESC") + ,Description_ASC("`description` ASC"),Description_DESC("`description` DESC") + ; + private String value; + Field(String value){ + this.value = value; + } + public String getValue() { + return value; + } + public void setCol(String value) { + this.value = value; + } + @Override + public String toString() { + return this.getValue(); + } + } + + public void setSortColumns(WxMsgRecord.Field... fields) + { + if (fields == null || fields.length == 0) { + return; + } + for (int k = 0; k < fields.length; k++) { + if (fields[k] == null) { + return; + } + } + StringBuilder sb = new StringBuilder(fields[0].toString()); + for (int k = 1; k < fields.length; k++) { + sb.append(","); + sb.append(fields[k].toString()); + } + + } + + public void setSortColumns(String sortColumns) + { + if (sortColumns == null || "".equals(sortColumns.trim())) { + return; + } + if (sortColumns.contains(",")) { + String[] cols = sortColumns.split(","); + List fList = new java.util.ArrayList(); + for (int k = 0; k < cols.length; k++) { + fList.add(Field.valueOf(cols[k])); + } + this.setSortColumns(fList.toArray(new Field[fList.size()])); + } else { + this.setSortColumns(Field.valueOf(sortColumns)); + } + } } diff --git a/mallinkService/src/main/java/com/iformall/mapper/WxMsgRecordMapper.java b/mallinkService/src/main/java/com/iformall/mapper/WxMsgRecordMapper.java index 6fbe62f0b..dc88cb555 100644 --- a/mallinkService/src/main/java/com/iformall/mapper/WxMsgRecordMapper.java +++ b/mallinkService/src/main/java/com/iformall/mapper/WxMsgRecordMapper.java @@ -1,12 +1,16 @@ package com.iformall.mapper; import com.iformall.common.CommonMapper; -import com.iformall.domain.po.BaseMsg; +import com.iformall.domain.po.MallRole; import com.iformall.domain.po.WxMsgRecord; +import org.apache.ibatis.annotations.Param; + +import java.util.List; +import java.util.Map; public interface WxMsgRecordMapper extends CommonMapper { - void update(BaseMsg record); + void update(WxMsgRecord record); diff --git a/mallinkService/src/main/java/com/iformall/mq/impl/RocketMqMessageProducer.java b/mallinkService/src/main/java/com/iformall/mq/impl/RocketMqMessageProducer.java index bb8c68801..16cfd94f0 100644 --- a/mallinkService/src/main/java/com/iformall/mq/impl/RocketMqMessageProducer.java +++ b/mallinkService/src/main/java/com/iformall/mq/impl/RocketMqMessageProducer.java @@ -1,5 +1,9 @@ package com.iformall.mq; +import com.alibaba.druid.support.json.JSONUtils; +import com.iformall.domain.po.WxMsgRecord; +import com.iformall.enums.EnumMsgRecordStatus; +import com.iformall.mapper.WxMsgRecordMapper; import lombok.extern.slf4j.Slf4j; import org.apache.rocketmq.client.producer.DefaultMQProducer; import org.apache.rocketmq.client.producer.SendCallback; @@ -8,10 +12,11 @@ import org.apache.rocketmq.common.message.Message; import org.apache.rocketmq.remoting.common.RemotingHelper; import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Value; import org.springframework.stereotype.Service; - import javax.annotation.PostConstruct; +import java.util.UUID; /** * @Auther: luozukai @@ -25,6 +30,8 @@ public class MqProducer { private static final DefaultMQProducer producer = new DefaultMQProducer("DefaultProducer"); @Value("${spring.rocketmq.nameServer}") private String namesrvAddr; + @Autowired + private WxMsgRecordMapper wxMsgRecordMapper; @PostConstruct public void start(){ @@ -39,23 +46,30 @@ public class MqProducer { } } - public void sendMessage(String data, String topic, String tags, String keys) throws Throwable{ + public void sendMessage(WxMsgRecord data, String topic, String tags, String keys) throws Throwable{ try { - byte[] messageBody = data.getBytes(RemotingHelper.DEFAULT_CHARSET); + data.setUuid(UUID.randomUUID().toString()); + wxMsgRecordMapper.insert(data); + + byte[] messageBody = JSONUtils.toJSONString(data).getBytes(RemotingHelper.DEFAULT_CHARSET); Message mqMsg = new Message(topic, tags, keys, messageBody); producer.send(mqMsg, new SendCallback() { @Override public void onSuccess(SendResult sendResult) { log.info("Message Producer: Send Message Success {}", sendResult); + + data.setStatus(EnumMsgRecordStatus.SEND_SUCC.getCode()); + wxMsgRecordMapper.update(data); } @Override public void onException(Throwable throwable) { - log.error("Message Producer: Send Message Error ", throwable); + log.error("Message Producer: Send Message Error (onException)", throwable); } }); } catch (Throwable e) { e.printStackTrace(); + log.error("Message Producer: Send Message Error (sendMessage)", e); } } } diff --git a/mallinkService/src/main/java/com/iformall/service/MsgSendService.java b/mallinkService/src/main/java/com/iformall/service/MsgSendService.java index 264ef6088..07c69453f 100644 --- a/mallinkService/src/main/java/com/iformall/service/MsgSendService.java +++ b/mallinkService/src/main/java/com/iformall/service/MsgSendService.java @@ -1,10 +1,12 @@ package com.iformall.service; -import com.iformall.domain.po.BaseMsg; +import com.iformall.domain.po.WxMsgRecord; -/** - * 发送消息 - */ public interface MsgSendService { - void send(BaseMsg baseMsg) throws Exception; + + void send(WxMsgRecord record); + + + + } diff --git a/mallinkService/src/main/java/com/iformall/service/impl/SendSmsServiceImpl.java b/mallinkService/src/main/java/com/iformall/service/impl/SendSmsServiceImpl.java index 2b120b0d3..b435f3563 100644 --- a/mallinkService/src/main/java/com/iformall/service/impl/SendSmsServiceImpl.java +++ b/mallinkService/src/main/java/com/iformall/service/impl/SendSmsServiceImpl.java @@ -1,25 +1,8 @@ package com.iformall.service.impl; -import com.alibaba.fastjson.JSONObject; -import com.iformall.common.ErrorCode; -import com.iformall.common.IdWorker; -import com.iformall.domain.po.*; -import com.iformall.enums.EnumVerifyCode; -import com.iformall.exception.MallinkException; -import com.iformall.mapper.WxMsgConfigMapper; -import com.iformall.mapper.WxMsgValidationcodeMapper; -import com.iformall.mapper.WxMsgValidationcodeModelMapper; +import com.iformall.domain.po.WxMsgRecord; import com.iformall.service.MsgSendService; -import com.iformall.service.WxMsgValidationcodeService; -import com.iformall.utils.WiwideUtil; -import org.apache.commons.collections.CollectionUtils; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; -import org.springframework.beans.factory.annotation.Autowired; import org.springframework.stereotype.Service; -import java.util.Date; -import java.util.List; -import java.util.Map; /** * @author luozukai @@ -27,82 +10,9 @@ import java.util.Map; */ @Service public class SendSmsServiceImpl implements MsgSendService { - private final Logger logger = LoggerFactory.getLogger(this.getClass()); - @Autowired - private WxMsgConfigMapper wxMsgConfigMapper; - @Autowired - private WxMsgValidationcodeModelMapper wxMsgValidationcodeModelMapper; - @Autowired - private WxMsgValidationcodeMapper wxMsgValidationcodeMapper; - @Autowired - private WxMsgValidationcodeService wxMsgValidationcodeService; @Override - public void send(BaseMsg baseMsg) { - WxMsgRecord record = (WxMsgRecord)baseMsg; - WxMsgValidationcode wxMsgValidationcode = new WxMsgValidationcode(); - wxMsgValidationcode.setPhone(record.getReceiver()); - wxMsgValidationcode.setTenantId(record.getTenantId()); - wxMsgValidationcode.setType(record.getModelType()); + public void send(WxMsgRecord record) { - //3、从短信配置中查询密钥 bid 等信息 - WxMsgConfig wxMsgConfig = new WxMsgConfig(); - wxMsgConfig.setTenantId(record.getTenantId()); - List wxMsgConfigs = wxMsgConfigMapper.findList(wxMsgConfig); - if(CollectionUtils.isEmpty(wxMsgConfigs)){ - throw new MallinkException(ErrorCode.MSG_CONFIG_NOT_FOUND); - } - - wxMsgConfig = wxMsgConfigs.get(0); - - WxMsgValidationcodeModel wxMsgValidationcodeModel = new WxMsgValidationcodeModel(); - wxMsgValidationcodeModel.setTenantId(record.getTenantId()); - wxMsgValidationcodeModel.setType(record.getModelType()); - List modelList = wxMsgValidationcodeModelMapper.findList(wxMsgValidationcodeModel); - if(CollectionUtils.isEmpty(modelList)){ - throw new MallinkException(ErrorCode.MSG_MODEL_NOT_FOUND); - } - - wxMsgValidationcodeModel = wxMsgValidationcodeModelMapper.findList(wxMsgValidationcodeModel).get(0); - - String secret = wxMsgConfig.getSecret(); - String bid = wxMsgConfig.getBid(); - String publickey = wxMsgConfig.getPublickey(); - String phone = record.getReceiver(); - String signature = wxMsgValidationcodeModel.getSignature(); - wxMsgValidationcode.setSignature(signature); - //替换内容 - String msg = wxMsgValidationcodeModel.getContent(); - for (Map.Entry entry : record.getDynamicContentMap().entrySet()) { - msg = msg.replace("{"+entry.getKey()+"}", entry.getValue()); - } - wxMsgValidationcode.setMsg(msg); - - String notifyUrl = wxMsgConfig.getNotifyurl(); - Integer modelId = wxMsgValidationcodeModel.getModelId(); - String result = WiwideUtil.sendMsg(secret, bid, publickey, phone, signature, msg, notifyUrl, EnumVerifyCode.YES.getCode().toString(), modelId); - JSONObject jsonObjectResult = JSONObject.parseObject(result); - String ret = jsonObjectResult.get("ret").toString(); - String batchNo = jsonObjectResult.get("data").toString(); - if (ret.equals("1")) { - final IdWorker idWorker = IdWorker.get(); - wxMsgValidationcode.setId(idWorker.nextId()); - long currentTime = System.currentTimeMillis() ; - Date createtime=new Date(currentTime); - wxMsgValidationcode.setCreatetime(createtime); - Integer minutes = wxMsgValidationcodeModel.getMinutes(); - if(minutes >0) { - currentTime += minutes * 60 * 1000; - Date expiredate = new Date(currentTime); - wxMsgValidationcode.setExpiretime(expiredate); - } - wxMsgValidationcodeMapper.insertSelective(wxMsgValidationcode); - wxMsgValidationcodeService.addMsgCallback(wxMsgValidationcode,batchNo); - logger.error("短信运营商返回成功,手机号:{}",phone); - }else{ - logger.error("短信运营商返回失败,手机号:{},返回信息:{}",phone,result); - } } - - } diff --git a/mallinkService/src/main/resources/mapper/WxMsgRecordMapper.xml b/mallinkService/src/main/resources/mapper/WxMsgRecordMapper.xml index cb7f821b3..132cd4069 100644 --- a/mallinkService/src/main/resources/mapper/WxMsgRecordMapper.xml +++ b/mallinkService/src/main/resources/mapper/WxMsgRecordMapper.xml @@ -4,18 +4,18 @@ - - + + - - - - - - - + + + + + + + @@ -23,10 +23,14 @@ - + + + `id`,`tenant_id`,`group`,`type`,`msg` ,send_time,model_id,from,from_user_id,from_user_name,to,to_user_id,to_user_name + ,status,status_message,createtime,updatetime,uuid,signature,model_type,dynamic_content + where 1 = 1 @@ -49,10 +53,10 @@ order by ${sortColumns} - + update wx_msg_record set updatetime = now() - ,msg_status = #{msgStatus} - ,status_message = #{statusMessage} + ,status = #{status} + ,message = #{message} where uuid = #{uuid}