java - BlockingQueue 丢失其引用并在Message Jetty WebSocket 上抛出 NullPointerException

Question

不得不问这个问题，因为这是试图解决问题的一天，但不能。
我正在使用 Netbeans 8.2 和 java 8。
拓扑 :

浏览器上的 WebSocket 客户端

Jetty WebSocket 服务器(带有 Swing GUI 的 Java 应用程序)

目标 : 将数据从客户端发送到服务器并在 JTextArea (GUI) 上显示数据
Main.java (GUI)

public class Main extends javax.swing.JPanel {

private WebSocketSwing websocketserver;
private BlockingQueue<String> stack = new ArrayBlockingQueue<String>(3);

public Main() {
    initComponents();
    // WebSocketServer
    websocketserver = new WebSocketSwing(stack);
    websocketserver.start();
    
    consumer.start();
}

Thread consumer = new Thread(new Runnable() {
    @Override
    public void run() {
        try{
            String msg;
            //consuming messages until exit message is received
            while((msg = stack.take()) !="exit"){
                Thread.sleep(10);
                System.out.println("Consumed: " + msg);
            }
        }catch(InterruptedException e) {
            e.printStackTrace();
        }
    }
});

private void initComponents() {
    //GUI code goes here
}

    public static void main(String[] args) {
    JFrame frame = new JFrame("Main GUI");
            
    java.awt.EventQueue.invokeLater(new Runnable() {
        public void run() {
            frame.setDefaultCloseOperation(javax.swing.WindowConstants.EXIT_ON_CLOSE);
            frame.getContentPane().add(new SDG());
            frame.pack();
            frame.setVisible(true);        
        }
    });   
}

private javax.swing.JLabel jLabel1;
private javax.swing.JScrollPane jScrollPane1;
private javax.swing.JTextArea txt_area;

}
WebSocketSwing 类

public class WebSocketSwing extends Thread {

/**
 * @param args the command line arguments
 */
private BlockingQueue<String> stack;

public WebSocketSwing(BlockingQueue<String> queue){
    this.stack = queue;
}
@Override
public void run(){
    super.run();
    
    try {
        Server server = new Server(2014);
        WSHandler mHandler = new WSHandler();
        mHandler.SetStack(stack);
        server.setHandler(mHandler);
        server.setStopTimeout(0);
        server.start();
        //
        server.join();
        } catch (Exception e) {
            e.printStackTrace();
        }
}
}

WSHandler 类

@WebSocket
public class WSHandler extends WebSocketHandler {
    private Session session;
    public BlockingQueue<String> stack = new ArrayBlockingQueue<String>(3); // **THIS instantiation should not be needed...** 
    private static ArrayList<WSHandler> sessions = new ArrayList<WSHandler>();
    
    public static ArrayList<WSHandler> getAllSessions() {
        return sessions;
    }

    /*WSHandler(BlockingQueue<String> stack) { // **I tried to send/assign the queue from the constructor but the method is not overridable**
        SetStack(stack);  // or this.stack = stack;
    }*/
    
    
    public void SetStack(BlockingQueue<String> queue){
        this.stack = queue;
        //Testing operations to see the reference to the queue was successfully passed
        System.out.println(stack.remainingCapacity());
        stack.offer("Something");  //**consumes just fine in the other Thread...**
        
    }
    
    @OnWebSocketClose
    public void onClose(int StatusCode, String reason){
        sessions.remove(this);
        System.out.println("Close: Status Code: " + StatusCode + ", reason: " + reason + ", sessions = " + sessions.size());
    }
    
    @OnWebSocketError
    public void onError(Throwable t) {
        System.out.println("Error: " + t.getMessage());
    }

    @OnWebSocketConnect
    public void onConnect(Session localSession) {
        session = localSession;
        sessions.add(this);
        
        System.out.println("Connect: " + session.getRemoteAddress().getAddress());
    }
    
    @OnWebSocketMessage
    public void onMessage(String message) {
        try {
            System.out.println("Message: " + message);
            session.getRemote().sendString("ACK");
            
            SetData(message);
            if(message.equals("exit")){
                System.out.println("Message: Bye!...");
                System.exit(0);
            }
        } catch (IOException ex) {
            Logger.getLogger(WSHandler.class.getName()).log(Level.SEVERE, null, ex);
        } 
    }
    
    private void SetData(String message){
        try{
        if (stack.offer(message)){
            System.out.print("Inserted");
        } else {
            System.out.print("NOT Inserted");
        }
        } catch(NullPointerException e){
            e.printStackTrace();
        }
    }

    @Override
    public void configure(WebSocketServletFactory factory) {
        factory.register(WSHandler.class);
    }
    
}

结果 > 好像 this.stack 丢失了对队列的引用...
好像我没有在类中初始化 BlockingQueue 在 SetStack 方法之外抛出 NPE ......
跟踪(当我没有在 WSHandler 类上初始化 BlockingQueue 时)
如果我的理解是正确的，如果来自 Main 类的引用已正确传递，我应该不需要在 Handler 中初始化 BlockingQueue ......然后我认为这是要解决的问题......
空指针异常 被抛出是因为对象丢失了它的引用(它在 SetStack 方法中的引用......)......这个原因是我无法找到的......

2021-06-26 15:35:41.990:INFO::Thread-2: Logging initialized @470ms to org.eclipse.jetty.util.log.StdErrLog
3
2021-06-26 15:35:42.077:INFO:oejs.Server:Thread-2: jetty-9.4.42.v20210604; built: 2021-06-04T17:33:38.939Z; git: 5cd5e6d2375eeab146813b0de9f19eda6ab6e6cb; jvm 1.8.0_111-b14
Consumed: Something
2021-06-26 15:35:42.827:INFO:oejs.AbstractConnector:Thread-2: Started ServerConnector@98f7e6f{HTTP/1.1, (http/1.1)}{0.0.0.0:2014}
2021-06-26 15:35:42.830:INFO:oejs.Server:Thread-2: Started @1314ms
Connect: /127.0.0.1
Message: sample message
Happened
java.lang.NullPointerException
    at websocketswing.WSHandler.SetData(WSHandler.java:100)
    at websocketswing.WSHandler.onMessage(WSHandler.java:78)
    at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
    at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
    at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
    at java.lang.reflect.Method.invoke(Method.java:498)
    at org.eclipse.jetty.websocket.common.events.annotated.CallableMethod.call(CallableMethod.java:70)
    at org.eclipse.jetty.websocket.common.events.annotated.OptionalSessionCallableMethod.call(OptionalSessionCallableMethod.java:72)
    at org.eclipse.jetty.websocket.common.events.JettyAnnotatedEventDriver.onTextMessage(JettyAnnotatedEventDriver.java:301)
    at org.eclipse.jetty.websocket.common.message.SimpleTextMessage.messageComplete(SimpleTextMessage.java:69)
    at org.eclipse.jetty.websocket.common.events.AbstractEventDriver.appendMessage(AbstractEventDriver.java:67)
    at org.eclipse.jetty.websocket.common.events.JettyAnnotatedEventDriver.onTextFrame(JettyAnnotatedEventDriver.java:287)
    at org.eclipse.jetty.websocket.common.events.AbstractEventDriver.incomingFrame(AbstractEventDriver.java:152)
    at org.eclipse.jetty.websocket.common.WebSocketSession.incomingFrame(WebSocketSession.java:326)
    at org.eclipse.jetty.websocket.common.extensions.AbstractExtension.nextIncomingFrame(AbstractExtension.java:148)
    at org.eclipse.jetty.websocket.common.extensions.compress.PerMessageDeflateExtension.nextIncomingFrame(PerMessageDeflateExtension.java:111)
    at org.eclipse.jetty.websocket.common.extensions.compress.CompressExtension.forwardIncoming(CompressExtension.java:169)
    at org.eclipse.jetty.websocket.common.extensions.compress.PerMessageDeflateExtension.incomingFrame(PerMessageDeflateExtension.java:90)
    at org.eclipse.jetty.websocket.common.extensions.ExtensionStack.incomingFrame(ExtensionStack.java:202)
    at org.eclipse.jetty.websocket.common.Parser.notifyFrame(Parser.java:225)
    at org.eclipse.jetty.websocket.common.Parser.parseSingleFrame(Parser.java:259)
    at org.eclipse.jetty.websocket.common.io.AbstractWebSocketConnection.onFillable(AbstractWebSocketConnection.java:459)
    at org.eclipse.jetty.websocket.common.io.AbstractWebSocketConnection.onFillable(AbstractWebSocketConnection.java:440)
    at org.eclipse.jetty.io.AbstractConnection$ReadCallback.succeeded(AbstractConnection.java:311)
    at org.eclipse.jetty.io.FillInterest.fillable(FillInterest.java:105)
    at org.eclipse.jetty.io.ChannelEndPoint$1.run(ChannelEndPoint.java:104)
    at org.eclipse.jetty.util.thread.strategy.EatWhatYouKill.runTask(EatWhatYouKill.java:338)
    at org.eclipse.jetty.util.thread.strategy.EatWhatYouKill.doProduce(EatWhatYouKill.java:315)
    at org.eclipse.jetty.util.thread.strategy.EatWhatYouKill.tryProduce(EatWhatYouKill.java:173)
    at org.eclipse.jetty.util.thread.strategy.EatWhatYouKill.produce(EatWhatYouKill.java:137)
    at org.eclipse.jetty.util.thread.QueuedThreadPool.runJob(QueuedThreadPool.java:882)
    at org.eclipse.jetty.util.thread.QueuedThreadPool$Runner.run(QueuedThreadPool.java:1036)
    at java.lang.Thread.run(Thread.java:745)

跟踪(初始化队列时)

2021-06-26 15:39:36.821:INFO::Thread-2: Logging initialized @470ms to org.eclipse.jetty.util.log.StdErrLog
3
2021-06-26 15:39:36.889:INFO:oejs.Server:Thread-2: jetty-9.4.42.v20210604; built: 2021-06-04T17:33:38.939Z; git: 5cd5e6d2375eeab146813b0de9f19eda6ab6e6cb; jvm 1.8.0_111-b14
Consumed: Something
2021-06-26 15:39:37.961:INFO:oejs.AbstractConnector:Thread-2: Started ServerConnector@358d4f07{HTTP/1.1, (http/1.1)}{0.0.0.0:2014}
2021-06-26 15:39:37.964:INFO:oejs.Server:Thread-2: Started @1615ms
Connect: /127.0.0.1
Message: sample message
Happened
Inserted
Message: sample message
Happened
NOT Inserted
Message: sample message
Happened
NOT Inserted

因此，我假设队列丢失了它的引用，因为“这个队列”永远不会被消费者线程消耗(就像在第一次分配中那样)
希望有人能看到我没有看到的东西...
此致，

Answer 1

全新 WSHandler由于您在 configure() 中注册它的方式，每个新的(和接受/升级的)WebSocket 连接都会创建实例。方法 ...

@Override
public void configure(WebSocketServletFactory factory) {
    factory.register(WSHandler.class);
}

重构你的代码。
首先分离 WSHandler来自 WebSocket 端点。
使新 MyEndpoint为您的 queue 设置一个构造函数(或 setter)目的。

@WebSocket
public class MyEndpoint {
   private .... queue;

   public MyEndpoint(... queue) {
       this.queue = queue;
   }

   @OnWebSocketMessage
   public void onMessage(String str) {
       this.queue.offer(str);
   }
}

接下来，您要创建一个自定义 org.eclipse.jetty.websocket.servlet.WebSocketCreator您自己设计的 WebSocket 端点实例，填充它，然后将其交还给 Jetty 实现。

public static class MyWebSocketCreator implements WebSocketCreator {
    private ... masterQueue = new ...;

    @Override
    public Object createWebSocket(ServletUpgradeRequest req, ServletUpgradeResponse resp) {
        return new MyEndpoint(masterQueue);
    }
}

最后，您想制作您的 configure()方法使用这个新创建者。

@Override
public void configure(WebSocketServletFactory factory) {
    factory.setCreator(new MyWebSocketCreator());
}

这在我之前的答案选项 2 中涵盖，位于 How do I access instantiated WebSockets in Jetty 9?