使用Jedis实现Redis内部通信时publish命令报错的问题排查

一、报错原因 #

Redis的连接一旦进入订阅模式（调用subscribe/psubscribe后），就会进入特殊的协议状态：这个连接只能执行和订阅相关的命令，比如SUBSCRIBE/UNSUBSCRIBE/PING/QUIT，任何其他命令（比如PUBLISH、GET、SET等）都会被Redis拒绝，这就是你看到ERR Can't execute 'publish'错误的原因。

你用了同一个Jedis实例既做订阅监听，又调用publish发消息，相当于在一个已经处于订阅状态的连接上执行非订阅命令，直接触发了Redis的协议限制。

二、解决发布报错的方法 #

核心是拆分连接：给订阅和发布分别创建独立的Jedis实例，两个连接互不干扰：

// 1. 订阅专用连接，单独开线程运行监听
Jedis subscribeJedis = new Jedis("localhost", 6379);
new Thread(() -> subscribeJedis.subscribe(new CustomPubSub(), "user_update")).start();

// 2. 发布专用连接，专门用来发消息
Jedis publishJedis = new Jedis("localhost", 6379);
// 发布消息时用这个实例
publishJedis.publish("user_update", "user_123_updated");

// 自定义订阅处理类
class CustomPubSub extends JedisPubSub {
    @Override
    public void onMessage(String channel, String message) {
        System.out.println(channel + " -> " + message);
        // 这里添加清除缓存的逻辑
    }
}

三、实现只接收其他服务器的消息 #

要过滤掉自己发送的消息，关键是给每个后端实例分配唯一标识，发布消息时带上这个标识，订阅端收到消息后判断发送者是否是自己，是则忽略：

1. 给每个实例生成唯一ID（比如启动时生成UUID，或者从配置读取实例编号）
2. 发布消息时，把业务内容和自身ID打包发送
3. 订阅端解析消息后，对比发送者ID和自身ID，不一致才处理

示例代码：

// 每个实例的唯一标识，启动时生成一次
private static final String INSTANCE_ID = UUID.randomUUID().toString();

// 发布端：打包消息内容和发送者ID
public void publishUserUpdate(String userId) {
    // 用JSON格式打包，实际项目建议用Jackson/Gson等工具类
    String message = String.format("{\"userId\":\"%s\",\"senderId\":\"%s\"}", userId, INSTANCE_ID);
    publishJedis.publish("user_update", message);
}

// 订阅端：解析并过滤消息
class CustomPubSub extends JedisPubSub {
    private final String ownInstanceId;

    public CustomPubSub(String ownInstanceId) {
        this.ownInstanceId = ownInstanceId;
    }

    @Override
    public void onMessage(String channel, String message) {
        // 解析JSON消息
        JSONObject msgJson = new JSONObject(message);
        String senderId = msgJson.getString("senderId");
        String userId = msgJson.getString("userId");

        // 只处理其他实例发送的消息
        if (!ownInstanceId.equals(senderId)) {
            System.out.println("收到其他实例的通知，清除用户" + userId + "的缓存");
            // 执行清除缓存的逻辑
        }
    }
}

// 初始化订阅时传入自身ID
new Thread(() -> subscribeJedis.subscribe(new CustomPubSub(INSTANCE_ID), "user_update")).start();

内容的提问来源于stack exchange，提问作者Lvkas_