浅谈java实现redis的发布订阅（简单易懂）

<dependency>
<groupId>redis.clients</groupId>
<artifactId>jedis</artifactId>
<version>2.9.0</version>
</dependency>

public class Publisher extends Thread{

  private final JedisPool jedisPool;

  public Publisher(JedisPool jedisPool) {
    this.jedisPool = jedisPool;
  }
  
  @Override
  public void run() {
    BufferedReader reader = new BufferedReader(new InputStreamReader(System.in));
    Jedis jedis = jedisPool.getResource();  //连接池中取出一个连接
    while (true) {
      String line = null;
      try {
        line = reader.readLine();
        if (!"quit".equals(line)) {
          jedis.publish("mychannel", line);  //从 mychannel 的频道上推送消息
        } else {
          break;
        }
      } catch (IOException e) {
        e.printStackTrace();
      }
    }
  }
}

public class Subscriber extends JedisPubSub {
  public Subscriber(){}
  @Override
  public void onMessage(String channel, String message) {    //收到消息会调用
    System.out.println(String.format("receive redis published message, channel %s, message %s", channel, message));
  }
  @Override
  public void onSubscribe(String channel, int subscribedChannels) {  //订阅了频道会调用
    System.out.println(String.format("subscribe redis channel success, channel %s, subscribedChannels %d",
        channel, subscribedChannels));
  }
  @Override
  public void onUnsubscribe(String channel, int subscribedChannels) {  //取消订阅 会调用
    System.out.println(String.format("unsubscribe redis channel, channel %s, subscribedChannels %d",
        channel, subscribedChannels));

  }
}

public class SubThread extends Thread {
  private final JedisPool jedisPool;
  private final Subscriber subscriber = new Subscriber();

  private final String channel = "mychannel";

  public SubThread(JedisPool jedisPool) {
    super("SubThread");
    this.jedisPool = jedisPool;
  }

  @Override
  public void run() {
    System.out.println(String.format("subscribe redis, channel %s, thread will be blocked", channel));
    Jedis jedis = null;
    try {
      jedis = jedisPool.getResource();  //取出一个连接
      jedis.subscribe(subscriber, channel);  //通过subscribe 的api去订阅，入参是订阅者和频道名
    } catch (Exception e) {
      System.out.println(String.format("subsrcibe channel error, %s", e));
    } finally {
      if (jedis != null) {
        jedis.close();
      }
    }
  }
}

public class PubSubDemo {
  public static void main( String[] args )
  {
    // 连接redis服务端
    JedisPool jedisPool = new JedisPool(new JedisPoolConfig(), "127.0.0.1", 6379);
    
    System.out.println(String.format("redis pool is starting, redis ip %s, redis port %d", "127.0.0.1", 6379));

    SubThread subThread = new SubThread(jedisPool); //订阅者
    subThread.start();

    Publisher publisher = new Publisher(jedisPool);  //发布者
    publisher.start();
  }
}