我有一个应用程序，它使用 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 并且我无法使用该线程订阅任何频道之外，这将起作用。
【问题讨论】：
标签： java multithreading redis jedis
我观察到同样的问题，即订阅线程在您订阅后会阻塞。为了解决这个问题，我使用 Netty 实现了一个优化的 pub/sub 客户端，并将其合并到 Jedis fork here 中。这不是一个全面的解决方案，我还没有时间真正完成它，但它适用于基本的频道和模式订阅。基础是：
使用以下方法获取 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 永远不会阻塞，事件会异步传递给注册的 SubListener。
fork 现在有点老了，所以它的当前形式可能对您没有用处，但您可以轻松地从该包中提取源代码并独立构建它。依赖项是 Jedis 和 Netty。
抱歉，我没有更全面的解决方案。
【讨论】：