Redis發佈與訂閱——PUBLISH & SUBSCRIBE

Redis發佈與訂閱——PUBLISH  & SUBSCRIBEjava

通常來講,發佈與訂閱(又稱pub/sub)的特色是訂閱者(listener)負責訂閱頻道(channel),發送者(publisher)負責向頻道發送二進制字符串消息(binary string message)。每當有消息發送至給定頻道時,頻道的訂閱者都會收到消息。咱們也能夠把頻道看做是電臺,其中訂閱者能夠同時收聽多個電臺,而發送者則能夠在任何電臺發送消息。redis

發佈與訂閱的模式

Redis 的 SUBSCRIBE 命令能夠讓客戶端訂閱任意數量的頻道, 每當有新信息發送到被訂閱的頻道時, 信息就會被髮送給全部訂閱指定頻道的客戶端。shell

做爲例子, 下圖展現了頻道 channel1 , 以及訂閱這個頻道的三個客戶端 —— client2 、 client5 和 client1 之間的關係:ide

當有新消息經過 PUBLISH 命令發送給頻道 channel1 時, 這個消息就會被髮送給訂閱它的三個客戶端:測試


發佈與訂閱相關命令

訂閱和發佈ui

首先打開一個cli,訂閱一個頻道,以下,this

127.0.0.1:7000> subscribe mychannel
Reading messages... (press Ctrl-C to quit)
1) "subscribe"
2) "mychannel"
3) (integer) 1

而後向該頻道發送一個消息,以下,spa

127.0.0.1:7000> publish mychannel 'hell world'
(integer) 1
127.0.0.1:7000>

剛纔的客戶端已經訂閱了該頻道,因此該頻道就會收到消息,以下,code

➜  ~ redis-cli -p 7000
127.0.0.1:7000> subscribe mychannel
Reading messages... (press Ctrl-C to quit)
1) "subscribe"
2) "mychannel"
3) (integer) 1
1) "message"
2) "mychannel"
3) "hell world"


其餘命令:字符串

UNSUBSCRIBE——退訂給定的一個或多個頻道,若是執行是沒有給定任何頻道,那麼退訂全部頻道

PSUBSCRIBE——PSUBSCRIBE pattern [pattern ...] 訂閱與給定模式相匹配的全部頻道

PUNSUBSCRIBE——PUNSUBSCRIBE  [pattern [pattern ...]] 退訂給定的模式,若是執行是沒有給定任何模式,那麼退訂全部模式。


使用Jedis客戶端實現發佈與訂閱

首先實現一個Listener,用於訂閱消息,接收消息,

package com.usoft.jedis.sample;

import redis.clients.jedis.JedisPubSub;

/**
 * Created by xinxingegeya on 16/4/13.
 */
public class MessageListener extends JedisPubSub {
    @Override
    public void onMessage(String channel, String message) {
        System.out.println("channel:" + channel + ",message:" + message);
        //此處咱們能夠取消訂閱
        if (message.equalsIgnoreCase("quit")) {
            this.unsubscribe(channel);
        }
    }
}

寫一個測試類,實現訂閱和發佈,

package com.usoft.jedis.sample;

import org.junit.After;
import org.junit.Before;
import org.junit.Test;
import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisPool;
import redis.clients.jedis.JedisPoolConfig;

/**
 * Created by xinxingegeya on 16/4/13.
 */
public class PubSubTest {
    /**
     * jedis鏈接池
     */
    public JedisPool jedisPool;
    /**
     * 發佈和訂閱的頻道
     */
    public String channel = "mychannel";

    @Before
    public void before() {
        JedisPoolConfig config = new JedisPoolConfig();
        config.setMaxTotal(10);
        config.setMaxIdle(5);
        config.setMaxWaitMillis(5000);
        config.setTestOnBorrow(true);
        jedisPool = new JedisPool(config, "127.0.0.1", 7000);
    }

    @After
    public void after() {
        jedisPool.close();
    }

    @Test
    public void subscribe() {
        Jedis jedis = jedisPool.getResource();
        jedis.subscribe(new MessageListener(), channel);
    }

    /**
     * 發佈9次消息後,在此發送quit消息,使listener(訂閱者)關閉
     */
    @Test
    public void publish() {
        Jedis jedis = jedisPool.getResource();
        for (int i = 0; i < 10; i++) {
            if (i == 9) {
                jedis.publish(channel, "quit");
            } else {
                jedis.publish(channel, "hello world");
            }
        }
    }
}

=========END=========

相關文章
相關標籤/搜索