【赵强老师】Redis的消息发布与订阅

  • Z时代
  • 2024-01-10
  • 分类：综合

database

Redis 作为一个publish/subscribe server，起到了消息路由的功能。订阅者可以通过subscribe和psubscribe命令向Redis server订阅自己感兴趣的消息类型，当发布者通过publish命令向Redis server发送特定类型的消息时。订阅该消息类型的全部client都会收到此消息。这里消息的传递是多对多的。一个client可以订阅多个channel,也可以向多个channel发送消息。

下图为大家展示了Redis消息机制的体系架构。

发布者和订阅者都是Redis客户端，Channel则为Redis服务器端，发布者将消息发送到某个的频道，订阅了这个频道的订阅者就能接收到这条消息。Redis的这种发布订阅机制与基于主题的发布订阅类似，Channel相当于主题。

下面列出来了Redis发布消息、订阅消息的相关命令。

publish：
发送消息：Redis采用PUBLISH命令发送消息，其返回值为接收到该消息的订阅者的数量。 
subscribe：
订阅某个频道：Redis采用SUBSCRIBE命令订阅某个频道，其返回值包括客户端订阅的频道，目前已订阅的频道数量，以及接收到的消息，其中subscribe表示已经成功订阅了某个频道。 
psubscribe：
模式匹配：模式匹配功能允许客户端订阅符合某个模式的频道，Redis采用PSUBSCRIBE订阅符合某个模式所有频道，用“”表示模式，“”可以被任意值代替。

 

案例一：一个消息生产者，两个消息消费者

案例二：两个消息生产者，一个消息消费者

案例三：Redis消息机制的Java API

添加依赖：

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

 

消息监听器类：

import redis.clients.jedis.JedisPubSub;
publicclass RedisMsgPubSubListener extends JedisPubSub {
    @Override
publicvoid unsubscribe() {
super.unsubscribe();
    }
    @Override
publicvoid unsubscribe(String... channels) {
super.unsubscribe(channels);
    }
    @Override
publicvoid subscribe(String... channels) {
super.subscribe(channels);
    }
    @Override
publicvoid psubscribe(String... patterns) {
super.psubscribe(patterns);
    }
    @Override
publicvoid punsubscribe() {
super.punsubscribe();
    }
    @Override
publicvoid punsubscribe(String... patterns) {
super.punsubscribe(patterns);
    }
    @Override
publicvoid onMessage(String channel, String message) {
        System.out.println("channel:" + channel + "receives message :" + message);
this.unsubscribe();
    }
    @Override
publicvoid onPMessage(String pattern, String channel, String message) {
    }
    @Override
publicvoid onSubscribe(String channel, int subscribedChannels) {
        System.out.println("channel:" + channel + "is been subscribed:" + subscribedChannels);
    }
    @Override
publicvoid onPUnsubscribe(String pattern, int subscribedChannels) {
    }
    @Override
publicvoid onPSubscribe(String pattern, int subscribedChannels) {
    }
    @Override
publicvoid onUnsubscribe(String channel, int subscribedChannels) {
        System.out.println("channel:" + channel + "is been unsubscribed:" + subscribedChannels);
    }
}

 

测试程序：

import redis.clients.jedis.Jedis;
publicclass TestMain {
   @Test
publicvoid testSubscribe() throws Exception{
        Jedis jedis = new Jedis("localhost");
        RedisMsgPubSubListener listener = new RedisMsgPubSubListener();
        jedis.subscribe(listener, "redisChatTest");
//other code
    }
    @Test
publicvoid testPublish() throws Exception{
        Jedis jedis = new Jedis("localhost");
        jedis.publish("redisChatTest", "Hello World");
        Thread.sleep(5000);
        jedis.publish("redisChatTest", "Hello Redis");
    }
}

以上是 【赵强老师】Redis的消息发布与订阅 的全部内容， 来源链接： utcz.com/z/533521.html

回到顶部