1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51

<script src="https://code.jquery.com/jquery-3.6.4.min.js"></script>
<script src="https://cdn.jsdelivr.net/sockjs/1.1.4/sockjs.min.js"></script>
<script src="https://cdn.bootcdn.net/ajax/libs/stomp.js/2.3.3/stomp.min.js"></script>
 
<script>
    var stompClient = null;
 
    function connect() {
	var content = $("#message").val();
        var sender = 'tbb';
        var socket = new SockJS('http://127.0.0.1:8080/chat');
        stompClient = Stomp.over(socket);
        stompClient.connect({}, function (frame) {
            console.log('Connected: ' + frame);
            stompClient.subscribe('/user/tbb/message', function (message) {
                showMessage(JSON.parse(message.body));
            });
        });
    }
 
    function disconnect() {
        if (stompClient !== null) {
            stompClient.disconnect();
        }
        console.log("Disconnected");
    }
 
    function sendMessage() {
        var content = $("#message").val();
        var sender = $("#sender").val();
        stompClient.send("/app/chat/send/to/user", {}, JSON.stringify({
            content: content,
            sender: sender,
	    unionId: sender,
        }));
    }
 
    function showMessage(message) {
        $("#chat").append("<div>" + message.sender + ": " + message.content + "</div>");
    }
</script>
 
<div>
    <input type="text" id="sender" placeholder="Enter your name">
    <input type="text" id="message" placeholder="Type a message...">
    <button onclick="sendMessage()">Send</button>
    <button onclick="connect()">Connect</button>
    <button onclick="disconnect()">Disconnect</button>
</div>
 
<div id="chat">123</div>

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24

package com.example.mengchuangyuan.common.chat.config;
 
import org.springframework.context.annotation.Configuration;
import org.springframework.messaging.simp.config.MessageBrokerRegistry;
import org.springframework.web.socket.config.annotation.EnableWebSocketMessageBroker;
import org.springframework.web.socket.config.annotation.StompEndpointRegistry;
import org.springframework.web.socket.config.annotation.WebSocketMessageBrokerConfigurer;
 
@Configuration
@EnableWebSocketMessageBroker
public class WebSocketConfig implements WebSocketMessageBrokerConfigurer {
 
    @Override
    public void configureMessageBroker(MessageBrokerRegistry config) {
        config.enableSimpleBroker("/user","/agent","/topic"); // 定义消息代理，客户端订阅的地址前缀
        config.setApplicationDestinationPrefixes("/app"); // 定义客户端发送消息的地址前缀
        config.setUserDestinationPrefix("/user");
    }
 
    @Override
    public void registerStompEndpoints(StompEndpointRegistry registry) {
        registry.addEndpoint("/chat").setAllowedOriginPatterns("*").withSockJS(); // 定义WebSocket端点，客户端连接的地址
    }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74

package com.example.mengchuangyuan.common.chat.controller;
 
import com.example.mengchuangyuan.common.chat.entry.ChatMessage;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.handler.annotation.MessageMapping;
import org.springframework.messaging.handler.annotation.SendTo;
import org.springframework.messaging.simp.SimpMessagingTemplate;
import org.springframework.messaging.simp.annotation.SendToUser;
import org.springframework.stereotype.Controller;
 
@Slf4j
@Controller
public class ChatController {
    @Autowired
    private SimpMessagingTemplate simpMessagingTemplate;
 
    /**
     * 发消息（群发）
     * @param chatMessage 获取用户消息
     * @return 返回要回复的消息
     */
    //第一种方式
    @MessageMapping("/chat/send") // 定义消息映射路径
    @SendTo("/agent/public") // 发送消息到指定的代理路径
    public ChatMessage sendMessage(ChatMessage chatMessage) {
        log.info(chatMessage.toString());
        return chatMessage;
    }
 
    //方法二
//    @MessageMapping("/agent/send")
//    public void getAgentInfo (@Payload ChatMessage chatMessage) {
//        System.out.println("发送群发消息");
//        // 使用api进行推送
//        simpMessagingTemplate.convertAndSend("/agent/public2", chatMessage);
//    }
 
    /**
     * 发送给自己？
     * @param chatMessage
     * @return
     */
    @MessageMapping("/agent/send/user")
// 这里的路径必须还是以广播的前缀为前缀，否则无法接收
    @SendToUser("/agent/info")
    public ChatMessage sendUserMessage(ChatMessage chatMessage) {
        log.info(chatMessage.toString());
        return chatMessage;
    }
 
    /**
     * 发送给指定用户
     * @param chatMessage
     * @return
     */
    @MessageMapping("/chat/send/to/user")
// 这里的路径必须还是以广播的前缀为前缀，否则无法接收
    public void sendToUserMessage(ChatMessage chatMessage) {
        log.info("发送给指定用户:"+chatMessage.toString());
        simpMessagingTemplate.convertAndSendToUser(chatMessage.getUnionId(),"/message",chatMessage);
    }
}

package com.example.mengchuangyuan.common.chat.entry;
 
import lombok.Data;
 
@Data
public class ChatMessage {
    private String unionId;
    private String content;
    private String sender; 
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13

function connect() {
	var content = $("#message").val();//发送的消息内容
    var sender = 'tbb';//接收的人
    var socket = new SockJS('http://127.0.0.1:8080/chat');//连接后端socket的地址
    stompClient = Stomp.over(socket);
    stompClient.connect({}, function (frame) {
        console.log('Connected: ' + frame);
        //这里我写死了，接收来自tbb这个人的消息，'/user/tbb/message'可以改成'/user/'+sender+'/message'
        stompClient.subscribe('/user/tbb/message', function (message) { 
            showMessage(JSON.parse(message.body));
        });
    });
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24

package com.example.mengchuangyuan.common.chat.config;
 
import org.springframework.context.annotation.Configuration;
import org.springframework.messaging.simp.config.MessageBrokerRegistry;
import org.springframework.web.socket.config.annotation.EnableWebSocketMessageBroker;
import org.springframework.web.socket.config.annotation.StompEndpointRegistry;
import org.springframework.web.socket.config.annotation.WebSocketMessageBrokerConfigurer;
 
@Configuration
@EnableWebSocketMessageBroker
public class WebSocketConfig implements WebSocketMessageBrokerConfigurer {
 
    @Override
    public void configureMessageBroker(MessageBrokerRegistry config) {
        config.enableSimpleBroker("/user","/agent","/topic"); // 定义消息代理，客户端订阅的地址前缀
        config.setApplicationDestinationPrefixes("/app"); // 定义客户端发送消息的地址前缀
        config.setUserDestinationPrefix("/user");
    }
 
    @Override
    public void registerStompEndpoints(StompEndpointRegistry registry) {
        registry.addEndpoint("/chat").setAllowedOriginPatterns("*").withSockJS(); // 定义WebSocket端点，客户端连接的地址
    }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21

function sendMessage() { //发消息给对方
    var content = $("#message").val();
    var sender = $("#sender").val();
    stompClient.send("/app/chat/send/to/user", {}, JSON.stringify({ //发消息的后端的路由
        content: content,//消息内容
        sender: sender,//消息姓名
	    unionId: sender,//消息id，这里我把消息姓名也作为id
    }));
}
/**
    * 发送给指定用户
    * @param chatMessage
    * @return
*/
@MessageMapping("/chat/send/to/user")
// 这里的路径必须还是以广播的前缀为前缀，否则无法接收
public void sendToUserMessage(ChatMessage chatMessage) {
    log.info("发送给指定用户:"+chatMessage.toString());
    //chatMessage.getUnionId()是发送给某人的id，/message对应四段中第一段的/user/tbb/message中的message
    simpMessagingTemplate.convertAndSendToUser(chatMessage.getUnionId(),"/message",chatMessage);
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13

package com.example.mengchuangyuan.common.chat.mini.config;
 
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.web.socket.server.standard.ServerEndpointExporter;
 
@Configuration
public class MiniWebSocketConfig {
    @Bean
    public ServerEndpointExporter serverEndpointExporter() {
        return new ServerEndpointExporter();
    }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75

package com.example.mengchuangyuan.common.chat.mini.controller;
 
import com.example.mengchuangyuan.common.redis.tool.SufferVariable;
import com.example.mengchuangyuan.common.tool.Result;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.web.bind.annotation.*;
import org.springframework.web.multipart.MultipartFile;
 
import java.io.File;
import java.io.IOException;
import java.util.*;
 
@Slf4j
@RestController
@RequestMapping("/mini/socket")
public class MiniWebSocketController {
    @Value("${upload.messageChatUrl}")
    private String messageChatUrl;
 
    @Value("${upload.messageChatPath}")
    private String messageChatPath;
 
    private final List<Map<String,Object>> mapList = new ArrayList<>();
 
    @GetMapping("/get/openid/history")
    private Result getOpenidHistory(@RequestParam("openid") String openid,@RequestParam("linkType") String linkType){
//        log.info(openid);
        List<String> openids = (List<String>) SufferVariable.openidKeys.get(linkType);
//        if (keyMap==null){
//            return Result.error();
//        }
//        List<String> openids = (List<String>) keyMap.get(openid);
//        log.info("所以openid"+String.valueOf(openids));
        if (openids==null){
            return Result.error();
        }
        mapList.clear();
        for (String openid1: openids) {
            if (openid1.contains(openid)) {
//            if (SufferVariable.messageMap.get(openid1)!=null){
                mapList.add((Map<String, Object>) SufferVariable.messageMap.get(openid1));
//                log.info("聊天记录缓存"+openid);
//                log.info(String.valueOf(SufferVariable.messageMap.get(openid1)));
 
//            }
            }
        }
        return Result.success(mapList);
    }
 
 
//上传聊天图片
    @PostMapping("/img/upload")
    public Result imgUpDown(@RequestParam("file") MultipartFile file,
                            @RequestParam("filename") String filename) throws IOException {
        System.out.println(filename);
        File file1 = new File(messageChatPath,filename);
        if(!file1.exists()) {
            if(!file1.mkdirs()){ //创建目录
                return Result.error();
            }
        }
        //获取文件名
        String fileName = file.getOriginalFilename();
        //获取文件后缀名。也可以在这里添加判断语句，规定特定格式的图片才能上传，否则拒绝保存。
        String suffixName = fileName.substring(fileName.lastIndexOf("."));
        //为了避免发生图片替换，这里使用了文件名重新生成
        fileName = UUID.randomUUID()+suffixName;
 
        file.transferTo(new File(file1,fileName));
 
        return Result.success(messageChatUrl+filename+"/"+fileName);
    }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83

package com.example.mengchuangyuan.common.chat.mini.mapper;
 
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.example.mengchuangyuan.common.redis.mapper.RedisUtils;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
 
import javax.annotation.PostConstruct;
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.io.IOException;
import java.util.HashMap;
import java.util.Map;
 
//对外公布的一个后端站点
//ws://localhost:8080/websocket/用户id
@ServerEndpoint(value = "/websocket/{userId}")
@Component
@Slf4j
public class WebSocketEndPoint {
    //与某个客户端的连接会话，需要他来给客户端发送数据
    private Session session;
 
    @Autowired
    private SessionPool sessionPool;
//    @Autowired
//    private RedisUtils redisUtils;
    private static WebSocketEndPoint webSocketEndPoint;
    //初始化 ②
    @PostConstruct
    public void init() {
        webSocketEndPoint = this;
        webSocketEndPoint.sessionPool = this.sessionPool;
    }
 
    //连接建立成功调用的方法
    @OnOpen
    public void onOpen(Session session, @PathParam("userId") String userId) {
        //把会话加入连接池中
        //userId通过用户传入，session是系统自动产生
        SessionPool.sessions.put(userId, session);
 
        //TODO 可以添加日志操作
    }
 
    //关闭会话的时候
    @OnClose
    public void onClose(Session session) throws IOException {
        webSocketEndPoint.sessionPool.close(session.getId());
        session.close();
    }
 
    //接收客户端的消息后调用的方法，在这里可以进行各种业务逻辑的操作
    @OnMessage
    public void onMessage(String message, Session session) {
        System.out.println(message);
//        log.info("redisUtils:"+redisUtils);
        //心跳检测
        if (message.equalsIgnoreCase("ping")) {
            try {
                Map<String, Object> params = new HashMap<>();
                params.put("type", "pong");
                session.getBasicRemote().sendText(JSON.toJSONString(params));
            } catch (Exception e) {
                e.printStackTrace();
            }
            return;
        }
        //将Json字符串转为键值对
//        HashMap params = JSON.parseObject(message, HashMap.class);
        JSONObject params =  JSON.parseObject(message);
        webSocketEndPoint.sessionPool.sendMessage(params);
 
        //这里的业务逻辑仅仅是把收到的消息返回给前端
//        SessionPool.sendMessage(message);
    }
}

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150

package com.example.mengchuangyuan.common.chat.mini.mapper;
 
import com.alibaba.fastjson.JSON;
import com.example.mengchuangyuan.common.redis.mapper.RedisUtils;
import com.example.mengchuangyuan.common.redis.tool.SufferVariable;
import com.example.mengchuangyuan.common.tool.DateYMDms;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import org.springframework.stereotype.Controller;
 
import javax.websocket.Session;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
 
@Component
@Slf4j
public class SessionPool {
    @Autowired
    private RedisUtils redisUtils;
 
    //key-value : userId - 会话（系统创建）
    public static Map<String, Session> sessions = new ConcurrentHashMap<>();//避免多线程问题
 
    public void close(String sessionId) {
        //sessionId是在session中添加了一个标识，准确定位某条session
        for (String userId : SessionPool.sessions.keySet()) {
            Session session = SessionPool.sessions.get(userId);
            if (session.getId().equals(sessionId)) {
                sessions.remove(userId);
                break;
            }
        }
    }
 
    public void sendMessage(String userId, String message) {
        sessions.get(userId).getAsyncRemote().sendText(message);
    }
 
 
    //消息的群发，业务逻辑的群发
    public void sendMessage(String message) {
//        redisUtils.cacheValue("chatMessage","String.valueOf(SufferVariable.messageMap)");
        for (String sessionId : SessionPool.sessions.keySet()) {
            SessionPool.sessions.get(sessionId).getAsyncRemote().sendText(message);
        }
    }
 
    //点对点的消息推送
    public void sendMessage(Map<String, Object> params) {
        log.info("消息内容："+String.valueOf(params));
        long mid = System.currentTimeMillis();
        Map<String,Object> formUser = (Map<String, Object>) params.get("formUser");
        Map<String,Object> toUser = (Map<String, Object>) params.get("toUser");
 
        String userId = formUser.get("openid").toString();
        String toUserId = toUser.get("openid").toString();
//        String msg = params.get("message").toString();
        String type  = params.get("type").toString();
        String linkType = params.get("linkType").toString();
        //获取用户session
        Session session = sessions.get(toUserId);
 
        Map<String,Object> keyMap;
        List<String> setOpenid;
        Map<String,Object> map;
        List<Object> list;
        String uid = userId+toUserId;
        params.put("mid",mid);
 
        if (SufferVariable.messageMap.get(userId+toUserId)==null&&SufferVariable.messageMap.get(toUserId+userId)==null){
//            messageMap.put(userId+toUserId,userId+toUserId);
            map = new HashMap<>();
            list = new ArrayList<>();
 
            map.put(linkType,list);
            SufferVariable.messageMap.put(uid,map);
 
//            list.add(params);
 
        }else {
            if (SufferVariable.messageMap.get(userId+toUserId)!=null){
                uid = userId+toUserId;
 
            }else {
               uid = toUserId+userId;
            }
            map = (Map<String, Object>) SufferVariable.messageMap.get(uid);
            list = (List<Object>) map.get(linkType);
 
        }
 
        //获取历史记录
        if (type.equalsIgnoreCase("history")){
            if (sessions.get(userId) != null) {
                Map<String, Object> map2 = new HashMap<>();
//                map = (Map<String, Object>) SufferVariable.messageMap.get(uid);
                map2.put("type", "isHistory");
                map2.put("message",list);
                sessions.get(userId).getAsyncRemote().sendText(JSON.toJSONString(map2));
            }
            return;
        }
        String nowDate = DateYMDms.getUtilDate();
        if (list.size()!=0) {
            Map<String, Object> lastMap = (Map<String, Object>) list.get(list.size() - 1);
            String date = DateYMDms.getYMDms(5, lastMap.get("nowDate").toString());
            params.put("date", date);
        }else {
            params.put("date", nowDate);
        }
//        if (SufferVariable.openidKeys.get(linkType)==null){
//            keyMap = new HashMap<>();
//        }else {
//            keyMap = (Map<String, Object>) SufferVariable.openidKeys.get(linkType);
//        }
//        if (keyMap.get(userId)==null){
//            setOpenid = new ArrayList<>();
//        }else {
//            log.info(String.valueOf(SufferVariable.openidKeys));
//            setOpenid = (List<String>) keyMap.get(userId);
//        }
        if (SufferVariable.openidKeys.get(linkType) == null){
            setOpenid = new ArrayList<>();
        }else {
            setOpenid = (List<String>) SufferVariable.openidKeys.get(linkType);
        }
        setOpenid.add(uid);
        Set<String> set = new LinkedHashSet<>(setOpenid);
        setOpenid.clear();
        setOpenid.addAll(set);
//        keyMap.put(userId,setOpenid);
        SufferVariable.openidKeys.put(linkType,setOpenid);
 
        params.put("nowDate",nowDate);
        list.add(params);
        map.put(linkType,list);
        SufferVariable.messageMap.put(uid,map);
        redisUtils.cacheValue("chatMessage",SufferVariable.messageMap);
        redisUtils.cacheValue("chatOpenidKeys",SufferVariable.openidKeys);
 
        if (session != null) {
            log.info(uid+"："+String.valueOf(SufferVariable.messageMap.get(uid)));
            session.getAsyncRemote().sendText(JSON.toJSONString(params));
        }
//        log.info(String.valueOf(messageMap));
//        params.remove("formUserId");
        //session不为空的情况下进行点对点推送
    }
}