Browse Source

[消息组件][添加][init]

release_toaliyun_real
luozukai 7 years ago
parent
commit
d67575193c
7 changed files with 254 additions and 214 deletions
  1. +24
    -11
      mallinkMQConsumer/src/main/java/com/iformall/mq/impl/RocketMqConsumer.java
  2. +180
    -87
      mallinkService/src/main/java/com/iformall/domain/po/WxMsgRecord.java
  3. +6
    -2
      mallinkService/src/main/java/com/iformall/mapper/WxMsgRecordMapper.java
  4. +18
    -4
      mallinkService/src/main/java/com/iformall/mq/impl/RocketMqMessageProducer.java
  5. +7
    -5
      mallinkService/src/main/java/com/iformall/service/MsgSendService.java
  6. +2
    -92
      mallinkService/src/main/java/com/iformall/service/impl/SendSmsServiceImpl.java
  7. +17
    -13
      mallinkService/src/main/resources/mapper/WxMsgRecordMapper.xml

+ 24
- 11
mallinkMQConsumer/src/main/java/com/iformall/mq/impl/RocketMqConsumer.java View File

@@ -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<String>{
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;
}



}

+ 180
- 87
mallinkService/src/main/java/com/iformall/domain/po/WxMsgRecord.java View File

@@ -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<Long> 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<Long> getIds() {
return ids;
}

public void setIds(List<Long> 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<String,String> 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<String, String> getDynamicContentMap() {
return dynamicContentMap;
public Integer getModelType() {
return modelType;
}


public void setDynamicContentMap(Map<String, String> 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<Field> 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));
}
}
}

+ 6
- 2
mallinkService/src/main/java/com/iformall/mapper/WxMsgRecordMapper.java View File

@@ -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<WxMsgRecord, String> {

void update(BaseMsg record);
void update(WxMsgRecord record);


+ 18
- 4
mallinkService/src/main/java/com/iformall/mq/impl/RocketMqMessageProducer.java View File

@@ -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);
}
}
}

+ 7
- 5
mallinkService/src/main/java/com/iformall/service/MsgSendService.java View File

@@ -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);


}

+ 2
- 92
mallinkService/src/main/java/com/iformall/service/impl/SendSmsServiceImpl.java View File

@@ -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<WxMsgConfig> 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<WxMsgValidationcodeModel> 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<String, String> 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);
}
}


}

+ 17
- 13
mallinkService/src/main/resources/mapper/WxMsgRecordMapper.xml View File

@@ -4,18 +4,18 @@
<resultMap id="BaseResultMap" type="com.iformall.domain.po.WxMsgRecord">
<id column="id" property="id" />
<result column="tenant_id" property="tenantId" />
<result column="domain" property="domain" />
<result column="msg_type" property="msgType" />
<result column="group" property="group" />
<result column="type" property="type" />
<result column="msg" property="msg" />
<result column="send_time" property="sendTime" />
<result column="model_id" property="modelId" />
<result column="sender" property="from" />
<result column="sender_user_id" property="fromUserId" />
<result column="sender_user_name" property="fromUserName" />
<result column="receiver" property="to" />
<result column="receiver_user_id" property="toUserId" />
<result column="receiver_user_name" property="toUserName" />
<result column="msg_status" property="msgStatus" />
<result column="from" property="from" />
<result column="from_user_id" property="fromUserId" />
<result column="from_user_name" property="fromUserName" />
<result column="to" property="to" />
<result column="to_user_id" property="toUserId" />
<result column="to_user_name" property="toUserName" />
<result column="status" property="status" />
<result column="status_message" property="statusMessage" />
<result column="createtime" property="createtime" />
<result column="updatetime" property="updatetime" />
@@ -23,10 +23,14 @@
<result column="signature" property="signature" />
<result column="model_type" property="modelType" />
<result column="dynamic_content" property="dynamicContent" />
<result column="msg_json" property="msgJson" />


</resultMap>
<sql id="allColumns">
`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
</sql>

<sql id="dynamicWhereConditions">
where 1 = 1
@@ -49,10 +53,10 @@
<if test=" null != sortColumns"> order by ${sortColumns} </if>
</sql>

<update id="update" parameterType="com.iformall.domain.po.BaseMsg">
<update id="update" parameterType="com.iformall.domain.po.WxMsgRecord">
update wx_msg_record set updatetime = now()
<if test=" null != msgStatus ">,msg_status = #{msgStatus}</if>
<if test=" null != statusMessage ">,status_message = #{statusMessage}</if>
<if test=" null != status ">,status = #{status}</if>
<if test=" null != message ">,message = #{message}</if>
where uuid = #{uuid}
</update>



Loading…
Cancel
Save