java - 如何使用 WebSocket 发送定期消息？

Question

我在Tomcat上使用WebSocket(实际实现是Tyrus，JSR 356的引用实现)。当我必须处理客户端消息并响应它们时，它工作得很好。但是，我想为我的几个客户端控件实现推送 解决方案。实际上我需要两种解决方案:

• 以特定间隔推送数据，
• 在出现系统消息时推出系统消息。

对于第一个，我认为ScheduledExecutorService可以是一个解决方案，我已经有一个或多或少的工作示例，但我在清理方面遇到了问题。对于第二个，我想我需要有一个线程，它会触发 WebSocket 端点中的一个方法，但我也不知道如何干净利落地做到这一点。所谓干净，我的意思是只有在连接到我的端点的 session 时我才希望运行线程。

总结一下我的问题:您将如何正确地使用 Java EE WebSocket API 实现推送消息解决方案？

ps.: 我更喜欢“纯”解决方案，但 Spring 也不受欢迎。

---

当前代码框架

这是我当前第一个问题的解决方案:

@ServerEndpoint(...)
public class MyEndPoint {
    // own class, abstracting away session handling
    private static SessionHandler sessionHandler = new SessionHandler();
    private static ScheduledExecutorService timer =
            Executors.newSingleThreadScheduledExecutor();
    private static boolean timerStarted = false;

    @OnOpen
    public void onOpen(Session session, EndpointConfig config) {
        sessionHandler.addSession(session);
        if (!timerStarted) {
            timer.scheduleAtFixedRate(new Runnable() {
                @Override
                public void run() {
                    sessionHandler.sendToAllSession("foo");
                }
            }, 0, 3, TimeUnit.SECONDS);
            timerStarted = true;
        }
    }

    @OnClose
    public void onClose(Session session) {
        sessionHandler.removeSession(session);
        if (0 == sessionHandler.countSessions()) {
            // TODO: cleanup thread properly
            timer.shutdown();
            try {
                while (!timer.awaitTermination(10, TimeUnit.SECONDS));
            } catch (InterruptedException e) {
                log.debug("Timer terminated.");
            }
            timerStarted = false;
        }
    }
}

这或多或少有效，但在重新加载几页后，它因 RejectedExecutionException 而死，我不太确定如何处理这种情况。

Answer 1

不幸的是你不能在 shutdown() 之后使用任何 ExecutorService;

所以在 OnClose() 方法之后，下一个 OnOpen() 方法将会崩溃。

只是一些演示代码:

public class TestThread {

    public static void main(String[] args) {

        final ScheduledExecutorService timer = Executors.newSingleThreadScheduledExecutor();
        boolean timerStarted = false;
        //OnOpen - 1;  - OK
        if (!timerStarted) {
            timer.scheduleAtFixedRate(new Runnable() {
                @Override
                public void run() {
                    System.out.println("foo");
                }
            }, 0, 3, TimeUnit.SECONDS);
            timerStarted = true;
        }

        //OnOpen - 2;  - OK
        if (!timerStarted) {
            timer.scheduleAtFixedRate(new Runnable() {
                @Override
                public void run() {
                    System.out.println("foo");
                }
            }, 0, 3, TimeUnit.SECONDS);
            timerStarted = true;
        }

        //OnClose - 1  - OK
        timer.shutdown();
        timerStarted = false;

        //OnOpen - 2;  - NOT OK, because after stop you can't use timer,  RejectedExecutionException will thrown
        if (!timerStarted) {
            // will crash at this invocke
            timer.scheduleAtFixedRate(new Runnable() {
                @Override
                public void run() {
                    System.out.println("foo");
                }
            }, 0, 3, TimeUnit.SECONDS);
            timerStarted = true;
        }
    }
}

您可以尝试将您的类也用作网络监听器 http://docs.oracle.com/javaee/7/api/javax/servlet/annotation/WebListener.html 并在服务器启动和销毁时执行的方法中创建计时器

@WebListener
@ServerEndpoint(...)  
public class MyEndPoint implements ServletContextListener{

    final ScheduledExecutorService timer = Executors.newSingleThreadScheduledExecutor();

    @Override
    public void contextInitialized(ServletContextEvent servletContextEvent) {
        timer.scheduleWithFixedDelay(...)
    }

    @Override
    public void contextDestroyed(ServletContextEvent servletContextEvent) {
        timer.shutdown();
    }

    ...
}