使用同一线程Jedis订阅多个频道

小编典典

使用同一线程Jedis订阅多个频道

redis

我有一个使用Redis发布/订阅在Java中使用Jedis客户端在客户端之间传输消息的应用程序。我希望能够在用户键入命令时在运行时订阅频道，但是由于订阅是一个阻塞操作，因为它在调用订阅的线程上进行侦听，所以我不确定以后如何订阅其他频道在原始线程上。

例：

private PubSubListener psl = new PubSubListener();

public void onCommand(String[] args) {
    subscribe(args[0]);
}

public void subscribe(String channel) {
    Jedis jedis = jedisPool.getResource();

    jedis.subscribe(psl, channel);
}

这将起作用，除了调度命令的线程将用于轮询Redis，而我将无法使用该线程订阅更多的频道。

阅读 579

2020-06-20

共1个答案

小编典典

我观察到了同样的问题，即订阅后订阅线程就会阻塞。为了解决这个问题，我使用Netty实现了一个优化的发布/
订阅客户端，并将其合并到此处的Jedis分支中。这不是一个全面的解决方案，我还没有时间真正完成它，但是它适用于基本的频道和模式订阅。

使用以下命令获取pubsub实例：

public static OptimizedPubSub getInstance(String host, int port, String auth, long timeout)

使用以下方式发布/取消模式订阅：

public ChannelFuture psubscribe(String... patterns)
public ChannelFuture punsubscribe(String... patterns)

您可以忽略返回的ChannelFuture，除非您想100％确定您的请求通过了（这是异步的）。

使用以下方式发布/取消频道订阅：

public ChannelFuture subscribe(String... channels)
public ChannelFuture unsubscribe(String... channels)

然后实现SubListener实例：

public interface SubListener {
    /**
     * Callback when a message is published on a subscribed channel
     * @param channel The channel the message was received on
     * @param message The received message
     */
    public void onChannelMessage(String channel, String message);

    /**
     * Callback when a message is published on a subscribed channel matching a subscribed pattern
     * @param pattern The pattern that the channel matched
     * @param channel The channel the message was received on
     * @param message The received message
     */
    public void onPatternMessage(String pattern, String channel, String message);
}

并使用以下方法注册/取消注册侦听器：

public void registerListener(SubListener listener)
public void unregisterListener(SubListener listener)

OptimizedPubSub永远不会阻止，并且事件会异步传递到已注册的SubListeners。

叉子现在有点旧了，所以它在当前形式下可能对您没有用，但是您可以轻松地从该包中提取源代码并独立构建它。依赖关系是Jedis和Netty。

抱歉，我没有更全面的解决方案。

2020-06-20