mq改造

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

@@ -122,14 +122,15 @@ public class TBUserController {
         Date expire = SystemConstant.getExpireTime(platform);
         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 {
 //            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());
+            //往mq发送消息插入会话信息
+            mqDto = new MqDto(dictionaryConfig.mqConfigDomain().getSessionTopic(), 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);

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

@@ -0,0 +1,19 @@
+package com.qmth.themis.backend.api;
+
+import io.swagger.annotations.Api;
+import org.springframework.web.bind.annotation.RequestMapping;
+import org.springframework.web.bind.annotation.RestController;
+
+/**
+ * @Description: 用户轨迹 前端控制器
+ * @Param:
+ * @return:
+ * @Author: wangliang
+ * @Date: 2020/7/2
+ */
+@Api(tags = "用户轨迹Controller")
+@RestController
+@RequestMapping("/${prefix.url.admin}/userLog")
+public class TEUserLogController {
+
+}

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

@@ -139,18 +139,30 @@ rocketmq.producer.secret-key=SK
 rocketmq.producer.enable-msg-trace=true
 rocketmq.producer.customized-trace-topic=my-trace-topic
 
-mq.config.sessionTopic=themis-session-topic
-mq.config.sessionConsumerGroup=themis-group
-mq.config.sessionTopicTag=web||wxapp||pc
+mq.config.server=themis
+#session_topic\u76D1\u542C
+mq.config.sessionTopic=${mq.config.server}-topic-session
+mq.config.sessionConsumerGroup=${mq.config.server}-group-session
 
 mq.config.sessionTopicWebTag=web
-mq.config.sessionConsumerWebGroup=themis-group-${mq.config.sessionTopicWebTag}
+mq.config.sessionConsumerWebGroup=${mq.config.sessionConsumerGroup}-${mq.config.sessionTopicWebTag}
 
 mq.config.sessionTopicWxappTag=wxapp
-mq.config.sessionConsumerWxappGroup=themis-group-${mq.config.sessionTopicWxappTag}
+mq.config.sessionConsumerWxappGroup=${mq.config.sessionConsumerGroup}-${mq.config.sessionTopicWxappTag}
 
 mq.config.sessionTopicPcTag=pc
-mq.config.sessionConsumerPcGroup=themis-group-${mq.config.sessionTopicPcTag}
+mq.config.sessionConsumerPcGroup=${mq.config.sessionConsumerGroup}-${mq.config.sessionTopicPcTag}
+
+#user_login\u76D1\u542C
+mq.config.userLogTopic=${mq.config.server}-topic-userLog
+mq.config.userLogConsumerGroup=${mq.config.server}-group-userLog
+
+mq.config.userLogTopicUserTag=user
+mq.config.userLogConsumerUserGroup=${mq.config.userLogConsumerGroup}-${mq.config.userLogTopicUserTag}
+
+mq.config.userLogTopicStudentTag=student
+mq.config.userLogConsumerStudentGroup=${mq.config.userLogConsumerGroup}-${mq.config.userLogTopicStudentTag}
+
 #dlq\u6B7B\u4FE1\u961F\u5217
 #mq.config.sessionConsumerGroupDlq=${mq.config.sessionConsumerGroup}-dlq
 #mq.config.sessionTopicDlq=%DLQ%${mq.config.sessionConsumerGroup}

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

@@ -42,6 +42,7 @@ public class SystemConstant {
     public static final String ACCOUNT = "account";
     public static final String ERROR = "/error";
     public static final String SESSION_TOPIC_ERROR_LIST = "session:topic:error:list";
+    public static final String USERLOG_TOPIC_ERROR_LIST = "userLog:topic:error:list";
     /**
      * session过期时间
      */
@@ -57,11 +58,8 @@ public class SystemConstant {
     /**
      * rocket mq
      */
-    public static final String SESSION_TOPIC = "themis-session-topic";
     public static final int CONSUME_MESSAGE_BATCH_MAX_SIZE = 10;
     public static final int MAXRECONSUMETIMES = 3;
-    public static final String PROPERTIES = "properties";
-
     public static final int DELIVERED_ACK_TYPE = 0;//消息"已发出"，但尚未处理结束
     public static final int POSION_ACK_TYPE = 1;//消息"错误",通常表示"抛弃"此消息，比如消息重发多次后，都无法正确处理时，消息将会被删除或者DLQ(死信队列)
     public static final int STANDARD_ACK_TYPE = 2;//"标准"类型,通常表示为消息"处理成功"，broker端可以删除消息了

themis-business/src/main/java/com/qmth/themis/business/dao/TEUserLogMapper.java

@@ -0,0 +1,17 @@
+package com.qmth.themis.business.dao;
+
+import com.baomidou.mybatisplus.core.mapper.BaseMapper;
+import com.qmth.themis.business.entity.TEUserLog;
+import org.apache.ibatis.annotations.Mapper;
+
+/** 
+* @Description: 用户轨迹 Mapper 接口
+* @Param:  
+* @return:  
+* @Author: wangliang
+* @Date: 2020/7/2 
+*/ 
+@Mapper
+public interface TEUserLogMapper extends BaseMapper<TEUserLog> {
+
+}

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

@@ -9,8 +9,8 @@ package com.qmth.themis.business.domain;
  */
 public class MqConfigDomain {
 
+    private String server;
     private String sessionTopic;
-        private String sessionTopicTag;
     private String sessionConsumerGroup;
     //    private String sessionConsumerGroupDlq;
 //    private String sessionTopicDlq;
@@ -21,6 +21,12 @@ public class MqConfigDomain {
     private String sessionConsumerWxappGroup;
     private String sessionConsumerPcGroup;
 
+    private String userLogTopic;
+    private String userLogConsumerGroup;
+    private String userLogTopicUserTag;
+    private String userLogConsumerUserGroup;
+    private String userLogTopicStudentTag;
+    private String userLogConsumerStudentGroup;
 
     public String getSessionTopic() {
         return sessionTopic;
@@ -30,14 +36,6 @@ public class MqConfigDomain {
         this.sessionTopic = sessionTopic;
     }
 
-    public String getSessionTopicTag() {
-        return sessionTopicTag;
-    }
-
-    public void setSessionTopicTag(String sessionTopicTag) {
-        this.sessionTopicTag = sessionTopicTag;
-    }
-
     public String getSessionConsumerGroup() {
         return sessionConsumerGroup;
     }
@@ -109,4 +107,60 @@ public class MqConfigDomain {
     public void setSessionConsumerPcGroup(String sessionConsumerPcGroup) {
         this.sessionConsumerPcGroup = sessionConsumerPcGroup;
     }
+
+    public String getServer() {
+        return server;
+    }
+
+    public void setServer(String server) {
+        this.server = server;
+    }
+
+    public String getUserLogTopic() {
+        return userLogTopic;
+    }
+
+    public void setUserLogTopic(String userLogTopic) {
+        this.userLogTopic = userLogTopic;
+    }
+
+    public String getUserLogConsumerGroup() {
+        return userLogConsumerGroup;
+    }
+
+    public void setUserLogConsumerGroup(String userLogConsumerGroup) {
+        this.userLogConsumerGroup = userLogConsumerGroup;
+    }
+
+    public String getUserLogTopicUserTag() {
+        return userLogTopicUserTag;
+    }
+
+    public void setUserLogTopicUserTag(String userLogTopicUserTag) {
+        this.userLogTopicUserTag = userLogTopicUserTag;
+    }
+
+    public String getUserLogConsumerUserGroup() {
+        return userLogConsumerUserGroup;
+    }
+
+    public void setUserLogConsumerUserGroup(String userLogConsumerUserGroup) {
+        this.userLogConsumerUserGroup = userLogConsumerUserGroup;
+    }
+
+    public String getUserLogTopicStudentTag() {
+        return userLogTopicStudentTag;
+    }
+
+    public void setUserLogTopicStudentTag(String userLogTopicStudentTag) {
+        this.userLogTopicStudentTag = userLogTopicStudentTag;
+    }
+
+    public String getUserLogConsumerStudentGroup() {
+        return userLogConsumerStudentGroup;
+    }
+
+    public void setUserLogConsumerStudentGroup(String userLogConsumerStudentGroup) {
+        this.userLogConsumerStudentGroup = userLogConsumerStudentGroup;
+    }
 }

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

@@ -0,0 +1,97 @@
+package com.qmth.themis.business.entity;
+
+import com.baomidou.mybatisplus.annotation.TableId;
+import io.swagger.annotations.ApiModel;
+import io.swagger.annotations.ApiModelProperty;
+
+import java.io.Serializable;
+import java.util.Date;
+
+/**
+ * @Description: 用户轨迹
+ * @Param:
+ * @return:
+ * @Author: wangliang
+ * @Date: 2020/7/2
+ */
+@ApiModel(value = "t_e_user_log", description = "用户轨迹")
+public class TEUserLog implements Serializable {
+
+    private static final long serialVersionUID = 1L;
+
+    @ApiModelProperty(value = "主键")
+    @TableId(value = "id")
+    private Long id;
+
+    @ApiModelProperty(value = "用户ID")
+    @TableId(value = "user_id")
+    private Long userId;
+
+    @ApiModelProperty(value = "类型")
+    @TableId(value = "type")
+    private Integer type;
+
+    @ApiModelProperty(value = "描述")
+    @TableId(value = "description")
+    private String description;
+
+    @ApiModelProperty(value = "备注")
+    @TableId(value = "remark")
+    private String remark;
+
+    @ApiModelProperty(value = "创建时间")
+    @TableId(value = "create_time")
+    private Date createTime;
+
+    public static long getSerialVersionUID() {
+        return serialVersionUID;
+    }
+
+    public Long getId() {
+        return id;
+    }
+
+    public void setId(Long id) {
+        this.id = id;
+    }
+
+    public Long getUserId() {
+        return userId;
+    }
+
+    public void setUserId(Long userId) {
+        this.userId = userId;
+    }
+
+    public Integer getType() {
+        return type;
+    }
+
+    public void setType(Integer type) {
+        this.type = type;
+    }
+
+    public String getDescription() {
+        return description;
+    }
+
+    public void setDescription(String description) {
+        this.description = description;
+    }
+
+    public String getRemark() {
+        return remark;
+    }
+
+    public void setRemark(String remark) {
+        this.remark = remark;
+    }
+
+    public Date getCreateTime() {
+        return createTime;
+    }
+
+    public void setCreateTime(Date createTime) {
+        this.createTime = createTime;
+    }
+}

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

@@ -0,0 +1,32 @@
+package com.qmth.themis.business.enums;
+
+/**
+ * @Description: 系统操作 enum
+ * @Param:
+ * @return:
+ * @Author: wangliang
+ * @Date: 2020/7/2
+ */
+public enum SystemOperationEnum {
+
+    LOGIN(1, "登录"),
+
+    LOGOUT(2, "注销");
+
+    private int id;
+
+    private String code;
+
+    private SystemOperationEnum(int id, String code) {
+        this.id = id;
+        this.code = code;
+    }
+
+    public int getId() {
+        return id;
+    }
+
+    public String getCode() {
+        return code;
+    }
+}

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

@@ -29,19 +29,19 @@ import static com.alibaba.fastjson.JSON.toJavaObject;
 import static com.alibaba.fastjson.JSONObject.parseObject;
 
 /**
- * @Description: 普通消息监听
+ * @Description: 普通消息监听 session_topic
  * @Param:
  * @return:
  * @Author: wangliang
  * @Date: 2020/6/28
  */
 @Service
-public class RocketConsumer implements
+public class RocketSessionConsumer implements
 //        MessageListenerOrderly
         MessageListenerConcurrently //并发消费
 {
 
-    private final static Logger log = LoggerFactory.getLogger(RocketConsumer.class);
+    private final static Logger log = LoggerFactory.getLogger(RocketSessionConsumer.class);
 
     @Resource
     TBSessionService tbSessionService;
@@ -140,7 +140,7 @@ public class RocketConsumer implements
             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);
+            defaultMQPushConsumer.registerMessageListener(RocketSessionConsumer.this::consumeMessage);
         }
     }
 
@@ -158,7 +158,7 @@ public class RocketConsumer implements
             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);
+            defaultMQPushConsumer.registerMessageListener(RocketSessionConsumer.this::consumeMessage);
         }
     }
 
@@ -177,7 +177,7 @@ public class RocketConsumer implements
             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);
+            defaultMQPushConsumer.registerMessageListener(RocketSessionConsumer.this::consumeMessage);
         }
     }
 

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

@@ -0,0 +1,74 @@
+package com.qmth.themis.business.listener;
+
+import com.qmth.themis.business.constant.SystemConstant;
+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.common.consumer.ConsumeFromWhere;
+import org.apache.rocketmq.common.message.Message;
+import org.apache.rocketmq.common.message.MessageExt;
+import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
+import org.apache.rocketmq.spring.annotation.SelectorType;
+import org.apache.rocketmq.spring.core.RocketMQListener;
+import org.apache.rocketmq.spring.core.RocketMQPushConsumerLifecycleListener;
+import org.springframework.stereotype.Service;
+
+import java.util.List;
+
+/**
+ * @Description: 普通消息监听 用户日志
+ * @Param:
+ * @return:
+ * @Author: wangliang
+ * @Date: 2020/7/2
+ */
+public class RocketUserLogConsumer implements MessageListenerConcurrently {
+
+    @Override
+    public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
+        try{
+
+        }catch(Exception e){
+            e.printStackTrace();
+            return ConsumeConcurrentlyStatus.RECONSUME_LATER;//重试
+        }
+        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;//成功
+    }
+
+    @Service
+    @RocketMQMessageListener(consumerGroup = "${mq.config.userLogConsumerUserGroup}", topic = "${mq.config.userLogTopic}", selectorType = SelectorType.TAG, selectorExpression = "${mq.config.userLogTopicUserTag}")
+    public class sessionConsumerUserLog 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(RocketUserLogConsumer.this::consumeMessage);
+        }
+    }
+
+    @Service
+    @RocketMQMessageListener(consumerGroup = "${mq.config.userLogConsumerStudentGroup}", topic = "${mq.config.userLogTopic}", selectorType = SelectorType.TAG, selectorExpression = "${mq.config.userLogTopicStudentTag}")
+    public class sessionConsumerStudentLog 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(RocketUserLogConsumer.this::consumeMessage);
+        }
+    }
+}

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

@@ -0,0 +1,15 @@
+package com.qmth.themis.business.service;
+
+import com.baomidou.mybatisplus.extension.service.IService;
+import com.qmth.themis.business.entity.TEUserLog;
+
+/**
+ * @Description: 用户轨迹 服务类
+ * @Param:
+ * @return:
+ * @Author: wangliang
+ * @Date: 2020/7/2
+ */
+public interface TEUserLogService extends IService<TEUserLog> {
+
+}

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

@@ -0,0 +1,19 @@
+package com.qmth.themis.business.service.impl;
+
+import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
+import com.qmth.themis.business.dao.TEUserLogMapper;
+import com.qmth.themis.business.entity.TEUserLog;
+import com.qmth.themis.business.service.TEUserLogService;
+import org.springframework.stereotype.Service;
+
+/**
+ * @Description: 用户轨迹 服务实现类
+ * @Param:
+ * @return:
+ * @Author: wangliang
+ * @Date: 2020/7/2
+ */
+@Service
+public class TEUserLogServiceImpl extends ServiceImpl<TEUserLogMapper, TEUserLog> implements TEUserLogService {
+
+}

themis-business/src/main/resources/mapper/TEUserLogMapper.xml

@@ -0,0 +1,5 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!DOCTYPE mapper PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN" "http://mybatis.org/dtd/mybatis-3-mapper.dtd">
+<mapper namespace="com.qmth.themis.business.dao.TEUserLogMapper">
+
+</mapper>