update 增加序列化

update redis订阅发布改为redisson订阅发布

coffee-common/src/main/java/com/coffee/common/config/websocket/DefaultMessageListener.java

@@ -1,25 +1,18 @@
 package com.coffee.common.config.websocket;
 
-import cn.hutool.core.collection.CollectionUtil;
-import cn.hutool.json.JSONObject;
 import cn.hutool.json.JSONUtil;
-import com.coffee.common.cache.value.Value;
-import com.coffee.common.config.websocket.handler.TopicWrapper;
 import com.fasterxml.jackson.core.JsonProcessingException;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import lombok.Data;
 import lombok.extern.slf4j.Slf4j;
-import org.springframework.data.redis.connection.Message;
-import org.springframework.data.redis.connection.MessageListener;
+import org.redisson.api.RPatternTopic;
+import org.redisson.api.listener.PatternMessageListener;
 import org.tio.core.ChannelContext;
 import org.tio.core.Tio;
 import org.tio.core.utils.TioUtils;
 import org.tio.websocket.common.WsPacket;
 import org.tio.websocket.common.WsResponse;
 
-import java.util.HashSet;
-import java.util.Set;
-
 /**
  * @author lifang
  * @version 1.0.0
@@ -29,34 +22,26 @@ import java.util.Set;
  */
 @Data
 @Slf4j
-public class DefaultMessageListener implements MessageListener {
+public class DefaultMessageListener implements PatternMessageListener<TopicMessage> {
     private final String id;
-    private final TopicWrapper topicWrapper;
     private final ObjectMapper objectMapper;
-    private final  Set<ChannelContext> channelContexts;
+    private final ChannelContext channelContext;
+    private final RPatternTopic rPatternTopic;
+
+
     @Override
-    public void onMessage(Message message, byte[] pattern) {
-        if (CollectionUtil.isNotEmpty(channelContexts)) {
-            Set<ChannelContext> closeChannel = new HashSet<>();
-            channelContexts.parallelStream()
-                    .forEach(channel -> {
-                        if (!TioUtils.checkBeforeIO(channel)) {
-                            closeChannel.add(channel);
-                            return;
-                        }
-                        String json = null;
-                        try {
-                            JSONObject jsonObject = JSONUtil.parseObj(message.toString());
-                            json = objectMapper.writeValueAsString(MessageResponse.of(id, "result", Value.simple( jsonObject.get("param")).asString(),
-                                    JSONUtil.parse(jsonObject.get("message"))));
-                            Tio.send(channel,WsResponse.fromText(json, WsPacket.CHARSET_NAME));
-                        } catch (JsonProcessingException e) {
-                            log.error("ws消息订阅,解析失败,message:【】",message.toString());
-                        }
-                            }
-                    );
-            channelContexts.removeAll(closeChannel);
+    public void onMessage(CharSequence pattern, CharSequence channel, TopicMessage msg) {
+        if (TioUtils.checkBeforeIO(channelContext)) {
+            String json = null;
+            try {
+                json = objectMapper.writeValueAsString(MessageResponse.of(id, "result", msg.getParam(),
+                        JSONUtil.parse(objectMapper.writeValueAsString(msg.getMessage()))));
+                Tio.send(channelContext, WsResponse.fromText(json, WsPacket.CHARSET_NAME));
+            } catch (JsonProcessingException e) {
+                log.error("ws消息订阅,解析失败,message:【】", JSONUtil.toJsonStr(msg));
+            }
+        }else {
+            rPatternTopic.removeListener(this);
         }
     }
-
-}
+}

coffee-common/src/main/java/com/coffee/common/config/websocket/DefaultWebSocketMsgHandler.java

@@ -6,7 +6,6 @@ import cn.hutool.core.util.StrUtil;
 import cn.hutool.json.JSONUtil;
 import com.coffee.common.Constants;
 import com.coffee.common.bo.LoginUser;
-import com.coffee.common.cache.value.Value;
 import com.coffee.common.config.websocket.handler.WsHandler;
 import com.coffee.common.result.R;
 import lombok.AllArgsConstructor;

coffee-common/src/main/java/com/coffee/common/config/websocket/TopicMessage.java

@@ -0,0 +1,16 @@
+package com.coffee.common.config.websocket;
+
+import lombok.AllArgsConstructor;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+import java.io.Serializable;
+
+@Data
+@AllArgsConstructor(staticName = "of")
+@NoArgsConstructor
+public class TopicMessage implements Serializable {
+    private Object message;
+    private String param;
+
+}

coffee-common/src/main/java/com/coffee/common/config/websocket/handler/Subscribe.java

@@ -1,30 +1,21 @@
 package com.coffee.common.config.websocket.handler;
 
 import cn.hutool.core.collection.CollectionUtil;
-import cn.hutool.core.util.StrUtil;
 import com.coffee.common.Constants;
 import com.coffee.common.bo.LoginUser;
 import com.coffee.common.config.websocket.DefaultMessageListener;
 import com.coffee.common.config.websocket.MessagingRequest;
+import com.coffee.common.config.websocket.TopicMessage;
 import com.coffee.common.config.websocket.WebSocketConstant;
+import com.coffee.common.util.RedissonUtil;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import lombok.extern.slf4j.Slf4j;
+import org.redisson.api.RPatternTopic;
 import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.dao.DataAccessException;
-import org.springframework.data.redis.connection.RedisConnection;
-import org.springframework.data.redis.connection.RedisPubSubCommands;
-import org.springframework.data.redis.connection.Subscription;
-import org.springframework.data.redis.core.RedisCallback;
-import org.springframework.data.redis.core.RedisTemplate;
 import org.springframework.util.ConcurrentReferenceHashMap;
 import org.tio.core.ChannelContext;
 
-import javax.annotation.Resource;
 import java.util.*;
-import java.util.concurrent.CompletableFuture;
-import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.TimeUnit;
-import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.stream.Collectors;
 
 /**
@@ -36,23 +27,15 @@ import java.util.stream.Collectors;
  */
 @Slf4j
 public abstract class Subscribe implements WsHandler {
-    @Autowired
-    private RedisTemplate<String,Object> redisTemplate;
-
     @Autowired
     private ObjectMapper objectMapper;
 
-    public static final String SUBSCRIBE_TOPIC="subscribe-topic";
-    /**
-     * 存储主题与ws通道关联
-     */
-    private static Map<String,Set<ChannelContext>> subscribeTopics=new ConcurrentHashMap<>();
-
-
+    @Autowired
+    private RedissonUtil redissonUtil;
     /**
      * 存储主题与redis通道关联
      */
-    private static Map<String,RedisConnection> redisConnectionMap=new ConcurrentReferenceHashMap<>();
+    private static Map<String, RPatternTopic> topicMap=new ConcurrentReferenceHashMap<>();
 
 
     public TopicWrapper getTopic(String productName,String param,String tenantId){
@@ -60,10 +43,6 @@ public abstract class Subscribe implements WsHandler {
 
     };
 
-    public static Set<ChannelContext> getSubscribeChannel(String topic){
-        return subscribeTopics.get(topic);
-    }
-
 
     @Override
     public void onMessage(MessagingRequest message, ChannelContext channelContext) {
@@ -91,13 +70,10 @@ public abstract class Subscribe implements WsHandler {
         MessagingRequest.Type type = message.getType();
         if(MessagingRequest.Type.sub==type){
             //订阅主题
-            for (TopicWrapper topicWrapper : subScribeTopic) {
-                this.subscribe(channelContext,topicWrapper);
-            }
-//            subScribeTopic.forEach(topicWrapper->this.subscribe(channelContext,topicWrapper));
+            subScribeTopic.forEach(topicWrapper->this.subscribe(channelContext,topicWrapper));
         }else {
             //取消订阅主题
-//            subScribeTopic.forEach(topicWrapper->this.unsubscribe(channelContext,topicWrapper.getTopic()));
+            subScribeTopic.forEach(topicWrapper->this.unsubscribe(channelContext,topicWrapper.getTopic()));
         }
         log.error("订阅成功{}",subScribeTopic.stream().map(TopicWrapper::getTopic).collect(Collectors.toList()));
 
@@ -110,40 +86,8 @@ public abstract class Subscribe implements WsHandler {
      */
     public void subscribe(ChannelContext channelContext, TopicWrapper topicWrapper){
         //同一主题只订阅一次
-        Set<ChannelContext> channelContexts = subscribeTopics.computeIfAbsent(topicWrapper.getTopic(), k -> new HashSet<>());
-        channelContexts.add(channelContext);
-        boolean subscribe=false;
-        RedisConnection redisConnection = redisConnectionMap.get(topicWrapper.getTopic());
-        if(redisConnection==null){
-            subscribe=true;
-        }else {
-            Subscription subscription = redisConnection.getSubscription();
-            if(subscription==null||CollectionUtil.isEmpty(subscription.getPatterns())){
-                subscribe=true;
-            }else {
-                Collection<byte[]> patterns = subscription.getPatterns();
-                for (byte[] pattern : patterns) {
-                    String patternName = new String(pattern);
-                    if(topicWrapper.getTopic().equals(patternName)){
-                        subscribe=true;
-                        break;
-                    }
-                }
-            }
-        }
-        if(subscribe){
-            redisTemplate.execute(new RedisCallback<Object>() {
-                @Override
-                public Object doInRedis(RedisConnection connection) throws DataAccessException {
-                    CompletableFuture.runAsync(()->{
-                        connection.pSubscribe(new DefaultMessageListener(getId(),topicWrapper,objectMapper,channelContexts),topicWrapper.getTopic().getBytes());
-                    });
-                    redisConnectionMap.put(topicWrapper.getTopic(),connection);
-                    return null;
-                }
-            });
-        }
-        getTopicByChannel(channelContext).add(topicWrapper.getTopic());
+        RPatternTopic rTopic = topicMap.computeIfAbsent(topicWrapper.getTopic(),topic->redissonUtil.getPatternTopic(topicWrapper.getTopic()));
+        addTopicListener(rTopic,channelContext, topicWrapper.getTopic());
     };
 
     /**
@@ -152,40 +96,34 @@ public abstract class Subscribe implements WsHandler {
      * @param topic
      */
     public void unsubscribe(ChannelContext channelContext, String topic){
-        if(StrUtil.isEmpty(topic)){
-            return;
-        }
-        AtomicBoolean remove = new AtomicBoolean(false);
-        subscribeTopics.computeIfPresent(topic,(k,v)->{
-            v.remove(channelContext);
-//            if(CollectionUtil.isEmpty(v)){
-                remove.set(true);
-//                redisTemplate.execute(new RedisCallback<Object>() {
-//                    @Override
-//                    public Object doInRedis(RedisConnection connection) throws DataAccessException {
-//                        RedisConnection redisConnection = redisConnectionMap.get(topic);
-//                        if (redisConnection!=null) {
-//                            Optional.ofNullable(redisConnection.getSubscription())
-//                                    .map(subscription -> {
-//                                        subscription.pUnsubscribe(topic.getBytes());
-//                                        return subscription;
-//                                    });
-//                        }
-//                        return null;
-//                    }
-//                });
-//                redisConnectionMap.remove(topic);
-//            }
-            return v;
-        });
+//        if(StrUtil.isEmpty(topic)){
+//            return;
+//        }
+//        topicMap.computeIfPresent(topic,(k,rTopic)->{
+//            rTopic.removeListener( getTopicListener(channelContext,k));
+//            return rTopic;
+//        });
     };
 
 
-    public Set<String> getTopicByChannel(ChannelContext channelContext){
-        Object result = Optional.ofNullable(channelContext.get(SUBSCRIBE_TOPIC)).orElse(new HashSet<>());
-        channelContext.set(SUBSCRIBE_TOPIC,result);
-        return  (Set<String>) result;
+    public Map<String,DefaultMessageListener> getTopicListeners(ChannelContext channelContext){
+        Object result = Optional.ofNullable(channelContext.get("topic")).orElse(new HashMap<String,DefaultMessageListener>());
+        channelContext.set("topic",result);
+        return  (Map<String,DefaultMessageListener>) result;
     }
 
+    public DefaultMessageListener getTopicListener(ChannelContext channelContext,String topic){
+        Map<String,DefaultMessageListener> result = (Map<String, DefaultMessageListener>) Optional.ofNullable(channelContext.get("topic")).orElse(new HashMap<>());
+        return  result.get(topic);
+    }
 
+    public DefaultMessageListener addTopicListener(RPatternTopic rTopic,ChannelContext channelContext,String topic){
+        Map<String, DefaultMessageListener> topicByChannel = getTopicListeners(channelContext);
+        DefaultMessageListener messageListener = topicByChannel.computeIfAbsent(topic, k -> {
+            DefaultMessageListener defaultMessageListener = new DefaultMessageListener(getId(), objectMapper, channelContext,rTopic);
+            rTopic.addListener(TopicMessage.class, defaultMessageListener);
+            return defaultMessageListener;
+        });
+        return  messageListener;
+    }
 }

coffee-common/src/main/java/com/coffee/common/entity/GenericEntity.java

@@ -14,10 +14,11 @@ import org.springframework.validation.annotation.Validated;
 
 
 import javax.validation.constraints.NotNull;
+import java.io.Serializable;
 import java.util.Date;
 
 @Data
-public abstract class  GenericEntity<PK> implements Entity,RecordModifierEntity,RecordCreationEntity {
+public abstract class  GenericEntity<PK> implements Entity,RecordModifierEntity,RecordCreationEntity, Serializable {
     private static final long serialVersionUID = 1L;
 
     @TableId(type = IdType.ASSIGN_ID)

coffee-common/src/main/java/com/coffee/common/util/RedissonUtil.java

@@ -0,0 +1,46 @@
+package com.coffee.common.util;
+
+import cn.hutool.json.JSONUtil;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.redisson.api.RPatternTopic;
+import org.redisson.api.RTopic;
+import org.redisson.api.RedissonClient;
+import org.redisson.client.codec.Codec;
+import org.redisson.codec.JsonJacksonCodec;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Component;
+
+import javax.annotation.PostConstruct;
+
+/**
+ * @author lifang
+ * @version 1.0.0
+ * @ClassName RedissonUtil.java
+ * @Description TODO
+ * @createTime 2022年05月17日 16:16:00
+ */
+@Component
+public class RedissonUtil {
+    @Autowired
+    private RedissonClient redissonClient;
+    @Autowired
+    private ObjectMapper objectMapper;
+    private JsonJacksonCodec jsonObjectCodec;
+
+    @PostConstruct
+    public void init(){
+        jsonObjectCodec=new JsonJacksonCodec(objectMapper);
+    }
+
+    public RPatternTopic getPatternTopic(String pattern){
+        return redissonClient.getPatternTopic(pattern);
+    }
+
+    public RTopic getTopic(String pattern){
+        return redissonClient.getTopic(pattern);
+    }
+
+    public RTopic getTopic(String pattern, Codec codec){
+        return redissonClient.getTopic(pattern,codec);
+    }
+}

coffee-framework/pom.xml

@@ -47,5 +47,11 @@
             <groupId>com.coffee</groupId>
             <artifactId>coffee-system</artifactId>
         </dependency>
+
+        <dependency>
+            <groupId>org.redisson</groupId>
+            <artifactId>redisson-spring-boot-starter</artifactId>
+        </dependency>
+
     </dependencies>
 </project>

coffee-framework/src/main/java/com/coffee/framework/config/RedissonClientAutoConfiguration.java

@@ -0,0 +1,174 @@
+//
+// Source code recreated from a .class file by IntelliJ IDEA
+// (powered by Fernflower decompiler)
+//
+
+package com.coffee.framework.config;
+
+import org.redisson.Redisson;
+import org.redisson.api.RedissonClient;
+import org.redisson.config.ClusterServersConfig;
+import org.redisson.config.Config;
+import org.redisson.config.SentinelServersConfig;
+import org.redisson.config.SingleServerConfig;
+import org.redisson.spring.starter.RedissonAutoConfigurationCustomizer;
+import org.redisson.spring.starter.RedissonProperties;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.autoconfigure.AutoConfigureBefore;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
+import org.springframework.boot.autoconfigure.data.redis.RedisAutoConfiguration;
+import org.springframework.boot.autoconfigure.data.redis.RedisProperties;
+import org.springframework.boot.autoconfigure.data.redis.RedisProperties.Sentinel;
+import org.springframework.boot.context.properties.EnableConfigurationProperties;
+import org.springframework.context.ApplicationContext;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.context.annotation.Primary;
+import org.springframework.core.io.Resource;
+import org.springframework.data.redis.connection.RedisConnectionFactory;
+import org.springframework.data.redis.core.RedisOperations;
+import org.springframework.data.redis.core.RedisTemplate;
+import org.springframework.data.redis.core.StringRedisTemplate;
+import org.springframework.util.ReflectionUtils;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.lang.reflect.Method;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Iterator;
+import java.util.List;
+
+@Configuration
+@ConditionalOnClass({Redisson.class, RedisOperations.class})
+@AutoConfigureBefore({RedisAutoConfiguration.class})
+@EnableConfigurationProperties({RedissonProperties.class, RedisProperties.class})
+public class RedissonClientAutoConfiguration {
+    private static final String REDIS_PROTOCOL_PREFIX = "redis://";
+    private static final String REDISS_PROTOCOL_PREFIX = "rediss://";
+    @Autowired(
+        required = false
+    )
+    private List<RedissonAutoConfigurationCustomizer> redissonAutoConfigurationCustomizers;
+    @Autowired
+    private RedissonProperties redissonProperties;
+    @Autowired
+    private RedisProperties redisProperties;
+    @Autowired
+    private ApplicationContext ctx;
+
+    public RedissonClientAutoConfiguration() {
+    }
+
+    @Bean(
+        destroyMethod = "shutdown"
+    )
+    @ConditionalOnMissingBean({RedissonClient.class})
+    public RedissonClient redisson() throws IOException {
+        Config config = null;
+        Method clusterMethod = ReflectionUtils.findMethod(RedisProperties.class, "getCluster");
+        Method timeoutMethod = ReflectionUtils.findMethod(RedisProperties.class, "getTimeout");
+        Object timeoutValue = ReflectionUtils.invokeMethod(timeoutMethod, this.redisProperties);
+        int timeout;
+        Method nodesMethod;
+        if (null == timeoutValue) {
+            timeout = 10000;
+        } else if (!(timeoutValue instanceof Integer)) {
+            nodesMethod = ReflectionUtils.findMethod(timeoutValue.getClass(), "toMillis");
+            timeout = ((Long) ReflectionUtils.invokeMethod(nodesMethod, timeoutValue)).intValue();
+        } else {
+            timeout = (Integer)timeoutValue;
+        }
+
+        if (this.redissonProperties.getConfig() != null) {
+            try {
+                config = Config.fromYAML(this.redissonProperties.getConfig());
+            } catch (IOException var13) {
+                try {
+                    config = Config.fromJSON(this.redissonProperties.getConfig());
+                } catch (IOException var12) {
+                    throw new IllegalArgumentException("Can't parse config", var12);
+                }
+            }
+        } else if (this.redissonProperties.getFile() != null) {
+            try {
+                InputStream is = this.getConfigStream();
+                config = Config.fromYAML(is);
+            } catch (IOException var11) {
+                try {
+                    InputStream is = this.getConfigStream();
+                    config = Config.fromJSON(is);
+                } catch (IOException var10) {
+                    throw new IllegalArgumentException("Can't parse config", var10);
+                }
+            }
+        } else if (this.redisProperties.getSentinel() != null) {
+            nodesMethod = ReflectionUtils.findMethod(Sentinel.class, "getNodes");
+            Object nodesValue = ReflectionUtils.invokeMethod(nodesMethod, this.redisProperties.getSentinel());
+            String[] nodes;
+            if (nodesValue instanceof String) {
+                nodes = this.convert(Arrays.asList(((String)nodesValue).split(",")));
+            } else {
+                nodes = this.convert((List)nodesValue);
+            }
+
+            config = new Config();
+            ((SentinelServersConfig)config.useSentinelServers().setMasterName(this.redisProperties.getSentinel().getMaster()).addSentinelAddress(nodes).setDatabase(this.redisProperties.getDatabase()).setConnectTimeout(timeout)).setPassword(this.redisProperties.getPassword());
+        } else {
+            Method method;
+            if (clusterMethod != null && ReflectionUtils.invokeMethod(clusterMethod, this.redisProperties) != null) {
+                Object clusterObject = ReflectionUtils.invokeMethod(clusterMethod, this.redisProperties);
+                method = ReflectionUtils.findMethod(clusterObject.getClass(), "getNodes");
+                List<String> nodesObject = (List) ReflectionUtils.invokeMethod(method, clusterObject);
+                String[] nodes = this.convert(nodesObject);
+                config = new Config();
+                ((ClusterServersConfig)config.useClusterServers().addNodeAddress(nodes).setConnectTimeout(timeout)).setPassword(this.redisProperties.getPassword());
+            } else {
+                config = new Config();
+                String prefix = "redis://";
+                method = ReflectionUtils.findMethod(RedisProperties.class, "isSsl");
+                if (method != null && (Boolean) ReflectionUtils.invokeMethod(method, this.redisProperties)) {
+                    prefix = "rediss://";
+                }
+
+                ((SingleServerConfig)config.useSingleServer().setAddress(prefix + this.redisProperties.getHost() + ":" + this.redisProperties.getPort()).setConnectTimeout(timeout)).setDatabase(this.redisProperties.getDatabase()).setPassword(this.redisProperties.getPassword());
+            }
+        }
+
+        if (this.redissonAutoConfigurationCustomizers != null) {
+            Iterator var19 = this.redissonAutoConfigurationCustomizers.iterator();
+
+            while(var19.hasNext()) {
+                RedissonAutoConfigurationCustomizer customizer = (RedissonAutoConfigurationCustomizer)var19.next();
+                customizer.customize(config);
+            }
+        }
+
+        return Redisson.create(config);
+    }
+
+    private String[] convert(List<String> nodesObject) {
+        List<String> nodes = new ArrayList(nodesObject.size());
+        Iterator var3 = nodesObject.iterator();
+
+        while(true) {
+            while(var3.hasNext()) {
+                String node = (String)var3.next();
+                if (!node.startsWith("redis://") && !node.startsWith("rediss://")) {
+                    nodes.add("redis://" + node);
+                } else {
+                    nodes.add(node);
+                }
+            }
+
+            return (String[])nodes.toArray(new String[nodes.size()]);
+        }
+    }
+
+    private InputStream getConfigStream() throws IOException {
+        Resource resource = this.ctx.getResource(this.redissonProperties.getFile());
+        InputStream is = resource.getInputStream();
+        return is;
+    }
+}

coffee-system/src/main/java/com/coffee/bus/service/dto/ClinicAnalClinicRecord.java

@@ -10,6 +10,7 @@ import io.swagger.annotations.ApiModelProperty;
 import lombok.Data;
 import org.hibernate.validator.constraints.Length;
 
+import java.io.Serializable;
 import java.util.List;
 
 /**
@@ -21,7 +22,7 @@ import java.util.List;
  */
 @ApiModel("访视记录单中返回的临床数据")
 @Data
-public class ClinicAnalClinicRecord {
+public class ClinicAnalClinicRecord  implements Serializable {
     private String id;
 
     private String patientCode;

coffee-system/src/main/java/com/coffee/bus/service/dto/ClinicResult.java

@@ -6,6 +6,7 @@ import io.swagger.annotations.ApiModel;
 import io.swagger.annotations.ApiModelProperty;
 import lombok.Data;
 
+import java.io.Serializable;
 import java.util.Date;
 
 /**
@@ -17,7 +18,7 @@ import java.util.Date;
  */
 @Data
 @ApiModel("临床管理接口")
-public class ClinicResult {
+public class ClinicResult  implements Serializable {
     @ApiModelProperty("临床id")
     private String clinicId;
 

coffee-system/src/main/java/com/coffee/bus/service/dto/DeviceResult.java

@@ -9,6 +9,7 @@ import io.swagger.annotations.ApiModel;
 import io.swagger.annotations.ApiModelProperty;
 import lombok.Data;
 
+import java.io.Serializable;
 import java.util.Date;
 
 /**
@@ -20,7 +21,7 @@ import java.util.Date;
  */
 @Data
 @ApiModel("设备查询结果")
-public class DeviceResult {
+public class DeviceResult  implements Serializable {
     private String id;
     @ApiModelProperty(value = "设备唯一id")
     private String deviceId;

coffee-system/src/main/java/com/coffee/bus/service/dto/ManualMonitorResult.java

@@ -10,6 +10,7 @@ import io.swagger.annotations.ApiModelProperty;
 import lombok.Data;
 import org.hibernate.validator.constraints.Length;
 
+import java.io.Serializable;
 import java.util.Date;
 import java.util.List;
 
@@ -22,7 +23,7 @@ import java.util.List;
  */
 @Data
 @ApiModel("其他监护查询返回结果")
-public class ManualMonitorResult {
+public class ManualMonitorResult  implements Serializable {
     @ApiModelProperty(value = "临床手术id")
     private String clinicId;
 

coffee-system/src/main/java/com/coffee/bus/service/dto/MonitorStatusStatsCountResult.java

@@ -4,6 +4,8 @@ import io.swagger.annotations.ApiModel;
 import io.swagger.annotations.ApiModelProperty;
 import lombok.Data;
 
+import java.io.Serializable;
+
 /**
  * @author lifang
  * @version 1.0.0
@@ -13,7 +15,7 @@ import lombok.Data;
  */
 @ApiModel("监控状态数量统计结果")
 @Data
-public class MonitorStatusStatsCountResult {
+public class MonitorStatusStatsCountResult  implements Serializable {
 
     @ApiModelProperty("运行数量")
     private int runningCount;

coffee-system/src/main/java/com/coffee/bus/service/dto/MonitorTimeStatsCountResult.java

@@ -4,6 +4,8 @@ import io.swagger.annotations.ApiModel;
 import io.swagger.annotations.ApiModelProperty;
 import lombok.Data;
 
+import java.io.Serializable;
+
 /**
  * @author lifang
  * @version 1.0.0
@@ -13,7 +15,7 @@ import lombok.Data;
  */
 @ApiModel("监控数量按照时间统计结果")
 @Data
-public class MonitorTimeStatsCountResult {
+public class MonitorTimeStatsCountResult  implements Serializable {
 
     @ApiModelProperty("当天")
     private int today;

coffee-system/src/main/java/com/coffee/bus/service/dto/PatientDeviceNoneResult.java

@@ -7,6 +7,8 @@ import io.swagger.annotations.ApiModel;
 import io.swagger.annotations.ApiModelProperty;
 import lombok.Data;
 
+import java.io.Serializable;
+
 /**
  * @author lifang
  * @version 1.0.0
@@ -16,7 +18,7 @@ import lombok.Data;
  */
 @Data
 @ApiModel("病人无设备返回数据")
-public class PatientDeviceNoneResult {
+public class PatientDeviceNoneResult  implements Serializable {
     @ApiModelProperty("病患名称")
     private String name;
 

coffee-system/src/main/java/com/coffee/bus/service/dto/PatientDeviceRepeatResult.java

@@ -10,6 +10,7 @@ import io.swagger.annotations.ApiOperation;
 import lombok.AllArgsConstructor;
 import lombok.Data;
 
+import java.io.Serializable;
 import java.util.*;
 
 /**
@@ -22,7 +23,7 @@ import java.util.*;
 @Data
 @ApiModel("病人重复设备返回数据")
 @AllArgsConstructor(staticName = "of")
-public class PatientDeviceRepeatResult {
+public class PatientDeviceRepeatResult  implements Serializable {
     @ApiModelProperty("病患名称")
     private String name;
 
@@ -53,7 +54,7 @@ public class PatientDeviceRepeatResult {
     @ApiModel("设备运行小数据集合")
     @Data
     @AllArgsConstructor(staticName = "of")
-    public static class DeviceRunningSmallInfo{
+    public static class DeviceRunningSmallInfo  implements Serializable {
         @ApiModelProperty("设备运行Id")
         private String deviceRunningId;
         @ApiModelProperty("设备名称")

coffee-system/src/main/java/com/coffee/bus/service/dto/PatientMonitorDetailResult.java

@@ -11,6 +11,7 @@ import lombok.Data;
 import lombok.NoArgsConstructor;
 
 import javax.validation.constraints.NotNull;
+import java.io.Serializable;
 
 /**
  * @author lifang
@@ -23,7 +24,7 @@ import javax.validation.constraints.NotNull;
 @Data
 @AllArgsConstructor
 @NoArgsConstructor
-public class PatientMonitorDetailResult {
+public class PatientMonitorDetailResult  implements Serializable {
     @ApiModelProperty("最后输注信息")
     private BusInfusionHistoryEntity infusion;
     @ApiModelProperty("手术信息")

coffee-system/src/main/java/com/coffee/bus/service/dto/PatientMonitorQuery.java

@@ -10,6 +10,7 @@ import io.swagger.annotations.ApiModel;
 import io.swagger.annotations.ApiModelProperty;
 import lombok.Data;
 
+import java.io.Serializable;
 import java.sql.Timestamp;
 import java.util.*;
 /**
@@ -21,7 +22,7 @@ import java.util.*;
  */
 @Data
 @ApiModel("病人监控查询参数")
-public class PatientMonitorQuery {
+public class PatientMonitorQuery  implements Serializable {
     @ApiModelProperty("模糊查询,床号、姓名、住院号,WEB端专用字段，安卓端请看下方")
     private String blurry;
 

coffee-system/src/main/java/com/coffee/bus/service/dto/PatientMonitorResult.java

@@ -8,6 +8,7 @@ import io.swagger.annotations.ApiModel;
 import io.swagger.annotations.ApiModelProperty;
 import lombok.Data;
 
+import java.io.Serializable;
 import java.math.BigDecimal;
 import java.util.Date;
 
@@ -20,7 +21,7 @@ import java.util.Date;
  */
 @ApiModel("病人监控实体")
 @Data
-public class PatientMonitorResult {
+public class PatientMonitorResult implements Serializable {
     @ApiModelProperty(value = "病号")
     private String patientCode;
 

coffee-system/src/main/java/com/coffee/bus/utils/WsPublishUtils.java

@@ -1,7 +1,9 @@
 package com.coffee.bus.utils;
 
 
+import cn.hutool.core.util.ObjectUtil;
 import cn.hutool.json.JSONObject;
+import com.alibaba.fastjson.serializer.JSONObjectCodec;
 import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
 import com.coffee.bus.entity.BusClinicEntity;
 import com.coffee.bus.entity.BusPatientEntity;
@@ -10,9 +12,18 @@ import com.coffee.bus.service.LocalBusClinicService;
 import com.coffee.bus.service.LocalBusPatientService;
 import com.coffee.bus.service.dto.MonitorStatusStatsCountResult;
 import com.coffee.bus.service.dto.PatientMonitorResult;
+import com.coffee.common.config.websocket.TopicMessage;
 import com.coffee.common.config.websocket.WebSocketConstant;
+import com.coffee.common.redis.RedisUtils;
+import com.coffee.common.util.RedissonUtil;
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.ObjectMapper;
 import lombok.AllArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
+import org.redisson.RedissonTopic;
+import org.redisson.api.RedissonClient;
+import org.redisson.codec.JsonJacksonCodec;
+import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.data.redis.core.RedisTemplate;
 import org.springframework.scheduling.annotation.Async;
 import org.springframework.stereotype.Component;
@@ -30,13 +41,13 @@ import java.io.Serializable;
 @Component
 @AllArgsConstructor
 @Slf4j
-public class WsPublishUtils {
-    private final RedisTemplate redisTemplate;
+public class WsPublishUtils implements Serializable{
     private final LocalBusPatientService patientService;
     private final LocalBusClinicService clinicService;
+    private final RedissonUtil redissonUtil;
 
     private void publish(String topic,TopicMessage msg){
-        redisTemplate.convertAndSend(topic,msg);
+        redissonUtil.getTopic(topic).publish(msg);
     }
 
     /**
@@ -52,16 +63,7 @@ public class WsPublishUtils {
         Assert.hasText(tenantId,"医院id不能为空");
         PatientMonitorResult message = patientService.lookMonitorByPatientCode(patientCode, tenantId);
         this.publish(WebSocketConstant.getPatientMonitor(null, patientCode, tenantId).getTopic(),
-                new TopicMessage<PatientMonitorResult>() {
-                    @Override
-                    public PatientMonitorResult getMessage() {
-                        return message;
-                    }
-                    @Override
-                    public String getParam() {
-                        return patientCode;
-                    }
-                }
+                TopicMessage.of(message,patientCode)
         );
     }
 
@@ -76,16 +78,9 @@ public class WsPublishUtils {
     public void publishMonitorStateCount(String tenantId){
         Assert.hasText(tenantId,"医院id不能为空");
         MonitorStatusStatsCountResult message = patientService.statusStats(tenantId);
-        this.publish(WebSocketConstant.getMonitorStateCount(tenantId).getTopic(),  new TopicMessage<MonitorStatusStatsCountResult>() {
-            @Override
-            public MonitorStatusStatsCountResult getMessage() {
-                return message;
-            }
-            @Override
-            public String getParam() {
-                return tenantId;
-            }
-        });
+        this.publish(WebSocketConstant.getMonitorStateCount(tenantId).getTopic(),
+                TopicMessage.of(message,tenantId)
+        );
     }
 
     /**
@@ -100,16 +95,7 @@ public class WsPublishUtils {
         Assert.hasText(tenantId,"医院id不能为空");
         JSONObject message = new JSONObject().putOpt("patientCode", patientCode);
         this.publish(WebSocketConstant.getPatientAdd(tenantId).getTopic(),
-                new TopicMessage<JSONObject>() {
-                    @Override
-                    public JSONObject getMessage() {
-                        return message;
-                    }
-                    @Override
-                    public String getParam() {
-                        return tenantId;
-                    }
-                });
+                TopicMessage.of(message,tenantId));
     }
     /**
      * 描述: 推送医院临床设备重复数量统计
@@ -124,16 +110,7 @@ public class WsPublishUtils {
         JSONObject message = new JSONObject().putOpt("count", patientService.count(new QueryWrapper<BusPatientEntity>().lambda().eq(BusPatientEntity::getTenantId, tenantId)
                 .eq(BusPatientEntity::getAlarm, PatientAlarmEnum.DEVICE_REPEAT)));
         this.publish(WebSocketConstant.getDeviceRepeat(tenantId).getTopic(),
-                new TopicMessage<JSONObject>() {
-                    @Override
-                    public JSONObject getMessage() {
-                        return message;
-                    }
-                    @Override
-                    public String getParam() {
-                        return tenantId;
-                    }
-                }
+                TopicMessage.of(message,tenantId)
         );
     }
 
@@ -151,16 +128,7 @@ public class WsPublishUtils {
                 .lambda().eq(BusPatientEntity::getTenantId, tenantId)
                 .eq(BusPatientEntity::getAlarm, PatientAlarmEnum.DEVICE_NONE)));
         this.publish(WebSocketConstant.getDeviceNone(tenantId).getTopic(),
-                new TopicMessage<JSONObject>() {
-                    @Override
-                    public JSONObject getMessage() {
-                        return message;
-                    }
-                    @Override
-                    public String getParam() {
-                        return tenantId;
-                    }
-                }
+                TopicMessage.of(message,tenantId)
         );
     }
 
@@ -180,21 +148,9 @@ public class WsPublishUtils {
                 .eq(BusClinicEntity::getTenantId,tenantId)
                 .eq(BusClinicEntity::getFinished,false)));
         this.publish(WebSocketConstant.getMonitorTotalCount(tenantId).getTopic(),
-                new TopicMessage<JSONObject>() {
-                    @Override
-                    public JSONObject getMessage() {
-                        return message;
-                    }
-                    @Override
-                    public String getParam() {
-                        return tenantId;
-                    }
-                }
+                TopicMessage.of(message,tenantId)
         );
     }
 
-    public interface TopicMessage<T> extends Serializable {
-        T getMessage();
-        String getParam();
-    }
+
 }

coffee-system/src/main/java/com/coffee/bus/websocket/WebSocketCloseHandler.java

@@ -1,14 +1,11 @@
 package com.coffee.bus.websocket;
 
-import com.coffee.common.config.websocket.MessagingRequest;
+import com.coffee.common.config.websocket.DefaultMessageListener;
 import com.coffee.common.config.websocket.handler.Subscribe;
-import com.coffee.common.config.websocket.handler.WsHandler;
 import org.springframework.stereotype.Component;
 import org.tio.core.ChannelContext;
 
-import java.util.HashSet;
-import java.util.Optional;
-import java.util.Set;
+import java.util.Map;
 
 /**
  * @author lifang
@@ -27,8 +24,8 @@ public class WebSocketCloseHandler extends Subscribe {
     @Override
     public void close(ChannelContext channelContext) {
         //关闭则取消订阅
-        Set<String> topicByChannel = getTopicByChannel(channelContext);
-        topicByChannel.forEach(topic->this.unsubscribe(channelContext,topic));
+        Map<String, DefaultMessageListener> topicListeners = getTopicListeners(channelContext);
+        topicListeners.forEach((topic,listener)->this.unsubscribe(channelContext,topic));
     }
 
     @Override

pom.xml

@@ -37,6 +37,7 @@
         <tio.version>3.6.0.v20200315-RELEASE</tio.version>
         <jython.version>2.7.1</jython.version>
         <knife4j.version>2.0.7</knife4j.version>
+        <redisson.version>3.13.6</redisson.version>
     </properties>
 
     <modules>
@@ -209,6 +210,20 @@
                 <artifactId>coffee-oss</artifactId>
                 <version>${coffee.project.version}</version>
             </dependency>
+
+            <!-- redisson配置 -->
+            <dependency>
+                <groupId>org.redisson</groupId>
+                <artifactId>redisson-spring-boot-starter</artifactId>
+                <version>${redisson.version}</version>
+                <exclusions>
+                    <exclusion>
+                        <groupId>org.springframework.boot</groupId>
+                        <artifactId>spring-boot-starter-web</artifactId>
+                    </exclusion>
+                </exclusions>
+            </dependency>
+
         </dependencies>
     </dependencyManagement>