发布和订阅

Posted 2020-07-06 JavaRecorder

一、概述

---

1). 发布和订阅是一种消息通信模式。

2). 优点：使消息订阅者和消息发布者耦合度降低，类似设计模式中的观察者模式。

 

二、发布和订阅

---

订阅命令： 

// 订阅一个或多个频道
// 返回值：value_1 为 “subcribe” 表示订阅成功，value_2 为订阅的频道名称，value_3 表示当前订阅的频道个数
subscribe  channel1  channel2  channel3 ...
// 模式订阅
// 返回值：value_1 为 “psubcribe” 表示订阅成功，value_2 为订模式订阅的 pattern，value_3 表示当前订阅的频道个数
psubscribe abc*  xyz*  ...

发布命令：

// 发布命令
publish channel msg

启动订阅者 X 和订阅者 Y，分别发出订阅命令，如下面两张图片：

 

 

启动发布者，发布如下消息：

 

订阅者 X 和订阅者 Y 都收到了订阅消息：

 

取消订阅（取消订阅在官方给的客户端上是无法模拟的）

UNSUBSCRIBE  cctv-1
PUNSUBSCRIBE  cctv-*

 

查看订阅

// 查看订阅的所有频道
pubsub  channels;
pubsub  channels  msg*;
// 查看该频道的订阅人数
pubsub  numsub  channel1 channel2 ... ;

   

 

三、编程展示 Redis 的发布和订阅

---

编程展示：

public class MySub extends JedisPubSub{

    @Override
    public void onMessage(String channel, String msg) {
        System.out.println("onMessage - " + channel +" - " +msg);
    }

    @Override
    public void onPMessage(String pattern, String channel, String msg) {
        System.out.println("onPMessage - " + pattern +" - "+ channel +" - " +msg);
    }

    @Override
    public void onPSubscribe(String channel, int msg) {
        System.out.println("onPsubscribe - " + channel +" - "+ msg);
    }

    @Override
    public void onPUnsubscribe(String arg0, int arg1) {}

    @Override
    public void onSubscribe(String channel, int number) {
        System.out.println("onSubscribe - " + channel +" - "+ number);
        
    }

    @Override
    public void onUnsubscribe(String arg0, int arg1) {}
}

 

订阅者 A

public class SubClient {
    public static void main(String[] args) {
        MySub sub = new MySub();
        Jedis jedis = new Jedis("127.0.0.1", 6379);
        jedis.subscribe(sub, "cctv-1");
    }
}

 

订阅者 B （模式订阅） 

public class PsubClient {
    public static void main(String[] args) {
        MySub sub = new MySub();
        Jedis jedis = new Jedis("127.0.0.1", 6379);
        jedis.psubscribe(sub, "cctv*");
    }
}

 

发布者 C

public class PubClient {
    public static void main(String[] args) {
        Jedis jedis = new Jedis("127.0.0.1", 6379);
        jedis.publish("cctv-1", "hello,this is cctv-1");
    }
}

 

依次运行客户端 ABC

A 端输出

onSubscribe - cctv-1 - 1
onMessage - cctv-1 - hello,this is cctv-1

B 端输出

onPsubscribe - cctv* - 1
onPMessage - cctv* - cctv-1 - hello,this is cctv-1