当关联对象“陈旧”时清理 Redisson 发布/订阅侦听器答案

【问题标题】：Clean up Redisson pub/sub listeners when assoicated object is 'stale'当关联对象“陈旧”时清理 Redisson 发布/订阅侦听器
【发布时间】：2021-03-08 03:32:02
【问题描述】：

我正在尝试使用 Redis 和 Redisson 库在 Java 中实现一个简单的 Websocket 应用程序，该应用程序能够水平扩展。

Websocket 服务器基本上跟踪连接的客户端，并将接收到的消息发送到 Rtopic - 这很好用。

为了消费，我有在注册客户端时添加监听器的代码：它通过以下方式将客户端对象与监听器相关联：

private static RedissonClient redisson = RedissonRedisServer.createRedisConnectionWithConfig();
public static final RTopic subcriberTopic = redisson.getTopic("clientsMapTopic");

public static boolean sendToPubSub(ConnectedClient q, String message) {
        boolean[] success = {true};
        MessageListener<Message> listener = new MessageListener<Message>() {
            @Override
            public void onMessage(CharSequence channel, Message message) {
                logger.debug("The message is : " + message.getMediaId());

                try {
                    logger.debug("ConnectedClient mediaid: " + q.getMediaid() + ",Message mediaid " + message.getMediaId());
                    if (q.getMediaid().equals(message.getMediaId())) {
                        // we need to verify if the message goes to the right receiver
                        logger.debug("MESSAGE from PUBSUB to (" + q.getId() + ") @ " + q.getSession().getId() + " " + message);
                        // this is the actual message to the websocket client
                        // this executes on the wrong connected client when the connection is closed and reopened
                        q.getSession().getBasicRemote().sendText(message.getMessage());
                    }
                } catch (Exception e) {
                    e.printStackTrace();
                    success[0] = false;
                }
            }
        };
        int listenerId = subcriberTopic.addListener(Message.class, listener);
}

我观察到的问题如下：

来自客户端的初始连接注册与该对象关联的侦听器
发送到ws服务器的消息被监听器接收并正确发送
断开 websocket - 创建新连接 - 创建新侦听器
发送到 ws 服务器的消息被相同的原始侦听器拾取并使用该连接的客户端而不是新注册的客户端
发送失败（因为客户端和 ws 连接不存在）并且没有进一步处理

如果客户端被删除，我似乎只需要删除客户端的侦听器，但我还没有找到一个好的方法，因为虽然我在调试器中看到侦听器具有关联的连接客户端对象，如果不为此添加代码，我将无法检索它们。

我是否正确地观察到了这一点，有什么好方法可以使它正常工作？

【问题讨论】：

标签： java websocket redis redisson

【解决方案1】：

当我写这个问题时，我有点倾向于一个我想到的答案并尝试过，这很有效。我添加了一个 ConcurrentHashmap 来跟踪连接的客户端和侦听器之间的关系。在我处理指向客户端删除的 websocket 错误的逻辑中，然后我删除了关联的侦听器（以及地图中的条目）。现在它按预期工作了。

小sn-p：

int listenerId = subcriberTopic.addListener(Message.class, listener);
clientListeners.put(q,(Integer)listenerId);

然后在触发清理的 websocket onError 处理程序中：

// remove the associated listener
int listenerIdForClient = MessageContainer.clientListeners.get(cP);
MessageContainer.subcriberTopic.removeListener((Integer) listenerIdForClient);
// remove entry from map
MessageContainer.clientListeners.remove(cP);

现在侦听器已正确清理，下次创建新侦听器并处理消息时。

【讨论】：