欢迎您访问程序员文章站本站旨在为大家提供分享程序员计算机编程知识!
您现在的位置是: 首页  >  IT编程

Redis的消息发布与订阅

程序员文章站 2022-07-02 15:22:50
Redis 作为一个publish/subscribe server,起到了消息路由的功能。订阅者可以通过subscribe和psubscribe命令向Redis server订阅自己感兴趣的消息类型,当发布者通过publish命令向Redis server发送特定类型的消息时。订阅该消息类型的全部 ......

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

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

Redis的消息发布与订阅

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

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

publish:
发送消息:redis采用publish命令发送消息,其返回值为接收到该消息的订阅者的数量。 

subscribe:
订阅某个频道:redis采用subscribe命令订阅某个频道,其返回值包括客户端订阅的频道,目前已订阅的频道数量,以及接收到的消息,其中subscribe表示已经成功订阅了某个频道。 

psubscribe:
模式匹配:模式匹配功能允许客户端订阅符合某个模式的频道,redis采用psubscribe订阅符合某个模式所有频道,用“”表示模式,“”可以被任意值代替。
 

案例一:一个消息生产者,两个消息消费者

Redis的消息发布与订阅

案例二:两个消息生产者,一个消息消费者

Redis的消息发布与订阅

案例三:redis消息机制的java api

添加依赖:

<dependency>
    <groupid>redis.clients</groupid>
    <artifactid>jedis</artifactid>
    <version>3.1.0</version>
</dependency> 

 

消息监听器类:

import redis.clients.jedis.jedispubsub;
 
public class redismsgpubsublistener extends jedispubsub {
    @override
    public void unsubscribe() {
        super.unsubscribe();
    }
 
    @override
    public void unsubscribe(string... channels) {
        super.unsubscribe(channels);
    }
 
    @override
    public void subscribe(string... channels) {
        super.subscribe(channels);
    }
 
    @override
    public void psubscribe(string... patterns) {
        super.psubscribe(patterns);
    }
 
    @override
    public void punsubscribe() {
        super.punsubscribe();
    }
 
    @override
    public void punsubscribe(string... patterns) {
        super.punsubscribe(patterns);
    }
 
    @override
    public void onmessage(string channel, string message) {
        system.out.println("channel:" + channel + "receives message :" + message);
        this.unsubscribe();
    }
 
    @override
    public void onpmessage(string pattern, string channel, string message) {
 
    }
 
    @override
    public void onsubscribe(string channel, int subscribedchannels) {
        system.out.println("channel:" + channel + "is been subscribed:" + subscribedchannels);
    }
 
    @override
    public void onpunsubscribe(string pattern, int subscribedchannels) {
 
    }
 
    @override
    public void onpsubscribe(string pattern, int subscribedchannels) {
 
    }
 
    @override
    public void onunsubscribe(string channel, int subscribedchannels) {
        system.out.println("channel:" + channel + "is been unsubscribed:" + subscribedchannels);
    }
}

 

测试程序:

import redis.clients.jedis.jedis;

public class testmain {

   @test
    public void testsubscribe() throws exception{
        jedis jedis = new jedis("localhost");
        redismsgpubsublistener listener = new redismsgpubsublistener();
        jedis.subscribe(listener, "redischattest");
        //other code
    }

    @test
    public void testpublish() throws exception{
        jedis jedis = new jedis("localhost");
        jedis.publish("redischattest", "hello world");
        thread.sleep(5000);
        jedis.publish("redischattest", "hello redis");
    }
}
转自:https://www.cnblogs.com/collen7788/p/12849908.html