如何用Java开发小程序实时聊天功能 Java WebSocket实现消息同步

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-websocket</artifactId>
</dependency>

import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.web.socket.server.standard.ServerEndpointExporter;

@Configuration
public class WebSocketConfig {

    /**
     * ServerEndpointExporter 负责扫描和注册所有带有 @ServerEndpoint 注解的 WebSocket 端点。
     * 如果使用独立的servlet容器，则无需提供此Bean。
     */
    @Bean
    public ServerEndpointExporter serverEndpointExporter() {
        return new ServerEndpointExporter();
    }
}

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;
import java.util.concurrent.CopyOnWriteArraySet;

@ServerEndpoint("/ws/chat/{userId}")
@Component // 确保Spring能够扫描到这个组件
public class ChatWebSocketEndpoint {

    // 存储所有在线的Session，线程安全
    private static CopyOnWriteArraySet<ChatWebSocketEndpoint> webSocketSet = new CopyOnWriteArraySet<>();
    // 存储用户ID和对应的WebSocket实例，方便点对点发送
    private static ConcurrentHashMap<String, Session> sessionPool = new ConcurrentHashMap<>();

    private Session session;
    private String userId; // 当前连接的用户ID

    /**
     * 连接建立成功调用的方法
     */
    @OnOpen
    public void onOpen(Session session, @PathParam("userId") String userId) {
        this.session = session;
        this.userId = userId;
        webSocketSet.add(this);
        sessionPool.put(userId, session);
        System.out.println("用户[" + userId + "]连接成功，当前在线人数为：" + webSocketSet.size());
        // 可以在这里广播上线消息
        sendMessageToAll("系统消息：用户[" + userId + "]上线了！");
    }

    /**
     * 收到客户端消息后调用的方法
     */
    @OnMessage
    public void onMessage(String message) {
        System.out.println("收到用户[" + userId + "]的消息：" + message);
        // 假设消息格式是 JSON，包含发送者、接收者和内容
        // 这里简化处理，直接广播
        sendMessageToAll("用户[" + userId + "]：" + message);
    }

    /**
     * 连接关闭调用的方法
     */
    @OnClose
    public void onClose() {
        webSocketSet.remove(this);
        sessionPool.remove(this.userId);
        System.out.println("用户[" + userId + "]断开连接，当前在线人数为：" + webSocketSet.size());
        // 可以在这里广播下线消息
        sendMessageToAll("系统消息：用户[" + userId + "]下线了！");
    }

    /**
     * 连接发生错误调用的方法
     */
    @OnError
    public void onError(Session session, Throwable error) {
        System.err.println("用户[" + this.userId + "]连接发生错误：" + error.getMessage());
        error.printStackTrace();
    }

    /**
     * 发送消息给指定用户
     */
    public void sendMessage(String userId, String message) {
        Session s = sessionPool.get(userId);
        if (s != null && s.isOpen()) {
            try {
                s.getBasicRemote().sendText(message);
            } catch (IOException e) {
                System.err.println("发送消息给[" + userId + "]失败：" + e.getMessage());
            }
        }
    }

    /**
     * 广播消息给所有在线用户
     */
    public void sendMessageToAll(String message) {
        for (ChatWebSocketEndpoint item : webSocketSet) {
            try {
                item.session.getBasicRemote().sendText(message);
            } catch (IOException e) {
                System.err.println("广播消息失败：" + e.getMessage());
            }
        }
    }
}

// app.js 或某个页面
App({
  globalData: {
    socketOpen: false,
    socketMsgQueue: [],
    // 假设用户ID从登录获取
    userId: 'user_' + Math.floor(Math.random() * 1000) 
  },

  onLaunch: function () {
    this.connectWebSocket();
  },

  connectWebSocket: function () {
    const self = this;
    const wsUrl = `ws://localhost:8080/ws/chat/${self.globalData.userId}`; // 替换为你的后端地址

    wx.connectSocket({
      url: wsUrl,
      success: res => {
        console.log('WebSocket连接成功', res);
      },
      fail: err => {
        console.error('WebSocket连接失败', err);
      }
    });

    wx.onSocketOpen(function (res) {
      self.globalData.socketOpen = true;
      console.log('WebSocket连接已打开！');
      // 连接打开后，发送队列中积压的消息
      while (self.globalData.socketMsgQueue.length > 0) {
        const msg = self.globalData.socketMsgQueue.shift();
        self.sendSocketMessage(msg);
      }
    });

    wx.onSocketMessage(function (res) {
      console.log('收到服务器消息：', res.data);
      // 在这里处理收到的消息，例如更新聊天界面
      // 注意：res.data 是字符串，如果后端发送的是JSON，需要JSON.parse(res.data)
      const message = res.data;
      // 假设有一个页面实例来更新UI
      const currentPage = getCurrentPages().pop(); // 获取当前页面实例
      if (currentPage && typeof currentPage.addChatMessage === 'function') {
        currentPage.addChatMessage(message);
      }
    });

    wx.onSocketError(function (res) {
      self.globalData.socketOpen = false;
      console.error('WebSocket连接发生错误！', res);
    });

    wx.onSocketClose(function (res) {
      self.globalData.socketOpen = false;
      console.log('WebSocket连接已关闭！', res);
      // 可以在这里尝试重连
      // setTimeout(() => self.connectWebSocket(), 3000); 
    });
  },

  sendSocketMessage: function (msg) {
    if (this.globalData.socketOpen) {
      wx.sendSocketMessage({
        data: msg,
        success: res => {
          console.log('消息发送成功', msg, res);
        },
        fail: err => {
          console.error('消息发送失败', msg, err);
        }
      });
    } else {
      // 如果连接未打开，将消息加入队列等待发送
      this.globalData.socketMsgQueue.push(msg);
      console.log('WebSocket未连接，消息已加入队列', msg);
    }
  },

  closeWebSocket: function () {
    if (this.globalData.socketOpen) {
      wx.closeSocket({
        success: res => {
          console.log('WebSocket连接关闭成功', res);
        },
        fail: err => {
          console.error('WebSocket连接关闭失败', err);
        }
      });
    }
  }
});

// pages/chat/chat.js
Page({
  data: {
    messages: [],
    inputContent: ''
  },

  onLoad: function () {
    // 确保WebSocket连接已建立
    if (!getApp().globalData.socketOpen) {
      getApp().connectWebSocket();
    }
  },

  onUnload: function() {
    // 页面卸载时可以考虑关闭WebSocket，或者根据业务需求保持连接
    // getApp().closeWebSocket(); 
  },

  // 接收到消息后更新UI
  addChatMessage: function(message) {
    this.setData({
      messages: [...this.data.messages, message]
    });
    // 滚动到最新消息
    wx.pageScrollTo({
      scrollTop: 99999,
      duration: 300
    });
  },

  // 监听输入框
  onInput: function(e) {
    this.setData({
      inputContent: e.detail.value
    });
  },

  // 发送消息
  sendMessage: function() {
    const content = this.data.inputContent.trim();
    if (content) {
      getApp().sendSocketMessage(content);
      this.setData({
        inputContent: '' // 清空输入框
      });
    }
  }
});