Java WebSocket服务器的ConcurrentModificationException

Question

我正在构建一个处理对象绘制的WebSocket服务器。下面是服务器的类的样子：

import fmi.whiteboard.models.paths.*;
import jakarta.websocket.*;
import jakarta.websocket.server.ServerEndpoint;

import java.io.IOException;
import java.util.Collections;
import java.util.HashSet;
import java.util.Set;

@ServerEndpoint(value = "/whiteboard",
        encoders = {DrawingEncoder.class, ShapeEncoder.class, PathEncoder.class},
        decoders = {DrawingDecoder.class, ShapeDecoder.class, PathDecoder.class})
public class WhiteboardServer {

    private static Set<Session> peers = Collections.synchronizedSet(new HashSet<Session>());

    @OnOpen
    public void onOpen(Session peer) {
        peers.add(peer);
    }

    @OnClose
    public void onClose(Session peer) {
        peers.remove(peer);
    }

    @OnMessage
    public void broadcastShape(Drawing drawing, Session session) throws IOException, EncodeException {
        for (Session peer : peers) {
            if (!peer.equals(session)) {
                peer.getAsyncRemote().sendObject(drawing);
            }
        }
    }

}

下面是处理绘图的ReactJS组件的外观：

export default function Whiteboard() {
  const canvas = useRef(null);

  const ws = useMemo(() => {
    const socket = new WebSocket("ws://localhost:8080/backend_war/whiteboard");
    socket.binaryType = "arraybuffer";
    return socket;
  }, []);

  useEffect(() => {
    ws.onmessage = (evt: MessageEvent) => {
      const data: SocketResponse = JSON.parse(evt.data);
      if (data && data.shapes) {
        canvas?.current?.loadPaths(data.shapes);
      }
    };

    ws.onerror= (err) => console.log(err);
    
  }, [ws]);

  const onDraw = (message: CanvasPath[]) => {
    ws.send(JSON.stringify({ shapes: message }));
  };

  return (
    <>
      <ReactSketchCanvas
        ref={canvas}
        style={styles}
        onUpdate={(paths) => setPaths(paths)}
        strokeWidth={4}
        strokeColor="red"
      />
    </>
  );
}

这个应用程序在2-3秒内运行良好，但当我开始绘制更多的形状时，socket服务器在与for (Session peer : peers)的线路上使用ConcurrentModificationException崩溃。

如果有帮助的话，我使用Java11和Tomcat10作为服务器。

Answer 1

Collections.synchronizedSet创建了一个同步单个项目访问的集合。这涉及到get、add、remove等方法。然而，迭代器并不同步。在迭代过程中，当另一个线程修改集合时，您将获得ConcurrentModificationException。

您有两个解决方案。或者像这样使用同步块来保护外观：

synchronized(peers) {
  for (Session peer: peers) {
    ...
  }
}

或者，更好的方法是切换到ConcurrentHashSet，它可以保证在不抛出ConcurrentModificationException的情况下实现多线程的优化访问。

Answer 2

当您使用Collections.synchronizedSet同步和现有集合时，它只同步访问数据的方法，如get()、set()...而不是迭代器。因此，您还需要像这样同步迭代器，例如

synchronized(peers){
        for (Session peer : peers) {
            if (!peer.equals(session)) {
                peer.getAsyncRemote().sendObject(drawing);
            }
        }   
    }

这是documentation