diff --git a/casic-metering-common/pom.xml b/casic-metering-common/pom.xml index 142b343..9818581 100644 --- a/casic-metering-common/pom.xml +++ b/casic-metering-common/pom.xml @@ -185,6 +185,12 @@ org.springframework.boot + spring-boot-starter-websocket + 2.4.5 + + + + org.springframework.boot spring-boot-starter-data-redis diff --git a/casic-metering-common/pom.xml b/casic-metering-common/pom.xml index 142b343..9818581 100644 --- a/casic-metering-common/pom.xml +++ b/casic-metering-common/pom.xml @@ -185,6 +185,12 @@ org.springframework.boot + spring-boot-starter-websocket + 2.4.5 + + + + org.springframework.boot spring-boot-starter-data-redis diff --git a/casic-metering-common/src/main/java/com/casic/missiles/config/WebSocketConfig.java b/casic-metering-common/src/main/java/com/casic/missiles/config/WebSocketConfig.java new file mode 100644 index 0000000..9c9c4b5 --- /dev/null +++ b/casic-metering-common/src/main/java/com/casic/missiles/config/WebSocketConfig.java @@ -0,0 +1,18 @@ +package com.casic.missiles.config; + +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.web.socket.server.standard.ServerEndpointExporter; + +// tomcat启动无需配置 +@Configuration +public class WebSocketConfig { + /** + * 注入ServerEndpointExporter， + * 这个bean会自动注册使用了@ServerEndpoint注解声明的Websocket endpoint + */ + @Bean + public ServerEndpointExporter serverEndpointExporter() { + return new ServerEndpointExporter(); + } +} diff --git a/casic-metering-common/pom.xml b/casic-metering-common/pom.xml index 142b343..9818581 100644 --- a/casic-metering-common/pom.xml +++ b/casic-metering-common/pom.xml @@ -185,6 +185,12 @@ org.springframework.boot + spring-boot-starter-websocket + 2.4.5 + + + + org.springframework.boot spring-boot-starter-data-redis diff --git a/casic-metering-common/src/main/java/com/casic/missiles/config/WebSocketConfig.java b/casic-metering-common/src/main/java/com/casic/missiles/config/WebSocketConfig.java new file mode 100644 index 0000000..9c9c4b5 --- /dev/null +++ b/casic-metering-common/src/main/java/com/casic/missiles/config/WebSocketConfig.java @@ -0,0 +1,18 @@ +package com.casic.missiles.config; + +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.web.socket.server.standard.ServerEndpointExporter; + +// tomcat启动无需配置 +@Configuration +public class WebSocketConfig { + /** + * 注入ServerEndpointExporter， + * 这个bean会自动注册使用了@ServerEndpoint注解声明的Websocket endpoint + */ + @Bean + public ServerEndpointExporter serverEndpointExporter() { + return new ServerEndpointExporter(); + } +} diff --git a/casic-metering-common/src/main/java/com/casic/missiles/utils/WebSocket.java b/casic-metering-common/src/main/java/com/casic/missiles/utils/WebSocket.java new file mode 100644 index 0000000..f61a7ff --- /dev/null +++ b/casic-metering-common/src/main/java/com/casic/missiles/utils/WebSocket.java @@ -0,0 +1,99 @@ +package com.casic.missiles.utils; + +import lombok.extern.slf4j.Slf4j; +import org.springframework.stereotype.Component; + +import javax.websocket.OnClose; +import javax.websocket.OnMessage; +import javax.websocket.OnOpen; +import javax.websocket.Session; +import javax.websocket.server.PathParam; +import javax.websocket.server.ServerEndpoint; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.concurrent.CopyOnWriteArraySet; + +/** + * @Description: Websocket ServerEndpoint + * @Author: wangpeng + * @Date: 2023/2/23 11:48 + */ +@Slf4j +@Component +@ServerEndpoint("/websocket/{userId}") +public class WebSocket { + private Session session; + + private static CopyOnWriteArraySet webSockets = new CopyOnWriteArraySet<>(); + private static Map sessionPool = new HashMap<>(); + + @OnOpen + public void onOpen(Session session, @PathParam(value = "userId") String userId) { + this.session = session; + webSockets.add(this); + sessionPool.put(userId, session); + log.info(userId + "【websocket消息】有新的连接，总数为:" + webSockets.size()); +// System.out.println(userId + "【websocket消息】有新的连接，总数为:" + webSockets.size()); + } + + @OnClose + public void onClose() { + webSockets.remove(this); + log.info("【websocket消息】连接断开，总数为:" + webSockets.size()); +// System.out.println("【websocket消息】连接断开，总数为:" + webSockets.size()); + } + + @OnMessage + public void onMessage(String message) { + log.info("【websocket消息】收到客户端消息:" + message); +// System.out.println("【websocket消息】收到客户端消息:" + message); + } + + // 此为广播消息 + public void sendAllMessage(String message) { + for (WebSocket webSocket : webSockets) { + log.info("【websocket消息】广播消息:" + message); +// System.out.println("【websocket消息】广播消息:" + message); + try { + webSocket.session.getAsyncRemote().sendText(message); + } catch (Exception e) { + log.error("【websocket消息】广播消息发送异常"); + e.printStackTrace(); + } + } + } + + // 发送列表消息 + public void sendListMessage(List userIds, String message) { + log.info("【websocket消息】列表消息:" + message); +// System.out.println("【websocket消息】列表消息:" + message); + for (String userId : userIds) { + Session session = sessionPool.get(userId); + if (session != null) { + try { + session.getAsyncRemote().sendText(message); + } catch (Exception e) { + log.error("【websocket消息】列表消息发送异常"); + e.printStackTrace(); + } + } + } + } + + // 此为单点消息 + public void sendOneMessage(String userId, String message) { + log.info("【websocket消息】单点消息:" + message); +// System.out.println("【websocket消息】单点消息:" + message); + Session session = sessionPool.get(userId); + if (session != null) { + try { + session.getAsyncRemote().sendText(message); + } catch (Exception e) { + log.error("【websocket消息】单点消息发送异常"); + e.printStackTrace(); + } + } + } + +} diff --git a/casic-metering-common/pom.xml b/casic-metering-common/pom.xml index 142b343..9818581 100644 --- a/casic-metering-common/pom.xml +++ b/casic-metering-common/pom.xml @@ -185,6 +185,12 @@ org.springframework.boot + spring-boot-starter-websocket + 2.4.5 + + + + org.springframework.boot spring-boot-starter-data-redis diff --git a/casic-metering-common/src/main/java/com/casic/missiles/config/WebSocketConfig.java b/casic-metering-common/src/main/java/com/casic/missiles/config/WebSocketConfig.java new file mode 100644 index 0000000..9c9c4b5 --- /dev/null +++ b/casic-metering-common/src/main/java/com/casic/missiles/config/WebSocketConfig.java @@ -0,0 +1,18 @@ +package com.casic.missiles.config; + +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.web.socket.server.standard.ServerEndpointExporter; + +// tomcat启动无需配置 +@Configuration +public class WebSocketConfig { + /** + * 注入ServerEndpointExporter， + * 这个bean会自动注册使用了@ServerEndpoint注解声明的Websocket endpoint + */ + @Bean + public ServerEndpointExporter serverEndpointExporter() { + return new ServerEndpointExporter(); + } +} diff --git a/casic-metering-common/src/main/java/com/casic/missiles/utils/WebSocket.java b/casic-metering-common/src/main/java/com/casic/missiles/utils/WebSocket.java new file mode 100644 index 0000000..f61a7ff --- /dev/null +++ b/casic-metering-common/src/main/java/com/casic/missiles/utils/WebSocket.java @@ -0,0 +1,99 @@ +package com.casic.missiles.utils; + +import lombok.extern.slf4j.Slf4j; +import org.springframework.stereotype.Component; + +import javax.websocket.OnClose; +import javax.websocket.OnMessage; +import javax.websocket.OnOpen; +import javax.websocket.Session; +import javax.websocket.server.PathParam; +import javax.websocket.server.ServerEndpoint; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.concurrent.CopyOnWriteArraySet; + +/** + * @Description: Websocket ServerEndpoint + * @Author: wangpeng + * @Date: 2023/2/23 11:48 + */ +@Slf4j +@Component +@ServerEndpoint("/websocket/{userId}") +public class WebSocket { + private Session session; + + private static CopyOnWriteArraySet webSockets = new CopyOnWriteArraySet<>(); + private static Map sessionPool = new HashMap<>(); + + @OnOpen + public void onOpen(Session session, @PathParam(value = "userId") String userId) { + this.session = session; + webSockets.add(this); + sessionPool.put(userId, session); + log.info(userId + "【websocket消息】有新的连接，总数为:" + webSockets.size()); +// System.out.println(userId + "【websocket消息】有新的连接，总数为:" + webSockets.size()); + } + + @OnClose + public void onClose() { + webSockets.remove(this); + log.info("【websocket消息】连接断开，总数为:" + webSockets.size()); +// System.out.println("【websocket消息】连接断开，总数为:" + webSockets.size()); + } + + @OnMessage + public void onMessage(String message) { + log.info("【websocket消息】收到客户端消息:" + message); +// System.out.println("【websocket消息】收到客户端消息:" + message); + } + + // 此为广播消息 + public void sendAllMessage(String message) { + for (WebSocket webSocket : webSockets) { + log.info("【websocket消息】广播消息:" + message); +// System.out.println("【websocket消息】广播消息:" + message); + try { + webSocket.session.getAsyncRemote().sendText(message); + } catch (Exception e) { + log.error("【websocket消息】广播消息发送异常"); + e.printStackTrace(); + } + } + } + + // 发送列表消息 + public void sendListMessage(List userIds, String message) { + log.info("【websocket消息】列表消息:" + message); +// System.out.println("【websocket消息】列表消息:" + message); + for (String userId : userIds) { + Session session = sessionPool.get(userId); + if (session != null) { + try { + session.getAsyncRemote().sendText(message); + } catch (Exception e) { + log.error("【websocket消息】列表消息发送异常"); + e.printStackTrace(); + } + } + } + } + + // 此为单点消息 + public void sendOneMessage(String userId, String message) { + log.info("【websocket消息】单点消息:" + message); +// System.out.println("【websocket消息】单点消息:" + message); + Session session = sessionPool.get(userId); + if (session != null) { + try { + session.getAsyncRemote().sendText(message); + } catch (Exception e) { + log.error("【websocket消息】单点消息发送异常"); + e.printStackTrace(); + } + } + } + +} diff --git a/casic-metering-dao/src/main/resources/mapper/workbench/WorkbenchMessageMapper.xml b/casic-metering-dao/src/main/resources/mapper/workbench/WorkbenchMessageMapper.xml index 627f8e5..707df30 100644 --- a/casic-metering-dao/src/main/resources/mapper/workbench/WorkbenchMessageMapper.xml +++ b/casic-metering-dao/src/main/resources/mapper/workbench/WorkbenchMessageMapper.xml @@ -5,12 +5,12 @@ - - - + + - + + @@ -21,7 +21,7 @@ - id, user_id, dept_id, message_type, business_id, business_code, message_topic, message_content, status, create_user, create_user_name, create_time, update_time + id, remind_id, remind_dept_id, business_id, business_code, message_type, message_title, message_content, status, create_user, create_user_name, create_time, update_time diff --git a/casic-metering-common/pom.xml b/casic-metering-common/pom.xml index 142b343..9818581 100644 --- a/casic-metering-common/pom.xml +++ b/casic-metering-common/pom.xml @@ -185,6 +185,12 @@ org.springframework.boot + spring-boot-starter-websocket + 2.4.5 + + + + org.springframework.boot spring-boot-starter-data-redis diff --git a/casic-metering-common/src/main/java/com/casic/missiles/config/WebSocketConfig.java b/casic-metering-common/src/main/java/com/casic/missiles/config/WebSocketConfig.java new file mode 100644 index 0000000..9c9c4b5 --- /dev/null +++ b/casic-metering-common/src/main/java/com/casic/missiles/config/WebSocketConfig.java @@ -0,0 +1,18 @@ +package com.casic.missiles.config; + +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.web.socket.server.standard.ServerEndpointExporter; + +// tomcat启动无需配置 +@Configuration +public class WebSocketConfig { + /** + * 注入ServerEndpointExporter， + * 这个bean会自动注册使用了@ServerEndpoint注解声明的Websocket endpoint + */ + @Bean + public ServerEndpointExporter serverEndpointExporter() { + return new ServerEndpointExporter(); + } +} diff --git a/casic-metering-common/src/main/java/com/casic/missiles/utils/WebSocket.java b/casic-metering-common/src/main/java/com/casic/missiles/utils/WebSocket.java new file mode 100644 index 0000000..f61a7ff --- /dev/null +++ b/casic-metering-common/src/main/java/com/casic/missiles/utils/WebSocket.java @@ -0,0 +1,99 @@ +package com.casic.missiles.utils; + +import lombok.extern.slf4j.Slf4j; +import org.springframework.stereotype.Component; + +import javax.websocket.OnClose; +import javax.websocket.OnMessage; +import javax.websocket.OnOpen; +import javax.websocket.Session; +import javax.websocket.server.PathParam; +import javax.websocket.server.ServerEndpoint; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.concurrent.CopyOnWriteArraySet; + +/** + * @Description: Websocket ServerEndpoint + * @Author: wangpeng + * @Date: 2023/2/23 11:48 + */ +@Slf4j +@Component +@ServerEndpoint("/websocket/{userId}") +public class WebSocket { + private Session session; + + private static CopyOnWriteArraySet webSockets = new CopyOnWriteArraySet<>(); + private static Map sessionPool = new HashMap<>(); + + @OnOpen + public void onOpen(Session session, @PathParam(value = "userId") String userId) { + this.session = session; + webSockets.add(this); + sessionPool.put(userId, session); + log.info(userId + "【websocket消息】有新的连接，总数为:" + webSockets.size()); +// System.out.println(userId + "【websocket消息】有新的连接，总数为:" + webSockets.size()); + } + + @OnClose + public void onClose() { + webSockets.remove(this); + log.info("【websocket消息】连接断开，总数为:" + webSockets.size()); +// System.out.println("【websocket消息】连接断开，总数为:" + webSockets.size()); + } + + @OnMessage + public void onMessage(String message) { + log.info("【websocket消息】收到客户端消息:" + message); +// System.out.println("【websocket消息】收到客户端消息:" + message); + } + + // 此为广播消息 + public void sendAllMessage(String message) { + for (WebSocket webSocket : webSockets) { + log.info("【websocket消息】广播消息:" + message); +// System.out.println("【websocket消息】广播消息:" + message); + try { + webSocket.session.getAsyncRemote().sendText(message); + } catch (Exception e) { + log.error("【websocket消息】广播消息发送异常"); + e.printStackTrace(); + } + } + } + + // 发送列表消息 + public void sendListMessage(List userIds, String message) { + log.info("【websocket消息】列表消息:" + message); +// System.out.println("【websocket消息】列表消息:" + message); + for (String userId : userIds) { + Session session = sessionPool.get(userId); + if (session != null) { + try { + session.getAsyncRemote().sendText(message); + } catch (Exception e) { + log.error("【websocket消息】列表消息发送异常"); + e.printStackTrace(); + } + } + } + } + + // 此为单点消息 + public void sendOneMessage(String userId, String message) { + log.info("【websocket消息】单点消息:" + message); +// System.out.println("【websocket消息】单点消息:" + message); + Session session = sessionPool.get(userId); + if (session != null) { + try { + session.getAsyncRemote().sendText(message); + } catch (Exception e) { + log.error("【websocket消息】单点消息发送异常"); + e.printStackTrace(); + } + } + } + +} diff --git a/casic-metering-dao/src/main/resources/mapper/workbench/WorkbenchMessageMapper.xml b/casic-metering-dao/src/main/resources/mapper/workbench/WorkbenchMessageMapper.xml index 627f8e5..707df30 100644 --- a/casic-metering-dao/src/main/resources/mapper/workbench/WorkbenchMessageMapper.xml +++ b/casic-metering-dao/src/main/resources/mapper/workbench/WorkbenchMessageMapper.xml @@ -5,12 +5,12 @@ - - - + + - + + @@ -21,7 +21,7 @@ - id, user_id, dept_id, message_type, business_id, business_code, message_topic, message_content, status, create_user, create_user_name, create_time, update_time + id, remind_id, remind_dept_id, business_id, business_code, message_type, message_title, message_content, status, create_user, create_user_name, create_time, update_time diff --git a/casic-metering-model/src/main/java/com/casic/missiles/model/workbench/WorkbenchMessage.java b/casic-metering-model/src/main/java/com/casic/missiles/model/workbench/WorkbenchMessage.java index 5626080..563ae8b 100644 --- a/casic-metering-model/src/main/java/com/casic/missiles/model/workbench/WorkbenchMessage.java +++ b/casic-metering-model/src/main/java/com/casic/missiles/model/workbench/WorkbenchMessage.java @@ -27,20 +27,14 @@ /** * 消息提醒的用户id */ - @TableField("user_id") - private Long userId; + @TableField("remind_id") + private Long remindId; /** * 消息提醒的用户部门id */ - @TableField("dept_id") - private Long deptId; - - /** - * 消息提醒的业务类型 - */ - @TableField("message_type") - private String messageType; + @TableField("remind_dept_id") + private Long remindDeptId; /** * 消息提醒的业务表id @@ -55,10 +49,16 @@ private String businessCode; /** + * 消息提醒的业务类型 + */ + @TableField("message_type") + private String messageType; + + /** * 消息主题（冗余） */ - @TableField("message_topic") - private String messageTopic; + @TableField("message_title") + private String messageTitle; /** * 消息内容（消息提醒的业务内容名称）