加入rocketmq

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

@@ -8,7 +8,7 @@ import com.qmth.themis.business.dto.AuthDto;
 import com.qmth.themis.business.entity.TBSession;
 import com.qmth.themis.business.entity.TBUser;
 import com.qmth.themis.business.service.EhcacheService;
-import com.qmth.themis.business.service.TBSessionService;
+import com.qmth.themis.business.service.ProducerServer;
 import com.qmth.themis.business.service.TBUserService;
 import com.qmth.themis.business.util.EhcacheUtil;
 import com.qmth.themis.business.util.RedisUtil;
@@ -59,10 +59,10 @@ public class TBUserController {
     TBUserService tbUserService;
 
     @Resource
-    TBSessionService tbSessionService;
+    EhcacheService ehcacheService;
 
     @Resource
-    EhcacheService ehcacheService;
+    ProducerServer producerServer;
 
     @ApiOperation(value = "用户登录接口")
     @RequestMapping(value = "/login/account", method = RequestMethod.POST)
@@ -106,8 +106,10 @@ public class TBUserController {
         String sessionId = SessionUtil.digest(user.getId(), authDto.getRoleEnum().name(), platform.getSource());
 
         Date expire = SystemConstant.getExpireTime(platform);
-        TBSession tbSession = tbSessionService.saveSessionInfo(sessionId, user.getId(), authDto.getRoleEnum().name(), platform.name(), platform.getSource(), deviceId, token, request.getLocalAddr(), expire);
+        TBSession tbSession = new TBSession(sessionId, String.valueOf(user.getId()), authDto.getRoleEnum().name(), platform.name(), platform.getSource(), deviceId, request.getLocalAddr(), token, expire);
         RedisUtil.setUserSession(sessionId, tbSession);
+        //往mq发送消息插入会话信息
+        producerServer.sendOneWay(SystemConstant.SESSION_TOPIC, platform.getSource(), tbSession);
 
         //测试
         String test = SignatureInfo.build(SignatureType.TOKEN, sessionId, token);

themis-backend/src/main/java/com/qmth/themis/backend/config/SwaggerConfig.java

@@ -2,6 +2,7 @@ package com.qmth.themis.backend.config;
 
 import com.google.common.base.Predicates;
 import com.qmth.themis.business.constant.SystemConstant;
+import com.qmth.themis.common.contanst.Constants;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
 import springfox.documentation.builders.ParameterBuilder;
@@ -31,9 +32,18 @@ public class SwaggerConfig {
     @Bean
     public Docket createRestApi() {
         ParameterBuilder tokenPar = new ParameterBuilder();
+        ParameterBuilder tokenPar1 = new ParameterBuilder();
+        ParameterBuilder tokenPar2 = new ParameterBuilder();
+        ParameterBuilder tokenPar3 = new ParameterBuilder();
         List<Parameter> pars = new ArrayList<>();
-        tokenPar.name(SystemConstant.ACCESS_TOKEN).description(SystemConstant.ACCESS_TOKEN).modelRef(new ModelRef("string")).parameterType("header").required(false).build();
+        tokenPar.name(Constants.HEADER_PLATFORM).description("平台").modelRef(new ModelRef("string")).parameterType("header").required(false).build();
+        tokenPar1.name(Constants.HEADER_DEVICE_ID).description("设备id").modelRef(new ModelRef("string")).parameterType("header").required(false).build();
+        tokenPar2.name(Constants.HEADER_AUTHORIZATION).description("鉴权token").modelRef(new ModelRef("string")).parameterType("header").required(false).build();
+        tokenPar3.name(Constants.HEADER_TIME).description("时间戳").modelRef(new ModelRef("long")).parameterType("header").required(false).build();
         pars.add(tokenPar.build());
+        pars.add(tokenPar1.build());
+        pars.add(tokenPar2.build());
+        pars.add(tokenPar3.build());
 
         return new Docket(DocumentationType.SWAGGER_2)
                 .groupName("themis-backend")

themis-backend/src/main/java/com/qmth/themis/backend/util/ServletUtil.java

@@ -29,11 +29,10 @@ public class ServletUtil {
      * @param message
      * @throws IOException
      */
-    public static void responseError(ServletResponse response, String code, String message) throws IOException {
-        HttpServletResponse httpResponse = (HttpServletResponse) response;
+    public static void responseError(HttpServletResponse response, String code, String message) throws IOException {
         Result result = ResultUtil.error(code, message);
         String json = JSONObject.toJSONString(result);
-        httpResponse.getWriter().print(json);
+        response.getWriter().print(json);
     }
 
     /**

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

@@ -125,9 +125,22 @@ sys.config.oss=false
 #sys.config.serverHost=localhost:7001
 #spring.resources.static-locations=file:${sys.config.serverUpload},classpath:/META-INF/resources/,classpath:/resources/
 
+rocketmq.name-server=127.0.0.1:9876
+rocketmq.producer.send-message-timeout=300000
+rocketmq.producer.group=my-group
+rocketmq.producer.compress-message-body-threshold=4096
+rocketmq.producer.max-message-size=4194304
+rocketmq.producer.retry-times-when-send-async-failed=0
+rocketmq.producer.retry-next-server=true
+rocketmq.producer.retry-times-when-send-failed=3
+rocketmq.producer.access-key=AK
+rocketmq.producer.secret-key=SK
+rocketmq.producer.enable-msg-trace=true
+rocketmq.producer.customized-trace-topic=my-trace-topic
+
 #api\u524D\u7F00
 prefix.url.admin=api/admin
 
 #\u65E0\u9700\u9274\u6743\u7684url
-no.auth.urls=/webjars/**,/druid/**,/swagger-ui.html,/doc.html,/swagger-resources,/v2/api-docs,/webjars/springfox-swagger-ui/**,/api/admin/user/login/account
+no.auth.urls=/webjars/**,/druid/**,/swagger-ui.html,/doc.html,/swagger-resources/**,/v2/api-docs,/webjars/springfox-swagger-ui/**,/api/admin/user/login/account
 common.system.urls=/api/admin/sys/getMenu,/api/admin/user/logout,/api/admin/sys/env

themis-business/pom.xml

@@ -87,5 +87,9 @@
             <groupId>org.springframework.boot</groupId>
             <artifactId>spring-boot-starter-aop</artifactId>
         </dependency>
+        <dependency>
+            <groupId>org.apache.rocketmq</groupId>
+            <artifactId>rocketmq-spring-boot-starter</artifactId>
+        </dependency>
     </dependencies>
 </project>

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

@@ -27,7 +27,6 @@ public class SystemConstant {
     public static final String ACCESS_KEY_ID = "accessKeyId";
     public static final String ACCESS_KEY_SECRET = "accessKeySecret";
     public static final String BUCKET = "bucket";
-
     /**
      * 系统相关
      */
@@ -40,14 +39,33 @@ public class SystemConstant {
     public static final String AUTH_CACHE = "auth_cache";
     public static final String ALL_PATH = "/**";
     public static final String ACCOUNT = "account";
-
+    public static final String ERROR = "/error";
+    /**
+     * session过期时间
+     */
     public static final int WEB_SESSION_EXPIRE = 1;//过期时间1天
     public static final int PC_SESSION_EXPIRE = 1;//过期时间1天
     public static final int WXAPP_SESSION_EXPIRE = 30;//过期时间30天
-
+    /**
+     * redis过期时间
+     */
     public static final long REDIS_EXPIRE_TIME = 60L * 1440L;//过期时间24小时
     public static final long REDIS_REFRESH_EXPIRE_TIME = 60L * 60L;//刷新时间1小时，理论上说24小时还不睡觉也已经快不行了
     public static final long REFRESH_EXPIRE_TIME = 60L * 5L;//过期剩余时间5分钟
+    /**
+     * 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端可以删除消息了
+    public static final int REDELIVERED_ACK_TYPE = 3;//消息需"重发"，比如consumer处理消息时抛出了异常，broker稍后会重新发送此消息
+    public static final int INDIVIDUAL_ACK_TYPE = 4;//表示只确认"单条消息",无论在任何ACK_MODE下
+    public static final int UNMATCHED_ACK_TYPE = 5;//BROKER间转发消息时,接收端"拒绝"消息
 
     /**
      * 获取过期时间

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

@@ -2,6 +2,7 @@ package com.qmth.themis.business.entity;
 
 import com.baomidou.mybatisplus.annotation.TableField;
 import com.baomidou.mybatisplus.annotation.TableId;
+import com.qmth.themis.business.constant.SystemConstant;
 import io.swagger.annotations.ApiModel;
 import io.swagger.annotations.ApiModelProperty;
 
@@ -68,6 +69,17 @@ 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;
+    }
+
     public static long getSerialVersionUID() {
         return serialVersionUID;
     }
@@ -167,4 +179,23 @@ public class TBSession implements Serializable {
     public void setExpireTime(Date expireTime) {
         this.expireTime = expireTime;
     }
+
+    public TBSession() {
+
+    }
+
+    public TBSession(String id, String identity, String type, String source, String platform, String deviceId, String address, String accessToken, Date expireTime) {
+        this.id = id;
+        this.identity = identity;
+        this.type = type;
+        this.source = source;
+        this.platform = platform;
+        this.deviceId = deviceId;
+        this.address = address;
+        this.accessToken = accessToken;
+        this.expireTime = expireTime;
+        this.lastAccessTime = new Date();
+        this.updateTime = new Date();
+        this.ack = SystemConstant.DELIVERED_ACK_TYPE;
+    }
 }

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

@@ -0,0 +1,76 @@
+package com.qmth.themis.business.listener;
+
+import com.alibaba.fastjson.JSONObject;
+import com.qmth.themis.business.constant.SystemConstant;
+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.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.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.stereotype.Service;
+
+import javax.annotation.Resource;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * @Description: 普通消息监听
+ * @Param:
+ * @return:
+ * @Author: wangliang
+ * @Date: 2020/6/28
+ */
+@Service
+public class RocketConsumer {
+
+    private final static Logger log = LoggerFactory.getLogger(RocketConsumer.class);
+
+    @Resource
+    TBSessionService tbSessionService;
+
+    @Service
+    @RocketMQMessageListener(consumerGroup = "themis-group", topic = "themis-session-topic", selectorType = SelectorType.TAG, selectorExpression = "web||wxapp||pc")
+    public class sessionConsumer implements RocketMQListener<Message>, RocketMQPushConsumerLifecycleListener {
+
+        @Override
+        public void onMessage(Message message) {
+            //实现RocketMQPushConsumerLifecycleListener监听器之后，此方法不调用
+            log.info("实现RocketMQPushConsumerLifecycleListener监听器之后，此方法不调用");
+        }
+
+        @Override
+        public void prepareStart(DefaultMQPushConsumer defaultMQPushConsumer) {
+            defaultMQPushConsumer.setConsumeMessageBatchMaxSize(SystemConstant.CONSUME_MESSAGE_BATCH_MAX_SIZE);//每次拉取10条
+            defaultMQPushConsumer.setMaxReconsumeTimes(SystemConstant.MAXRECONSUMETIMES);//最大重试次数
+            defaultMQPushConsumer.registerMessageListener(new MessageListenerConcurrently() {
+                @Override
+                public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
+                    for (MessageExt messageExt : msgs) {
+                        log.info("重试次数:{}", messageExt.getReconsumeTimes());
+                        JSONObject jsonObject = JSONObject.parseObject(new String(messageExt.getBody(), Constants.CHARSET));
+                        log.info("接受到的消息:{}", jsonObject.toJSONString());
+                        Map map = (Map) jsonObject.get(SystemConstant.PROPERTIES);
+                        TBSession tbSession = JSONObject.toJavaObject(JSONObject.parseObject(String.valueOf(map.get(String.valueOf(map.get("TAGS"))))), TBSession.class);
+                        tbSessionService.saveSessionInfo(tbSession);
+                        tbSession.setAck(SystemConstant.STANDARD_ACK_TYPE);
+                        RedisUtil.setUserSession(tbSession.getId(), tbSession);
+                        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
+                    }
+                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;//成功
+//                    return ConsumeConcurrentlyStatus.RECONSUME_LATER;//重试
+                }
+            });
+        }
+    }
+}

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

@@ -0,0 +1,58 @@
+package com.qmth.themis.business.listener;
+
+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.message.Message;
+import org.apache.rocketmq.common.message.MessageExt;
+import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
+import org.apache.rocketmq.spring.core.RocketMQListener;
+import org.apache.rocketmq.spring.core.RocketMQPushConsumerLifecycleListener;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.stereotype.Service;
+
+import java.util.List;
+
+/**
+ * @Description: 事务消息监听
+ * @Param:
+ * @return:
+ * @Author: wangliang
+ * @Date: 2020/6/28
+ */
+@Service
+public class RocketConsumerTranListener {
+    private final static Logger log = LoggerFactory.getLogger(RocketConsumerTranListener.class);
+
+    @Service
+    @RocketMQMessageListener(consumerGroup = "themis-tran-group", topic = "themis-tran-topic")
+    public class consumerTranListener implements RocketMQListener<Message>, RocketMQPushConsumerLifecycleListener {
+
+        @Override
+        public void onMessage(Message message) {
+            //实现RocketMQPushConsumerLifecycleListener监听器之后，此方法不调用
+            log.info("实现RocketMQPushConsumerLifecycleListener监听器之后，此方法不调用");
+        }
+
+        @Override
+        public void prepareStart(DefaultMQPushConsumer defaultMQPushConsumer) {
+            defaultMQPushConsumer.setConsumeMessageBatchMaxSize(10);//每次拉取10条
+            defaultMQPushConsumer.setMaxReconsumeTimes(1);//最大重试次数
+            defaultMQPushConsumer.registerMessageListener(new MessageListenerConcurrently() {
+                @Override
+                public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
+                    for (MessageExt messageExt : msgs) {
+                        log.info("重试次数:{}", messageExt.getReconsumeTimes());
+                        log.info("接受到的消息:{}", new String(messageExt.getBody()));
+                        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
+                    }
+                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
+//                    return ConsumeConcurrentlyStatus.RECONSUME_LATER;//重试
+//                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;//成功
+                }
+            });
+        }
+    }
+}

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

@@ -0,0 +1,55 @@
+package com.qmth.themis.business.listener;
+
+import com.alibaba.fastjson.JSONObject;
+import org.apache.rocketmq.spring.annotation.RocketMQTransactionListener;
+import org.apache.rocketmq.spring.core.RocketMQLocalTransactionListener;
+import org.apache.rocketmq.spring.core.RocketMQLocalTransactionState;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.stereotype.Service;
+
+/**
+ * @Description: 事务消息实现
+ * @Param:
+ * @return:
+ * @Author: wangliang
+ * @Date: 2020/6/28
+ */
+@Service
+@RocketMQTransactionListener
+public class RocketConsumerTransaction implements RocketMQLocalTransactionListener {
+    private final static Logger log = LoggerFactory.getLogger(RocketConsumerTransaction.class);
+
+    /**
+     * - 发送prepare消息成功此方法被回调，该方法用于执行本地事务
+     * - @param message 回传的消息，利用transactionId即可获取到该消息的唯一Id
+     * - @param arg 调用send方法时传递的参数，当send时候若有额外的参数可以传递到send方法中，这里能获取到
+     * - @return 返回事务状态，COMMIT：提交 ROLLBACK：回滚 UNKNOW：回调
+     */
+    @Override
+    public RocketMQLocalTransactionState executeLocalTransaction(org.springframework.messaging.Message message, Object arg) {
+//            try {
+//                log.info("【本地业务执行完毕】 msg:{}, Object:{}", message, o);
+//                localTrans.put(message.getHeaders().getId()+"", message.getPayload());
+//                return RocketMQLocalTransactionState.COMMIT;
+//            } catch (Exception e) {
+//                e.printStackTrace();
+//                log.error("【执行本地业务异常】 exception message:{}", e.getMessage());
+//                return RocketMQLocalTransactionState.ROLLBACK;
+//            }
+        log.info("本地事务和消息发送:{},o:{}", JSONObject.toJSONString(message), JSONObject.toJSONString(arg));
+        return RocketMQLocalTransactionState.UNKNOWN;
+    }
+
+    /**
+     * - @param message 通过获取transactionId来判断这条消息的本地事务执行状态
+     * - @return 返回事务状态，COMMIT：提交 ROLLBACK：回滚 UNKNOW：回调
+     */
+    @Override
+    public RocketMQLocalTransactionState checkLocalTransaction(org.springframework.messaging.Message message) {
+//            log.info("【执行检查任务】");
+//            return RocketMQLocalTransactionState.UNKNOWN;
+        log.info("回查信息:{}", JSONObject.toJSONString(message));
+        return RocketMQLocalTransactionState.COMMIT;
+    }
+}

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

@@ -0,0 +1,53 @@
+package com.qmth.themis.business.service;
+
+import com.qmth.themis.common.util.Result;
+
+/**
+ * @Description: mq 服务类
+ * @Param:
+ * @return:
+ * @Author: wangliang
+ * @Date: 2020/6/28
+ */
+public interface ProducerServer {
+
+    /**
+     * 同步消息
+     *
+     * @param topic
+     * @param tags
+     * @param msg
+     * @return
+     */
+    Result syncMsg(String topic, String tags, Object msg);
+
+    /**
+     * 异步消息
+     *
+     * @param topic
+     * @param tags
+     * @param msg
+     * @return
+     */
+    Result asyncMsg(String topic, String tags, Object msg);
+
+    /**
+     * 单向消息
+     *
+     * @param topic
+     * @param tags
+     * @param msg
+     * @return
+     */
+    Result sendOneWay(String topic, String tags, Object msg);
+
+    /**
+     * 事务消息
+     *
+     * @param topic
+     * @param tags
+     * @param msg
+     * @return
+     */
+    Result sendMsgTran(String topic, String tags, Object msg);
+}

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

@@ -15,8 +15,8 @@ public interface TBSessionService extends IService<TBSession> {
     /**
      * 保存session信息
      *
-     * @param o
+     * @param tbSession
      * @return
      */
-    TBSession saveSessionInfo(Object... o);
+    boolean saveSessionInfo(TBSession tbSession);
 }

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

@@ -0,0 +1,131 @@
+package com.qmth.themis.business.service.impl;
+
+import com.alibaba.fastjson.JSONObject;
+import com.qmth.themis.business.constant.SystemConstant;
+import com.qmth.themis.business.service.ProducerServer;
+import com.qmth.themis.common.exception.BusinessException;
+import com.qmth.themis.common.util.Result;
+import com.qmth.themis.common.util.ResultUtil;
+import org.apache.rocketmq.client.producer.SendCallback;
+import org.apache.rocketmq.client.producer.SendResult;
+import org.apache.rocketmq.client.producer.TransactionSendResult;
+import org.apache.rocketmq.common.message.Message;
+import org.apache.rocketmq.spring.core.RocketMQTemplate;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.messaging.support.MessageBuilder;
+import org.springframework.stereotype.Service;
+
+import javax.annotation.Resource;
+
+/**
+ * @Description: mq 服务实现类
+ * @Param:
+ * @return:
+ * @Author: wangliang
+ * @Date: 2020/6/28
+ */
+@Service
+public class ProducerServerImpl implements ProducerServer {
+    private final static Logger log = LoggerFactory.getLogger(ProducerServerImpl.class);
+
+    @Resource
+    RocketMQTemplate rocketMQTemplate;
+
+    /**
+     * 同步消息
+     *
+     * @param topic
+     * @param tags
+     * @param msg
+     * @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(JSONObject.toJSONString(msg).getBytes());
+        message.setTags(tags);
+        message.putUserProperty(tags, JSONObject.toJSONString(msg));
+        SendResult sendResult = rocketMQTemplate.syncSend(topic, message);
+        // 同步消息发送成功会有一个返回值，我们可以用这个返回值进行判断和获取一些信息
+        log.info("sendResult:{}", JSONObject.toJSONString(sendResult));
+        return ResultUtil.ok(SystemConstant.SUCCESS);
+    }
+
+    /**
+     * 异步消息
+     *
+     * @param topic
+     * @param tags
+     * @param msg
+     * @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(JSONObject.toJSONString(msg).getBytes());
+        message.setTags(tags);
+        message.putUserProperty(tags, JSONObject.toJSONString(msg));
+        rocketMQTemplate.asyncSend(topic, message, 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 topic
+     * @param tags
+     * @param msg
+     * @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(JSONObject.toJSONString(msg).getBytes());
+        message.setTags(tags);
+        message.putUserProperty(tags, JSONObject.toJSONString(msg));
+        rocketMQTemplate.sendOneWay(topic + ":" + tags, message);
+        return ResultUtil.ok(SystemConstant.SUCCESS);
+    }
+
+    /**
+     * 事务消息
+     *
+     * @param topic
+     * @param tags
+     * @param msg
+     * @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(JSONObject.toJSONString(msg).getBytes());
+        message.setTags(tags);
+        message.putUserProperty(tags, JSONObject.toJSONString(msg));
+        org.springframework.messaging.Message messageTran = MessageBuilder.withPayload(message).build();
+        //发送事务消息
+        TransactionSendResult transactionSendResult = rocketMQTemplate.sendMessageInTransaction(topic, messageTran, null);
+        log.info("transactionSendResult:{}", JSONObject.toJSONString(transactionSendResult));
+        return ResultUtil.ok(SystemConstant.SUCCESS);
+    }
+}

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

@@ -2,6 +2,7 @@ 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;
@@ -23,43 +24,26 @@ public class TBSessionServiceImpl extends ServiceImpl<TBSessionMapper, TBSession
     /**
      * 保存session信息
      *
-     * @param o
+     * @param tbSession
      * @return
      */
     @Override
-    public TBSession saveSessionInfo(Object... o) {
-        String id = String.valueOf(o[0]);
-        String type = String.valueOf(o[2]);
-        String platform = String.valueOf(o[3]);
-        String source = String.valueOf(o[4]);
-        String deviceId = String.valueOf(o[5]);
-        String token = String.valueOf(o[6]);
-        String ip = String.valueOf(o[7]);
+    public boolean saveSessionInfo(TBSession tbSession) {
         QueryWrapper<TBSession> sqw = new QueryWrapper<>();
-        sqw.lambda().eq(TBSession::getId, id).eq(TBSession::getType, type).eq(TBSession::getPlatform, platform);
-        TBSession tbSession = this.getOne(sqw);
-        if (Objects.isNull(tbSession)) {
-            tbSession = new TBSession();
-            tbSession.setId(id);
-            tbSession.setIdentity(String.valueOf(o[1]));
-            tbSession.setType(type);
-            tbSession.setPlatform(platform);
-            tbSession.setSource(source);
-            tbSession.setDeviceId(deviceId);
-            tbSession.setAccessToken(token);
-            tbSession.setAddress(ip);
-            tbSession.setLastAccessTime(new Date());
-            tbSession.setUpdateTime(new Date());
-            tbSession.setExpireTime((Date) o[8]);
+        sqw.lambda().eq(TBSession::getId, tbSession.getId()).eq(TBSession::getType, tbSession.getType()).eq(TBSession::getPlatform, tbSession.getPlatform());
+        TBSession tbSessionDb = this.getOne(sqw);
+        if (Objects.isNull(tbSessionDb)) {
+            Gson gson = new Gson();
+            tbSessionDb = gson.fromJson(gson.toJson(tbSession), TBSession.class);
         } else {
-            tbSession.setDeviceId(deviceId);
-            tbSession.setAccessToken(token);
-            tbSession.setLastAccessIp(ip);
-            tbSession.setLastAccessTime(new Date());
-            tbSession.setUpdateTime(new Date());
-            tbSession.setExpireTime((Date) o[8]);
+            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(tbSession);
-        return tbSession;
+        this.saveOrUpdate(tbSessionDb);
+        return true;
     }
 }

themis-common/src/main/java/com/qmth/themis/common/contanst/Constants.java

@@ -35,4 +35,6 @@ public interface Constants {
     public static final String AES_MODE_PKCS5 = "AES/CBC/PKCS5Padding";//用这个模式，规则必须为16位
     public static final String AES_MODE_PKCS7 = "AES/CBC/PKCS7Padding";//用这个模式，规则必须为16位
     public static final String AES_RULE = "1234567890123456";//aes密钥
+
+    public static final String MD5 = "MD5";
 }

themis-common/src/main/java/com/qmth/themis/common/enums/ExceptionResultEnum.java

@@ -90,6 +90,8 @@ public enum ExceptionResultEnum {
 
     NOT_FOUND("404", "请求地址错误"),
 
+    SERVICE_NOT_FOUND("500", "服务器错误"),
+
     UN_AUTHORIZATION("401", "没有权限"),
 
     AUTHORIZATION_ERROR("401", "签名验证失败"),

themis-common/src/main/java/com/qmth/themis/common/util/MD5Util.java

@@ -1,5 +1,7 @@
 package com.qmth.themis.common.util;
 
+import com.qmth.themis.common.contanst.Constants;
+
 import java.nio.charset.Charset;
 import java.security.MessageDigest;
 import java.security.NoSuchAlgorithmException;
@@ -24,8 +26,8 @@ public class MD5Util {
      */
     public static String encoder(String text) throws NoSuchAlgorithmException {
         text = Optional.of(text).get();
-        MessageDigest digest = MessageDigest.getInstance("MD5");
-        digest.update(text.getBytes(Charset.forName("utf8")));
+        MessageDigest digest = MessageDigest.getInstance(Constants.MD5);
+        digest.update(text.getBytes(Constants.CHARSET));
         byte s[] = digest.digest();
         StringJoiner result = new StringJoiner("");
         for (int i = 0; i < s.length; i++) {

themis-common/src/main/java/com/qmth/themis/common/util/ResultUtil.java

@@ -1,5 +1,6 @@
 package com.qmth.themis.common.util;
 
+import cn.hutool.http.HttpStatus;
 import com.qmth.themis.common.enums.ExceptionResultEnum;
 import com.qmth.themis.common.exception.BusinessException;
 
@@ -33,12 +34,12 @@ public class ResultUtil {
     }
 
     public static Result ok(String message) {
-        return new Result(String.valueOf(cn.hutool.http.HttpStatus.HTTP_OK), message);
+        return new Result(String.valueOf(HttpStatus.HTTP_OK), message);
     }
 
     public static Result ok(Object data) {
         Result r = new Result();
-        r.setCode(String.valueOf(cn.hutool.http.HttpStatus.HTTP_OK));
+        r.setCode(String.valueOf(HttpStatus.HTTP_OK));
         r.setData(data);
         r.setMessage(ExceptionResultEnum.SUCCESS.getMessage());
         return r;
@@ -46,7 +47,7 @@ public class ResultUtil {
 
     public static Result ok(Object data, String message) {
         Result r = new Result();
-        r.setCode(String.valueOf(cn.hutool.http.HttpStatus.HTTP_OK));
+        r.setCode(String.valueOf(HttpStatus.HTTP_OK));
         r.setData(data);
         r.setMessage(message);
         return r;
@@ -54,7 +55,7 @@ public class ResultUtil {
 
     public static Result ok() {
         Result r = new Result();
-        r.setCode(String.valueOf(cn.hutool.http.HttpStatus.HTTP_OK));
+        r.setCode(String.valueOf(HttpStatus.HTTP_OK));
         return r;
     }
 }

themis-exam/src/main/java/com/qmth/themis/exam/config/SwaggerConfig.java

@@ -2,6 +2,7 @@ package com.qmth.themis.exam.config;
 
 import com.google.common.base.Predicates;
 import com.qmth.themis.business.constant.SystemConstant;
+import com.qmth.themis.common.contanst.Constants;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
 import springfox.documentation.builders.ParameterBuilder;
@@ -31,9 +32,18 @@ public class SwaggerConfig {
     @Bean
     public Docket createRestApi() {
         ParameterBuilder tokenPar = new ParameterBuilder();
+        ParameterBuilder tokenPar1 = new ParameterBuilder();
+        ParameterBuilder tokenPar2 = new ParameterBuilder();
+        ParameterBuilder tokenPar3 = new ParameterBuilder();
         List<Parameter> pars = new ArrayList<>();
-        tokenPar.name(SystemConstant.ACCESS_TOKEN).description(SystemConstant.ACCESS_TOKEN).modelRef(new ModelRef("string")).parameterType("header").required(false).build();
+        tokenPar.name(Constants.HEADER_PLATFORM).description("平台").modelRef(new ModelRef("string")).parameterType("header").required(false).build();
+        tokenPar1.name(Constants.HEADER_DEVICE_ID).description("设备id").modelRef(new ModelRef("string")).parameterType("header").required(false).build();
+        tokenPar2.name(Constants.HEADER_AUTHORIZATION).description("鉴权token").modelRef(new ModelRef("string")).parameterType("header").required(false).build();
+        tokenPar3.name(Constants.HEADER_TIME).description("时间戳").modelRef(new ModelRef("long")).parameterType("header").required(false).build();
         pars.add(tokenPar.build());
+        pars.add(tokenPar1.build());
+        pars.add(tokenPar2.build());
+        pars.add(tokenPar3.build());
 
         return new Docket(DocumentationType.SWAGGER_2)
                 .groupName("themis-exam")

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

@@ -129,5 +129,5 @@ sys.config.oss=false
 prefix.url.exam=api/exam
 
 #\u65E0\u9700\u9274\u6743\u7684url
-no.auth.urls=/webjars/**,/druid/**,/swagger-ui.html,/doc.html,/swagger-resources,/v2/api-docs,/webjars/springfox-swagger-ui/**,/api/admin/user/login/account
+no.auth.urls=/webjars/**,/druid/**,/swagger-ui.html,/doc.html,/swagger-resources/**,/v2/api-docs,/webjars/springfox-swagger-ui/**,/api/admin/user/login/account
 common.system.urls=/api/admin/sys/getMenu,/api/admin/user/logout