backend-service/themis-backend/src/main/java/com/qmth/themis/backend/websocket/WebSocketServer.java

//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);
////        }
////    }
//}
//