|
|
@@ -0,0 +1,213 @@
|
|
|
+package com.qmth.themis.backend.websocket;
|
|
|
+
|
|
|
+import com.alibaba.fastjson.JSON;
|
|
|
+import com.alibaba.fastjson.JSONObject;
|
|
|
+import org.apache.commons.lang3.StringUtils;
|
|
|
+import org.slf4j.Logger;
|
|
|
+import org.slf4j.LoggerFactory;
|
|
|
+import org.springframework.stereotype.Component;
|
|
|
+
|
|
|
+import javax.websocket.*;
|
|
|
+import javax.websocket.server.PathParam;
|
|
|
+import javax.websocket.server.ServerEndpoint;
|
|
|
+import java.io.IOException;
|
|
|
+import java.util.concurrent.ConcurrentHashMap;
|
|
|
+
|
|
|
+/**
|
|
|
+ * @Description: websocker服务端
|
|
|
+ * @Param:
|
|
|
+ * @return:
|
|
|
+ * @Author: wangliang
|
|
|
+ * @Date: 2020/7/10
|
|
|
+ */
|
|
|
+@ServerEndpoint("/imserver/{userId}")
|
|
|
+@Component
|
|
|
+public class WebSocketServer
|
|
|
+// implements MessageListenerConcurrently
|
|
|
+{
|
|
|
+ private final static Logger log = LoggerFactory.getLogger(WebSocketServer.class);
|
|
|
+ /**
|
|
|
+ * 静态变量，用来记录当前在线连接数。应该把它设计成线程安全的。
|
|
|
+ */
|
|
|
+ private static int onlineCount = 0;
|
|
|
+ /**
|
|
|
+ * concurrent包的线程安全Set，用来存放每个客户端对应的MyWebSocket对象。
|
|
|
+ */
|
|
|
+ private static ConcurrentHashMap<String, WebSocketServer> webSocketMap = new ConcurrentHashMap<>();
|
|
|
+ /**
|
|
|
+ * 与某个客户端的连接会话，需要通过它来给客户端发送数据
|
|
|
+ */
|
|
|
+ private Session session;
|
|
|
+ /**
|
|
|
+ * 接收userId
|
|
|
+ */
|
|
|
+ private String userId = null;
|
|
|
+
|
|
|
+ /**
|
|
|
+ * 连接建立成功调用的方法
|
|
|
+ */
|
|
|
+ @OnOpen
|
|
|
+ public void onOpen(Session session, @PathParam("userId") String userId) {
|
|
|
+ this.session = session;
|
|
|
+ this.userId = userId;
|
|
|
+ if (webSocketMap.containsKey(userId)) {
|
|
|
+ webSocketMap.remove(userId);
|
|
|
+ webSocketMap.put(userId, this);
|
|
|
+ //加入set中
|
|
|
+ } else {
|
|
|
+ webSocketMap.put(userId, this);
|
|
|
+ //加入set中
|
|
|
+ addOnlineCount();
|
|
|
+ //在线数加1
|
|
|
+ }
|
|
|
+ log.info("用户连接:" + userId + ",当前在线人数为:" + getOnlineCount());
|
|
|
+ try {
|
|
|
+ sendMessage("连接成功");
|
|
|
+ } catch (IOException e) {
|
|
|
+ log.error("用户:" + userId + ",网络异常!!!!!!");
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+ /**
|
|
|
+ * 连接关闭调用的方法
|
|
|
+ */
|
|
|
+ @OnClose
|
|
|
+ public void onClose() {
|
|
|
+ if (webSocketMap.containsKey(userId)) {
|
|
|
+ webSocketMap.remove(userId);
|
|
|
+ //从set中删除
|
|
|
+ subOnlineCount();
|
|
|
+ }
|
|
|
+ log.info("用户退出:" + userId + ",当前在线人数为:" + getOnlineCount());
|
|
|
+ }
|
|
|
+
|
|
|
+ /**
|
|
|
+ * 收到客户端消息后调用的方法
|
|
|
+ *
|
|
|
+ * @param message 客户端发送过来的消息
|
|
|
+ */
|
|
|
+ @OnMessage
|
|
|
+ public void onMessage(String message, Session session) {
|
|
|
+ log.info("用户消息:" + userId + ",报文:" + message);
|
|
|
+ //可以群发消息
|
|
|
+ //消息保存到数据库、redis
|
|
|
+ if (StringUtils.isNotBlank(message)) {
|
|
|
+ try {
|
|
|
+ //解析发送的报文
|
|
|
+ JSONObject jsonObject = JSON.parseObject(message);
|
|
|
+ //追加发送人(防止串改)
|
|
|
+ jsonObject.put("fromUserId", this.userId);
|
|
|
+ String toUserId = jsonObject.getString("toUserId");
|
|
|
+ //传送给对应toUserId用户的websocket
|
|
|
+ if (StringUtils.isNotBlank(toUserId) && webSocketMap.containsKey(toUserId)) {
|
|
|
+ webSocketMap.get(toUserId).sendMessage(jsonObject.toJSONString());
|
|
|
+ } else {
|
|
|
+ log.error("请求的userId:" + toUserId + "不在该服务器上");
|
|
|
+ //否则不在这个服务器上，发送到mysql或者redis
|
|
|
+ }
|
|
|
+ } catch (Exception e) {
|
|
|
+ e.printStackTrace();
|
|
|
+ }
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+ /**
|
|
|
+ * @param session
|
|
|
+ * @param error
|
|
|
+ */
|
|
|
+ @OnError
|
|
|
+ public void onError(Session session, Throwable error) {
|
|
|
+ log.error("用户错误:" + this.userId + ",原因:" + error.getMessage());
|
|
|
+ error.printStackTrace();
|
|
|
+ }
|
|
|
+
|
|
|
+ /**
|
|
|
+ * 实现服务器主动推送
|
|
|
+ */
|
|
|
+ public void sendMessage(String message) throws IOException {
|
|
|
+ this.session.getBasicRemote().sendText(message);
|
|
|
+ }
|
|
|
+
|
|
|
+
|
|
|
+ /**
|
|
|
+ * 发送自定义消息
|
|
|
+ */
|
|
|
+ public static void sendInfo(String message, @PathParam("userId") String userId) throws IOException {
|
|
|
+ log.info("发送消息到:" + userId + "，报文:" + message);
|
|
|
+ if (StringUtils.isNotBlank(userId) && webSocketMap.containsKey(userId)) {
|
|
|
+ webSocketMap.get(userId).sendMessage(message);
|
|
|
+ } else {
|
|
|
+ log.error("用户" + userId + ",不在线！");
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+ public void wxappPhotoReady(){
|
|
|
+ log.info("wxappPhotoReady is come in");
|
|
|
+ }
|
|
|
+
|
|
|
+ public static synchronized int getOnlineCount() {
|
|
|
+ return onlineCount;
|
|
|
+ }
|
|
|
+
|
|
|
+ public static synchronized void addOnlineCount() {
|
|
|
+ WebSocketServer.onlineCount++;
|
|
|
+ }
|
|
|
+
|
|
|
+ public static synchronized void subOnlineCount() {
|
|
|
+ WebSocketServer.onlineCount--;
|
|
|
+ }
|
|
|
+
|
|
|
+// @Override
|
|
|
+// public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
|
|
|
+// try {
|
|
|
+// long threadId = Thread.currentThread().getId();
|
|
|
+// String threadName = Thread.currentThread().getName();
|
|
|
+// for (MessageExt messageExt : msgs) {
|
|
|
+// log.info(":{}-:{} websocketConsumer 重试次数:{}", threadId, threadName, messageExt.getReconsumeTimes());
|
|
|
+// MqDto mqDto = JacksonUtil.readJson(new String(messageExt.getBody(), Constants.CHARSET), MqDto.class);
|
|
|
+// log.info(":{}-:{} websocketConsumer 接收到的消息:{}", threadId, threadName, JacksonUtil.parseJson(mqDto));
|
|
|
+// log.info(":{}-:{} websocketConsumer mqDto sequence:{},tag:{}", threadId, threadName, mqDto.getSequence(), mqDto.getTag());
|
|
|
+// Map map = mqDto.getProperties();
|
|
|
+// String body = JacksonUtil.parseJson(mqDto.getBody());
|
|
|
+// log.info("map:{},body:{}", JacksonUtil.parseJson(map), body);
|
|
|
+// String model = String.valueOf(map.get("model"));
|
|
|
+// MessageModel messageModel = MessageModel.valueOf(model);
|
|
|
+// if (messageModel.ordinal() == MessageModel.CLUSTERING.ordinal()) {
|
|
|
+// webSocketMap.get(map.get("toUserId")).sendMessage(body);
|
|
|
+// } else {
|
|
|
+// webSocketMap.forEach((k, v) -> {
|
|
|
+// try {
|
|
|
+// v.sendMessage(body);
|
|
|
+// } catch (IOException e) {
|
|
|
+// e.printStackTrace();
|
|
|
+// }
|
|
|
+// });
|
|
|
+// }
|
|
|
+// }
|
|
|
+// } catch (Exception e) {
|
|
|
+// e.printStackTrace();
|
|
|
+// return ConsumeConcurrentlyStatus.RECONSUME_LATER;//重试
|
|
|
+// }
|
|
|
+// return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;//成功
|
|
|
+// }
|
|
|
+//
|
|
|
+// @Service
|
|
|
+// @RocketMQMessageListener(consumerGroup = "websocketConsumerImGroup", topic = "websocketImTopic", selectorType = SelectorType.TAG, selectorExpression = "*")
|
|
|
+// public class sessionConsumerWeb 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.setMessageModel(MessageModel.BROADCASTING);
|
|
|
+// defaultMQPushConsumer.registerMessageListener(WebSocketServer.this::consumeMessage);
|
|
|
+// }
|
|
|
+// }
|
|
|
+}
|
|
|
+
|
wangliang