mq改造

themis-backend/src/main/java/com/qmth/themis/backend/api/TBUserController.java

@@ -7,8 +7,10 @@ import com.qmth.themis.backend.quartz.MqJob;
 import com.qmth.themis.backend.util.ServletUtil;
 import com.qmth.themis.business.constant.SystemConstant;
 import com.qmth.themis.business.dto.AuthDto;
+import com.qmth.themis.business.dto.MqDto;
 import com.qmth.themis.business.entity.TBSession;
 import com.qmth.themis.business.entity.TBUser;
+import com.qmth.themis.business.enums.MqEnum;
 import com.qmth.themis.business.service.EhcacheService;
 import com.qmth.themis.business.service.ProducerServer;
 import com.qmth.themis.business.service.QuartzService;
@@ -28,7 +30,6 @@ import com.qmth.themis.common.util.Result;
 import com.qmth.themis.common.util.ResultUtil;
 import io.swagger.annotations.*;
 import org.apache.commons.lang3.RandomStringUtils;
-import org.assertj.core.util.Maps;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.web.bind.annotation.RequestBody;
@@ -122,12 +123,25 @@ public class TBUserController {
         TBSession tbSession = new TBSession(sessionId, String.valueOf(user.getId()), authDto.getRoleEnum().name(), platform.getSource(), platform.name(), deviceId, request.getLocalAddr(), token, expire);
         redisUtil.setUserSession(sessionId, tbSession);
         //往mq发送消息插入会话信息
+        MqDto mqDto = null;
         try {
-            producerServer.sendOneWay(SystemConstant.SESSION_TOPIC, platform.getSource(), tbSession.getId());
-            tbSession.setAck(SystemConstant.DELIVERED_ACK_TYPE);
-            redisUtil.setSessionTopicList(sessionId, tbSession);
+//            for (int i = 0; i < 100; i++) {
+//                TBSession tbSession = new TBSession(sessionId, String.valueOf(user.getId()), authDto.getRoleEnum().name(), platform.getSource(), platform.name(), deviceId, request.getLocalAddr(), token, expire);
+//                redisUtil.setUserSession(sessionId, tbSession);
+//                int random = (int) (Math.random() * Source.values().length);
+            mqDto = new MqDto(SystemConstant.SESSION_TOPIC, platform.getSource(), tbSession, MqEnum.SESSION, tbSession.getId());
+//                mqDto = new MqDto(SystemConstant.SESSION_TOPIC, Source.values()[random].name(), tbSession, MqEnum.SESSION, tbSession.getId());
+//                mqDto.setSequence(i);
+            mqDto.setAck(SystemConstant.UNSEND_ACK_TYPE);
+            producerServer.sendOneWay(mqDto);
+            mqDto.setAck(SystemConstant.DELIVERED_ACK_TYPE);
+            redisUtil.setSessionTopicList(mqDto.getId(), mqDto);
+//            }
         } catch (Exception e) {
             e.printStackTrace();
+            if (Objects.nonNull(mqDto)) {
+                mqDto.setAck(SystemConstant.UNSEND_ACK_TYPE);
+            }
         }
         //测试
         String test = SignatureInfo.build(SignatureType.TOKEN, sessionId, token);

themis-backend/src/main/java/com/qmth/themis/backend/interceptor/AuthInterceptor.java

@@ -76,7 +76,7 @@ public class AuthInterceptor implements HandlerInterceptor {
         }
         Long userId = null;
 //        Long timestamp = Long.parseLong(ServletUtil.getRequestTime(request));
-//        if (!expire(timestamp.longValue())) {
+//        if (!SystemConstant.expire(timestamp.longValue())) {
 //            final SignatureInfo info = SignatureInfo
 //                    .parse(method, url, timestamp, ServletUtil.getRequestAuthorization(request));
         //测试
@@ -151,9 +151,4 @@ public class AuthInterceptor implements HandlerInterceptor {
                                 HttpServletResponse response,
                                 Object o, Exception e) throws Exception {
     }
-
-    public boolean expire(long timestamp) {
-        long diff = (System.currentTimeMillis() - timestamp) / 1000;
-        return diff < -1 * Constants.SIGNATURE_AHEAD_SECONDS || diff > Constants.SIGNATURE_EXPIRE_SECONDS;
-    }
 }

themis-backend/src/main/java/com/qmth/themis/backend/quartz/MqJob.java

@@ -1,7 +1,7 @@
 package com.qmth.themis.backend.quartz;
 
 import com.qmth.themis.business.constant.SystemConstant;
-import com.qmth.themis.business.entity.TBSession;
+import com.qmth.themis.business.dto.MqDto;
 import com.qmth.themis.business.service.ProducerServer;
 import com.qmth.themis.business.util.RedisUtil;
 import org.quartz.JobExecutionContext;
@@ -37,8 +37,8 @@ public class MqJob extends QuartzJobBean {
             log.info("session_topic有异常的消息数为:{}", size);
             Map map = redisUtil.getHashEntries(SystemConstant.SESSION_TOPIC_ERROR_LIST);
             map.forEach((k, v) -> {
-                TBSession tbSession = (TBSession) v;
-                producerServer.sendOneWay(SystemConstant.SESSION_TOPIC, tbSession.getSource(), v);
+                MqDto mqDto = (MqDto) v;
+                producerServer.sendOneWay(mqDto);
             });
         }
     }

themis-backend/src/main/resources/application.properties

@@ -140,11 +140,20 @@ rocketmq.producer.enable-msg-trace=true
 rocketmq.producer.customized-trace-topic=my-trace-topic
 
 mq.config.sessionTopic=themis-session-topic
-mq.config.sessionTopicTag=web||wxapp||pc
 mq.config.sessionConsumerGroup=themis-group
+mq.config.sessionTopicTag=web||wxapp||pc
+
+mq.config.sessionTopicWebTag=web
+mq.config.sessionConsumerWebGroup=themis-group-${mq.config.sessionTopicWebTag}
+
+mq.config.sessionTopicWxappTag=wxapp
+mq.config.sessionConsumerWxappGroup=themis-group-${mq.config.sessionTopicWxappTag}
+
+mq.config.sessionTopicPcTag=pc
+mq.config.sessionConsumerPcGroup=themis-group-${mq.config.sessionTopicPcTag}
 #dlq\u6B7B\u4FE1\u961F\u5217
-mq.config.sessionConsumerGroupDlq=${mq.config.sessionConsumerGroup}-dlq
-mq.config.sessionTopicDlq=%DLQ%${mq.config.sessionConsumerGroup}
+#mq.config.sessionConsumerGroupDlq=${mq.config.sessionConsumerGroup}-dlq
+#mq.config.sessionTopicDlq=%DLQ%${mq.config.sessionConsumerGroup}
 #============================================================================
 # \u914D\u7F6EJobStore
 #============================================================================

themis-business/src/main/java/com/qmth/themis/business/constant/SystemConstant.java

@@ -1,5 +1,6 @@
 package com.qmth.themis.business.constant;
 
+import com.qmth.themis.common.contanst.Constants;
 import com.qmth.themis.common.enums.Platform;
 import com.qmth.themis.common.enums.Source;
 
@@ -88,4 +89,15 @@ public class SystemConstant {
         }
         return calendar.getTime();
     }
+
+    /**
+     * 比较过期
+     *
+     * @param timestamp
+     * @return
+     */
+    public static boolean expire(long timestamp) {
+        long diff = (System.currentTimeMillis() - timestamp) / 1000;
+        return diff < -1 * Constants.SIGNATURE_AHEAD_SECONDS || diff > Constants.SIGNATURE_EXPIRE_SECONDS;
+    }
 }

themis-business/src/main/java/com/qmth/themis/business/domain/MqConfigDomain.java

@@ -10,10 +10,17 @@ package com.qmth.themis.business.domain;
 public class MqConfigDomain {
 
     private String sessionTopic;
-    private String sessionTopicTag;
+        private String sessionTopicTag;
     private String sessionConsumerGroup;
-    private String sessionConsumerGroupDlq;
-    private String sessionTopicDlq;
+    //    private String sessionConsumerGroupDlq;
+//    private String sessionTopicDlq;
+    private String sessionTopicWebTag;
+    private String sessionTopicWxappTag;
+    private String sessionTopicPcTag;
+    private String sessionConsumerWebGroup;
+    private String sessionConsumerWxappGroup;
+    private String sessionConsumerPcGroup;
+
 
     public String getSessionTopic() {
         return sessionTopic;
@@ -39,19 +46,67 @@ public class MqConfigDomain {
         this.sessionConsumerGroup = sessionConsumerGroup;
     }
 
-    public String getSessionConsumerGroupDlq() {
-        return sessionConsumerGroupDlq;
+//    public String getSessionConsumerGroupDlq() {
+//        return sessionConsumerGroupDlq;
+//    }
+//
+//    public void setSessionConsumerGroupDlq(String sessionConsumerGroupDlq) {
+//        this.sessionConsumerGroupDlq = sessionConsumerGroupDlq;
+//    }
+//
+//    public String getSessionTopicDlq() {
+//        return sessionTopicDlq;
+//    }
+//
+//    public void setSessionTopicDlq(String sessionTopicDlq) {
+//        this.sessionTopicDlq = sessionTopicDlq;
+//    }
+
+    public String getSessionTopicWebTag() {
+        return sessionTopicWebTag;
+    }
+
+    public void setSessionTopicWebTag(String sessionTopicWebTag) {
+        this.sessionTopicWebTag = sessionTopicWebTag;
+    }
+
+    public String getSessionTopicWxappTag() {
+        return sessionTopicWxappTag;
+    }
+
+    public void setSessionTopicWxappTag(String sessionTopicWxappTag) {
+        this.sessionTopicWxappTag = sessionTopicWxappTag;
+    }
+
+    public String getSessionTopicPcTag() {
+        return sessionTopicPcTag;
+    }
+
+    public void setSessionTopicPcTag(String sessionTopicPcTag) {
+        this.sessionTopicPcTag = sessionTopicPcTag;
+    }
+
+    public String getSessionConsumerWebGroup() {
+        return sessionConsumerWebGroup;
+    }
+
+    public void setSessionConsumerWebGroup(String sessionConsumerWebGroup) {
+        this.sessionConsumerWebGroup = sessionConsumerWebGroup;
+    }
+
+    public String getSessionConsumerWxappGroup() {
+        return sessionConsumerWxappGroup;
     }
 
-    public void setSessionConsumerGroupDlq(String sessionConsumerGroupDlq) {
-        this.sessionConsumerGroupDlq = sessionConsumerGroupDlq;
+    public void setSessionConsumerWxappGroup(String sessionConsumerWxappGroup) {
+        this.sessionConsumerWxappGroup = sessionConsumerWxappGroup;
     }
 
-    public String getSessionTopicDlq() {
-        return sessionTopicDlq;
+    public String getSessionConsumerPcGroup() {
+        return sessionConsumerPcGroup;
     }
 
-    public void setSessionTopicDlq(String sessionTopicDlq) {
-        this.sessionTopicDlq = sessionTopicDlq;
+    public void setSessionConsumerPcGroup(String sessionConsumerPcGroup) {
+        this.sessionConsumerPcGroup = sessionConsumerPcGroup;
     }
 }

themis-business/src/main/java/com/qmth/themis/business/dto/MqDto.java

@@ -0,0 +1,139 @@
+package com.qmth.themis.business.dto;
+
+import com.qmth.themis.business.constant.SystemConstant;
+import com.qmth.themis.business.enums.MqEnum;
+
+import java.io.Serializable;
+import java.util.Map;
+import java.util.UUID;
+
+/**
+ * @Description: mq dto
+ * @Param:
+ * @return:
+ * @Author: wangliang
+ * @Date: 2020/7/1
+ */
+public class MqDto implements Serializable {
+    private static final long serialVersionUID = 1L;
+
+    private String id;//消息id
+    private String topic;//消息topic
+    private String tag;//消息tag
+    private long timestamp;//时间戳
+    private Object body;//消息体
+    private MqEnum type;//业务类型
+    private String bid;//关联业务id
+    private Integer ack;//ack
+    private Integer sequence;//序号
+    private Map<String, Object> properties;//扩展类型
+
+    public MqDto() {
+
+    }
+
+    public MqDto(String topic, String tag, Object body, MqEnum type, String bid) {
+        this.topic = topic;
+        this.tag = tag;
+        this.body = body;
+        this.type = type;
+        this.bid = bid;
+        this.timestamp = System.currentTimeMillis();
+        this.id = String.valueOf(UUID.randomUUID()).replaceAll("-", "");
+    }
+
+    public MqDto(String topic, String tag, Object body, MqEnum type, String bid, Map properties) {
+        this.topic = topic;
+        this.tag = tag;
+        this.body = body;
+        this.type = type;
+        this.bid = bid;
+        this.timestamp = System.currentTimeMillis();
+        this.id = String.valueOf(UUID.randomUUID()).replaceAll("-", "");
+        this.properties = properties;
+    }
+
+    public static long getSerialVersionUID() {
+        return serialVersionUID;
+    }
+
+    public MqEnum getType() {
+        return type;
+    }
+
+    public void setType(MqEnum type) {
+        this.type = type;
+    }
+
+    public String getId() {
+        return id;
+    }
+
+    public void setId(String id) {
+        this.id = id;
+    }
+
+    public String getTopic() {
+        return topic;
+    }
+
+    public void setTopic(String topic) {
+        this.topic = topic;
+    }
+
+    public String getTag() {
+        return tag;
+    }
+
+    public void setTag(String tag) {
+        this.tag = tag;
+    }
+
+    public long getTimestamp() {
+        return timestamp;
+    }
+
+    public void setTimestamp(long timestamp) {
+        this.timestamp = timestamp;
+    }
+
+    public Object getBody() {
+        return body;
+    }
+
+    public void setBody(Object body) {
+        this.body = body;
+    }
+
+    public String getBid() {
+        return bid;
+    }
+
+    public void setBid(String bid) {
+        this.bid = bid;
+    }
+
+    public Map<String, Object> getProperties() {
+        return properties;
+    }
+
+    public void setProperties(Map<String, Object> properties) {
+        this.properties = properties;
+    }
+
+    public Integer getAck() {
+        return ack;
+    }
+
+    public void setAck(Integer ack) {
+        this.ack = ack;
+    }
+
+    public Integer getSequence() {
+        return sequence;
+    }
+
+    public void setSequence(Integer sequence) {
+        this.sequence = sequence;
+    }
+}

themis-business/src/main/java/com/qmth/themis/business/entity/TBSession.java

@@ -69,16 +69,16 @@ public class TBSession implements Serializable {
     @TableId(value = "expire_time")
     private Date expireTime;
 
-    @TableField(exist = false)
-    private Integer ack; //mq ack
-
-    public Integer getAck() {
-        return ack;
-    }
-
-    public void setAck(Integer ack) {
-        this.ack = ack;
-    }
+//    @TableField(exist = false)
+//    private Integer ack; //mq ack
+//
+//    public Integer getAck() {
+//        return ack;
+//    }
+//
+//    public void setAck(Integer ack) {
+//        this.ack = ack;
+//    }
 
     public static long getSerialVersionUID() {
         return serialVersionUID;
@@ -196,6 +196,6 @@ public class TBSession implements Serializable {
         this.expireTime = expireTime;
         this.lastAccessTime = new Date();
         this.updateTime = new Date();
-        this.ack = SystemConstant.UNSEND_ACK_TYPE;
+//        this.ack = SystemConstant.UNSEND_ACK_TYPE;
     }
 }

themis-business/src/main/java/com/qmth/themis/business/enums/MqEnum.java

@@ -0,0 +1,94 @@
+package com.qmth.themis.business.enums;
+
+import java.util.Objects;
+
+/**
+ * @Description: mq enum
+ * @Param:
+ * @return:
+ * @Author: wangliang
+ * @Date: 2020/7/1
+ */
+public enum MqEnum {
+
+    /**
+     * 用户session记录
+     */
+    SESSION(1, "用户session记录"),
+
+    /**
+     * 考生轨迹
+     */
+    EXAM_STUDENT_LOG(2, "考生轨迹"),
+    /**
+     * 预警日志
+     */
+    WARMING_LOG(3, "预警日志"),
+
+    /**
+     * 异常日志
+     */
+    EXCEPTION_LOG(4, "异常日志"),
+
+    /**
+     * 消息日志
+     */
+    MESSAGE_LOG(5, "消息日志");
+
+    private int id;
+
+    private String code;
+
+    private MqEnum(int id, String code) {
+        this.id = id;
+        this.code = code;
+    }
+
+    /**
+     * 状态转换 toId
+     *
+     * @param value
+     * @return
+     */
+    public static int convertToId(String value) {
+        if (Objects.equals(value.trim(), SESSION.name())) {
+            return SESSION.getId();
+        } else if (Objects.equals(value.trim(), EXAM_STUDENT_LOG.name())) {
+            return EXAM_STUDENT_LOG.getId();
+        } else if (Objects.equals(value.trim(), WARMING_LOG.name())) {
+            return WARMING_LOG.getId();
+        } else if (Objects.equals(value.trim(), EXCEPTION_LOG.name())) {
+            return EXCEPTION_LOG.getId();
+        } else {
+            return MESSAGE_LOG.getId();
+        }
+    }
+
+    /**
+     * 状态转换 toName
+     *
+     * @param value
+     * @return
+     */
+    public static String convertToName(String value) {
+        if (Objects.equals(value.trim(), SESSION.getCode())) {
+            return SESSION.name();
+        } else if (Objects.equals(value.trim(), EXAM_STUDENT_LOG.getCode())) {
+            return EXAM_STUDENT_LOG.name();
+        } else if (Objects.equals(value.trim(), WARMING_LOG.getCode())) {
+            return WARMING_LOG.name();
+        } else if (Objects.equals(value.trim(), EXCEPTION_LOG.getCode())) {
+            return EXCEPTION_LOG.name();
+        } else {
+            return MESSAGE_LOG.name();
+        }
+    }
+
+    public int getId() {
+        return id;
+    }
+
+    public String getCode() {
+        return code;
+    }
+}

themis-business/src/main/java/com/qmth/themis/business/listener/RocketConsumer.java

@@ -1,15 +1,16 @@
 package com.qmth.themis.business.listener;
 
+import com.alibaba.fastjson.JSON;
 import com.alibaba.fastjson.JSONObject;
 import com.qmth.themis.business.constant.SystemConstant;
+import com.qmth.themis.business.dto.MqDto;
 import com.qmth.themis.business.entity.TBSession;
 import com.qmth.themis.business.service.TBSessionService;
 import com.qmth.themis.business.util.RedisUtil;
 import com.qmth.themis.common.contanst.Constants;
 import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
-import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
-import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
-import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
+import org.apache.rocketmq.client.consumer.listener.*;
+import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
 import org.apache.rocketmq.common.message.Message;
 import org.apache.rocketmq.common.message.MessageExt;
 import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
@@ -22,9 +23,9 @@ import org.springframework.stereotype.Service;
 
 import javax.annotation.Resource;
 import java.util.List;
-import java.util.Map;
 import java.util.Objects;
 
+import static com.alibaba.fastjson.JSON.toJavaObject;
 import static com.alibaba.fastjson.JSONObject.parseObject;
 
 /**
@@ -35,7 +36,10 @@ import static com.alibaba.fastjson.JSONObject.parseObject;
  * @Date: 2020/6/28
  */
 @Service
-public class RocketConsumer implements MessageListenerConcurrently {
+public class RocketConsumer implements
+//        MessageListenerOrderly
+        MessageListenerConcurrently //并发消费
+{
 
     private final static Logger log = LoggerFactory.getLogger(RocketConsumer.class);
 
@@ -45,28 +49,34 @@ public class RocketConsumer implements MessageListenerConcurrently {
     @Resource
     RedisUtil redisUtil;
 
+    /**
+     * 并发消费
+     *
+     * @param msgs
+     * @param consumeConcurrentlyContext
+     * @return
+     */
     @Override
     public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
         try {
             for (MessageExt messageExt : msgs) {
                 log.info("sessionConsumer重试次数:{}", messageExt.getReconsumeTimes());
-                JSONObject jsonObject = parseObject(new String(messageExt.getBody(), Constants.CHARSET));
-                log.info("sessionConsumer接受到的消息:{}", jsonObject.toJSONString());
-                boolean waitStoreMsgOK = jsonObject.getBoolean("waitStoreMsgOK");
-                if (waitStoreMsgOK) {
-                    Map map = (Map) jsonObject.get(SystemConstant.PROPERTIES);
-                    String sessionId = (String.valueOf(map.get(String.valueOf(map.get("TAGS")))));
-                    if (Objects.nonNull(sessionId)) {
-                        TBSession tbSession = (TBSession) redisUtil.getSessionTopicList(sessionId);
-                        if (Objects.nonNull(tbSession.getAck()) && tbSession.getAck().intValue() != SystemConstant.STANDARD_ACK_TYPE) {
-                            tbSessionService.saveSessionInfo(tbSession);
-                            tbSession.setAck(SystemConstant.STANDARD_ACK_TYPE);
-                            redisUtil.setUserSession(tbSession.getId(), tbSession);
-                            redisUtil.deleteSessionTopicList(tbSession.getId());
-                        }
+                MqDto mqDto = (MqDto) toJavaObject(parseObject(new String(messageExt.getBody(), Constants.CHARSET)), MqDto.class);
+                log.info("sessionConsumer接受到的消息:{}", JSONObject.toJSONString(mqDto));
+                log.info("mqDto sequence:{},tag:{}", mqDto.getSequence(), mqDto.getTag());
+                MqDto redisMqdto = (MqDto) redisUtil.getSessionTopicList(mqDto.getId());
+                if (Objects.nonNull(redisMqdto)) {
+                    if (Objects.nonNull(redisMqdto.getAck()) && redisMqdto.getAck().intValue() != SystemConstant.STANDARD_ACK_TYPE) {
+                        log.info("更新db");
+                        tbSessionService.saveSessionInfo(toJavaObject((JSON) mqDto.getBody(), TBSession.class), redisMqdto.getTimestamp());
+                        redisUtil.deleteSessionTopicList(redisMqdto.getId());
+                        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
+                    } else {
+                        log.info("消息ack未确认，重发");
+                        return ConsumeConcurrentlyStatus.RECONSUME_LATER;//重试
                     }
-                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;//成功
                 } else {
+                    log.info("消息数据为空，重发消息");
                     return ConsumeConcurrentlyStatus.RECONSUME_LATER;//重试
                 }
             }
@@ -77,9 +87,48 @@ public class RocketConsumer implements MessageListenerConcurrently {
         return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;//成功
     }
 
+//    /**
+//     * 顺序消费
+//     *
+//     * @param msgs
+//     * @param consumeOrderlyContext
+//     * @return
+//     */
+//    @Override
+//    public ConsumeOrderlyStatus consumeMessage(List<MessageExt> msgs, ConsumeOrderlyContext
+//            consumeOrderlyContext) {
+//        try {
+//            for (MessageExt messageExt : msgs) {
+//                log.info("sessionConsumer重试次数:{}", messageExt.getReconsumeTimes());
+//                MqDto mqDto = (MqDto) toJavaObject(parseObject(new String(messageExt.getBody(), Constants.CHARSET)), MqDto.class);
+//                log.info("sessionConsumer接受到的消息:{}", JSONObject.toJSONString(mqDto));
+//                log.info("mqDto sequence:{},tag:{}", mqDto.getSequence(), mqDto.getTag());
+//                MqDto redisMqdto = (MqDto) redisUtil.getSessionTopicList(mqDto.getId());
+//                if (Objects.nonNull(redisMqdto)) {
+//                    if (Objects.nonNull(redisMqdto.getAck()) && redisMqdto.getAck().intValue() != SystemConstant.STANDARD_ACK_TYPE) {
+//                        log.info("更新db");
+//                        tbSessionService.saveSessionInfo(toJavaObject((JSON) mqDto.getBody(), TBSession.class), redisMqdto.getTimestamp());
+//                        redisUtil.deleteSessionTopicList(redisMqdto.getId());
+//                        return ConsumeOrderlyStatus.SUCCESS;
+//                    } else {
+//                        log.info("消息ack未确认，重发");
+//                        return ConsumeOrderlyStatus.SUSPEND_CURRENT_QUEUE_A_MOMENT;//重试
+//                    }
+//                } else {
+//                    log.info("消息数据为空，重发消息");
+//                    return ConsumeOrderlyStatus.SUSPEND_CURRENT_QUEUE_A_MOMENT;//重试
+//                }
+//            }
+//        } catch (Exception e) {
+//            e.printStackTrace();
+//            return ConsumeOrderlyStatus.SUSPEND_CURRENT_QUEUE_A_MOMENT;//重试
+//        }
+//        return ConsumeOrderlyStatus.SUCCESS;//成功
+//    }
+
     @Service
-    @RocketMQMessageListener(consumerGroup = "${mq.config.sessionConsumerGroup}", topic = "${mq.config.sessionTopic}", selectorType = SelectorType.TAG, selectorExpression = "${mq.config.sessionTopicTag}")
-    public class sessionConsumer implements RocketMQListener<Message>, RocketMQPushConsumerLifecycleListener {
+    @RocketMQMessageListener(consumerGroup = "${mq.config.sessionConsumerWebGroup}", topic = "${mq.config.sessionTopic}", selectorType = SelectorType.TAG, selectorExpression = "${mq.config.sessionTopicWebTag}")
+    public class sessionConsumerWeb implements RocketMQListener<Message>, RocketMQPushConsumerLifecycleListener {
 
         @Override
         public void onMessage(Message message) {
@@ -89,14 +138,15 @@ public class RocketConsumer implements MessageListenerConcurrently {
         @Override
         public void prepareStart(DefaultMQPushConsumer defaultMQPushConsumer) {
             defaultMQPushConsumer.setConsumeMessageBatchMaxSize(SystemConstant.CONSUME_MESSAGE_BATCH_MAX_SIZE);//每次拉取10条
+            defaultMQPushConsumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
             defaultMQPushConsumer.setMaxReconsumeTimes(SystemConstant.MAXRECONSUMETIMES);//最大重试次数
             defaultMQPushConsumer.registerMessageListener(RocketConsumer.this::consumeMessage);
         }
     }
 
     @Service
-    @RocketMQMessageListener(consumerGroup = "${mq.config.sessionConsumerGroupDlq}", topic = "${mq.config.sessionTopicDlq}", selectorType = SelectorType.TAG, selectorExpression = "${mq.config.sessionTopicTag}")
-    public class dlqSessionConsumer implements RocketMQListener<Message>, RocketMQPushConsumerLifecycleListener {
+    @RocketMQMessageListener(consumerGroup = "${mq.config.sessionConsumerPcGroup}", topic = "${mq.config.sessionTopic}", selectorType = SelectorType.TAG, selectorExpression = "${mq.config.sessionTopicPcTag}")
+    public class sessionConsumerPc implements RocketMQListener<Message>, RocketMQPushConsumerLifecycleListener {
 
         @Override
         public void onMessage(Message message) {
@@ -105,10 +155,50 @@ public class RocketConsumer implements MessageListenerConcurrently {
 
         @Override
         public void prepareStart(DefaultMQPushConsumer defaultMQPushConsumer) {
-            log.info("dlqSessionConsumer死信队列进来了");
             defaultMQPushConsumer.setConsumeMessageBatchMaxSize(SystemConstant.CONSUME_MESSAGE_BATCH_MAX_SIZE);//每次拉取10条
+            defaultMQPushConsumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
             defaultMQPushConsumer.setMaxReconsumeTimes(SystemConstant.MAXRECONSUMETIMES);//最大重试次数
             defaultMQPushConsumer.registerMessageListener(RocketConsumer.this::consumeMessage);
         }
     }
+
+    //
+    @Service
+    @RocketMQMessageListener(consumerGroup = "${mq.config.sessionConsumerWxappGroup}", topic = "${mq.config.sessionTopic}", selectorType = SelectorType.TAG, selectorExpression = "${mq.config.sessionTopicWxappTag}")
+    public class sessionConsumerWxapp implements RocketMQListener<Message>, RocketMQPushConsumerLifecycleListener {
+
+        @Override
+        public void onMessage(Message message) {
+            //实现RocketMQPushConsumerLifecycleListener监听器之后，此方法不调用
+        }
+
+        @Override
+        public void prepareStart(DefaultMQPushConsumer defaultMQPushConsumer) {
+            defaultMQPushConsumer.setConsumeMessageBatchMaxSize(SystemConstant.CONSUME_MESSAGE_BATCH_MAX_SIZE);//每次拉取10条
+            defaultMQPushConsumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
+            defaultMQPushConsumer.setMaxReconsumeTimes(SystemConstant.MAXRECONSUMETIMES);//最大重试次数
+            defaultMQPushConsumer.registerMessageListener(RocketConsumer.this::consumeMessage);
+        }
+    }
+
+    /**
+     * 死信队列
+     */
+//    @Service
+//    @RocketMQMessageListener(consumerGroup = "${mq.config.sessionConsumerGroupDlq}", topic = "${mq.config.sessionTopicDlq}", selectorType = SelectorType.TAG, selectorExpression = "${mq.config.sessionTopicTag}")
+//    public class dlqSessionConsumer implements RocketMQListener<Message>, RocketMQPushConsumerLifecycleListener {
+//
+//        @Override
+//        public void onMessage(Message message) {
+//            //实现RocketMQPushConsumerLifecycleListener监听器之后，此方法不调用
+//        }
+//
+//        @Override
+//        public void prepareStart(DefaultMQPushConsumer defaultMQPushConsumer) {
+//            log.info("dlqSessionConsumer死信队列进来了");
+//            defaultMQPushConsumer.setConsumeMessageBatchMaxSize(SystemConstant.CONSUME_MESSAGE_BATCH_MAX_SIZE);//每次拉取10条
+//            defaultMQPushConsumer.setMaxReconsumeTimes(SystemConstant.MAXRECONSUMETIMES);//最大重试次数
+//            defaultMQPushConsumer.registerMessageListener(RocketConsumer.this::consumeMessage);
+//        }
+//    }
 }

themis-business/src/main/java/com/qmth/themis/business/service/ProducerServer.java

@@ -1,5 +1,6 @@
 package com.qmth.themis.business.service;
 
+import com.qmth.themis.business.dto.MqDto;
 import com.qmth.themis.common.util.Result;
 
 /**
@@ -14,40 +15,57 @@ public interface ProducerServer {
     /**
      * 同步消息
      *
-     * @param topic
-     * @param tags
-     * @param msg
+     * @param mqDto
      * @return
      */
-    Result syncMsg(String topic, String tags, Object msg);
+    Result syncMsg(MqDto mqDto);
+
+    /**
+     * 同步顺序消息
+     *
+     * @param mqDto
+     * @return
+     */
+    Result syncOrderlyMsg(MqDto mqDto);
 
     /**
      * 异步消息
      *
-     * @param topic
-     * @param tags
-     * @param msg
+     * @param mqDto
+     * @return
+     */
+    Result asyncMsg(MqDto mqDto);
+
+    /**
+     * 异步顺序消息
+     *
+     * @param mqDto
      * @return
      */
-    Result asyncMsg(String topic, String tags, Object msg);
+    Result asyncOrderlyMsg(MqDto mqDto);
 
     /**
      * 单向消息
      *
-     * @param topic
-     * @param tags
-     * @param msg
+     * @param mqDto
+     * @return
+     */
+    Result sendOneWay(MqDto mqDto);
+
+    /**
+     * 单向顺序消息
+     *
+     * @param mqDto
      * @return
      */
-    Result sendOneWay(String topic, String tags, Object msg);
+    Result sendOneWayOrderly(MqDto mqDto);
+
 
     /**
      * 事务消息
      *
-     * @param topic
-     * @param tags
-     * @param msg
+     * @param mqDto
      * @return
      */
-    Result sendMsgTran(String topic, String tags, Object msg);
+    Result sendMsgTran(MqDto mqDto);
 }

themis-business/src/main/java/com/qmth/themis/business/service/TBSessionService.java

@@ -16,7 +16,8 @@ public interface TBSessionService extends IService<TBSession> {
      * 保存session信息
      *
      * @param tbSession
+     * @param redisTimestamp
      * @return
      */
-    boolean saveSessionInfo(TBSession tbSession);
+    boolean saveSessionInfo(TBSession tbSession, long redisTimestamp);
 }

themis-business/src/main/java/com/qmth/themis/business/service/impl/ProducerServerImpl.java

@@ -2,6 +2,7 @@ package com.qmth.themis.business.service.impl;
 
 import com.alibaba.fastjson.JSONObject;
 import com.qmth.themis.business.constant.SystemConstant;
+import com.qmth.themis.business.dto.MqDto;
 import com.qmth.themis.business.service.ProducerServer;
 import com.qmth.themis.common.exception.BusinessException;
 import com.qmth.themis.common.util.Result;
@@ -35,42 +36,68 @@ public class ProducerServerImpl implements ProducerServer {
     /**
      * 同步消息
      *
-     * @param topic
-     * @param tags
-     * @param msg
+     * @param mqDto
      * @return
      */
     @Override
-    public Result syncMsg(String topic, String tags, Object msg) {
-        log.info("syncMsg topic:{},tags:{},msg:{}", topic, tags, JSONObject.toJSONString(msg));
-        Message message = new Message();
-        message.setTopic(topic);
-        message.setBody(String.valueOf(msg).getBytes());
-        message.setTags(tags);
-        message.putUserProperty(tags, String.valueOf(msg));
-        SendResult sendResult = rocketMQTemplate.syncSend(topic, message);
+    public Result syncMsg(MqDto mqDto) {
+        log.info("syncMsg mqDto:{}", JSONObject.toJSONString(mqDto));
+        SendResult sendResult = rocketMQTemplate.syncSend(mqDto.getTopic() + ":" + mqDto.getTag(), mqDto);
         // 同步消息发送成功会有一个返回值，我们可以用这个返回值进行判断和获取一些信息
         log.info("sendResult:{}", JSONObject.toJSONString(sendResult));
         return ResultUtil.ok(SystemConstant.SUCCESS);
     }
 
+    /**
+     * 同步顺序消息
+     *
+     * @param mqDto
+     * @return
+     */
+    @Override
+    public Result syncOrderlyMsg(MqDto mqDto) {
+        log.info("syncOrderlyMsg mqDto:{}", JSONObject.toJSONString(mqDto));
+        SendResult sendResult = rocketMQTemplate.syncSendOrderly(mqDto.getTopic() + ":" + mqDto.getTag(), mqDto, mqDto.getBid());
+        // 同步顺序消息发送成功会有一个返回值，我们可以用这个返回值进行判断和获取一些信息
+        log.info("sendResult:{}", JSONObject.toJSONString(sendResult));
+        return ResultUtil.ok(SystemConstant.SUCCESS);
+    }
+
     /**
      * 异步消息
      *
-     * @param topic
-     * @param tags
-     * @param msg
+     * @param mqDto
      * @return
      */
     @Override
-    public Result asyncMsg(String topic, String tags, Object msg) {
-        log.info("asyncMsg topic:{},tags:{},msg:{}", topic, tags, JSONObject.toJSONString(msg));
-        Message message = new Message();
-        message.setTopic(topic);
-        message.setBody(String.valueOf(msg).getBytes());
-        message.setTags(tags);
-        message.putUserProperty(tags, String.valueOf(msg));
-        rocketMQTemplate.asyncSend(topic, message, new SendCallback() {
+    public Result asyncMsg(MqDto mqDto) {
+        log.info("asyncMsg mqDto:{}", JSONObject.toJSONString(mqDto));
+        rocketMQTemplate.asyncSend(mqDto.getTopic() + ":" + mqDto.getTag(), mqDto, new SendCallback() {
+            @Override
+            public void onSuccess(SendResult sendResult) {
+                // 成功回调
+                log.info("sendResult:{}", JSONObject.toJSONString(sendResult));
+            }
+
+            @Override
+            public void onException(Throwable throwable) {
+                // 失败回调
+                throw new BusinessException(throwable.getMessage());
+            }
+        });
+        return ResultUtil.ok(SystemConstant.SUCCESS);
+    }
+
+    /**
+     * 异步顺序消息
+     *
+     * @param mqDto
+     * @return
+     */
+    @Override
+    public Result asyncOrderlyMsg(MqDto mqDto) {
+        log.info("asyncOrderlyMsg mqDto:{}", JSONObject.toJSONString(mqDto));
+        rocketMQTemplate.asyncSendOrderly(mqDto.getTopic() + ":" + mqDto.getTag(), mqDto, mqDto.getBid(), new SendCallback() {
             @Override
             public void onSuccess(SendResult sendResult) {
                 // 成功回调
@@ -89,43 +116,58 @@ public class ProducerServerImpl implements ProducerServer {
     /**
      * 单向消息
      *
-     * @param topic
-     * @param tags
-     * @param msg
+     * @param mqDto
      * @return
      */
     @Override
-    public Result sendOneWay(String topic, String tags, Object msg) {
-        log.info("sendOneWay topic:{},tags:{},msg:{}", topic, tags, JSONObject.toJSONString(msg));
-        Message message = new Message();
-        message.setTopic(topic);
-        message.setBody(String.valueOf(msg).getBytes());
-        message.setTags(tags);
-        message.putUserProperty(tags, String.valueOf(msg));
-        rocketMQTemplate.sendOneWay(topic + ":" + tags, message);
+    public Result sendOneWay(MqDto mqDto) {
+        log.info("sendOneWay mqDto:{}", JSONObject.toJSONString(mqDto));
+        rocketMQTemplate.sendOneWay(mqDto.getTopic() + ":" + mqDto.getTag(), mqDto);
+        return ResultUtil.ok(SystemConstant.SUCCESS);
+    }
+
+    /**
+     * 单向顺序消息
+     *
+     * @param mqDto
+     * @return
+     */
+    @Override
+    public Result sendOneWayOrderly(MqDto mqDto) {
+        log.info("sendOneWayOrderly mqDto:{}", JSONObject.toJSONString(mqDto));
+        rocketMQTemplate.sendOneWayOrderly(mqDto.getTopic() + ":" + mqDto.getTag(), mqDto, mqDto.getBid());
         return ResultUtil.ok(SystemConstant.SUCCESS);
     }
 
     /**
      * 事务消息
      *
-     * @param topic
-     * @param tags
-     * @param msg
+     * @param mqDto
      * @return
      */
     @Override
-    public Result sendMsgTran(String topic, String tags, Object msg) {
-        log.info("sendMsgTran topic:{},tags:{},msg:{}", topic, tags, JSONObject.toJSONString(msg));
-        Message message = new Message();
-        message.setTopic(topic);
-        message.setBody(String.valueOf(msg).getBytes());
-        message.setTags(tags);
-        message.putUserProperty(tags, String.valueOf(msg));
+    public Result sendMsgTran(MqDto mqDto) {
+        log.info("sendMsgTran mqDto:{}", JSONObject.toJSONString(mqDto));
+        Message message = assembleMessage(mqDto);
         org.springframework.messaging.Message messageTran = MessageBuilder.withPayload(message).build();
         //发送事务消息
-        TransactionSendResult transactionSendResult = rocketMQTemplate.sendMessageInTransaction(topic, messageTran, null);
+        TransactionSendResult transactionSendResult = rocketMQTemplate.sendMessageInTransaction(mqDto.getTopic() + ":" + mqDto.getTag(), messageTran, null);
         log.info("transactionSendResult:{}", JSONObject.toJSONString(transactionSendResult));
         return ResultUtil.ok(SystemConstant.SUCCESS);
     }
+
+    /**
+     * 组装message
+     *
+     * @param mqDto
+     * @return
+     */
+    public Message assembleMessage(MqDto mqDto) {
+        Message message = new Message();
+        message.setTopic(mqDto.getTopic());
+        message.setBody(String.valueOf(mqDto.getBody()).getBytes());
+        message.setTags(mqDto.getTag());
+        message.putUserProperty("bid", String.valueOf(mqDto.getBid()));
+        return message;
+    }
 }

themis-business/src/main/java/com/qmth/themis/business/service/impl/TBSessionServiceImpl.java

@@ -1,11 +1,12 @@
 package com.qmth.themis.business.service.impl;
 
-import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
 import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
 import com.google.gson.Gson;
 import com.qmth.themis.business.dao.TBSessionMapper;
 import com.qmth.themis.business.entity.TBSession;
 import com.qmth.themis.business.service.TBSessionService;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 import org.springframework.stereotype.Service;
 
 import java.util.Date;
@@ -20,30 +21,37 @@ import java.util.Objects;
  */
 @Service
 public class TBSessionServiceImpl extends ServiceImpl<TBSessionMapper, TBSession> implements TBSessionService {
+    private final static Logger log = LoggerFactory.getLogger(TBSessionServiceImpl.class);
 
     /**
      * 保存session信息
      *
      * @param tbSession
+     * @param redisTimestamp
      * @return
      */
     @Override
-    public boolean saveSessionInfo(TBSession tbSession) {
-        QueryWrapper<TBSession> sqw = new QueryWrapper<>();
-        sqw.lambda().eq(TBSession::getId, tbSession.getId()).eq(TBSession::getType, tbSession.getType()).eq(TBSession::getPlatform, tbSession.getPlatform());
-        TBSession tbSessionDb = this.getOne(sqw);
+    public boolean saveSessionInfo(TBSession tbSession, long redisTimestamp) {
+        TBSession tbSessionDb = this.getById(tbSession.getId());
         if (Objects.isNull(tbSessionDb)) {
             Gson gson = new Gson();
             tbSessionDb = gson.fromJson(gson.toJson(tbSession), TBSession.class);
-        } else {
-            tbSessionDb.setDeviceId(tbSession.getDeviceId());
-            tbSessionDb.setAccessToken(tbSession.getAccessToken());
-            tbSessionDb.setLastAccessIp(tbSession.getLastAccessIp());
-            tbSessionDb.setLastAccessTime(new Date());
-            tbSessionDb.setUpdateTime(new Date());
-            tbSessionDb.setExpireTime(tbSession.getExpireTime());
+            this.saveOrUpdate(tbSessionDb);
+        } else if (Objects.nonNull(tbSessionDb.getUpdateTime())) {
+            long dbtimestamp = tbSessionDb.getUpdateTime().getTime();
+            if (redisTimestamp > dbtimestamp) {
+                log.info("redis时间大于db时间，可以更新");
+                tbSessionDb.setDeviceId(tbSession.getDeviceId());
+                tbSessionDb.setAccessToken(tbSession.getAccessToken());
+                tbSessionDb.setLastAccessIp(tbSession.getLastAccessIp());
+                tbSessionDb.setLastAccessTime(new Date());
+                tbSessionDb.setUpdateTime(new Date());
+                tbSessionDb.setExpireTime(tbSession.getExpireTime());
+                this.saveOrUpdate(tbSessionDb);
+            } else {
+                log.info("redis时间小于db时间，不予更新");
+            }
         }
-        this.saveOrUpdate(tbSessionDb);
         return true;
     }
 }