redis订阅发布示例

下面的代码使用redis实现订阅/发布。需要jedis的2.0.0和SLF4J的jar包。

在redis中可以同时订阅多个频道，如果有消息发布是redis会发出通知。jedis中提供了JedisPubSub抽象类来提供发布/订阅的机制，在实际应用中需要实现JedisPubSub类。

 

package practice.redis.pubsub;

import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisPubSub;

/**
 * 订阅
 * @author 15031483
 *
 */
public class Subscriber extends JedisPubSub {

	@Override
	public void onMessage(String channel, String message) {
		System.out.println("onMessage1: channel[" + channel + "], message["
				+ message + "]");
	}

	@Override
	public void onPMessage(String pattern, String channel, String message) {
		System.out.println("onPMessage2: channel[" + channel + "], message["
				+ message + "]");
	}

	@Override
	public void onSubscribe(String channel, int subscribedChannels) {
		System.out.println("onSubscribe3: channel[" + channel + "]," +
		"subscribedChannels[" + subscribedChannels + "]");
	}

	@Override
	public void onUnsubscribe(String channel, int subscribedChannels) {
		System.out.println("onUnsubscribe4: channel[" + channel + "], " +
		"subscribedChannels[" + subscribedChannels + "]");
	}

	@Override
	public void onPUnsubscribe(String pattern, int subscribedChannels) {
		System.out.println("onPUnsubscribe5: pattern[" + pattern + "]," +
		"subscribedChannels[" + subscribedChannels + "]");
	}

	@Override
	public void onPSubscribe(String pattern, int subscribedChannels) {
		System.out.println("onPSubscribe6: pattern[" + pattern + "], " +
		"subscribedChannels[" + subscribedChannels + "]");
	}

	public static void main(String[] args) {
		Jedis jr = null;
		try {
			jr = new Jedis("127.0.0.1", 6379, 0);// redis服务地址和端口号
			Subscriber sp = new Subscriber();
			sp.proceed(jr.getClient(), "news.share");
//			sp.proceed(jr.getClient(), "news.share", "news.blog");
			// sp.proceedWithPatterns(jr.getClient(), "news.*");
		} catch (Exception e) {
			e.printStackTrace();
		} finally {
			if (jr != null) {
				jr.disconnect();
			}
		}
	}

}

 

package practice.redis.pubsub;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import redis.clients.jedis.Jedis;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;

/**
 * 发布
 * @author 15031483
 *
 */
public class Publisher {

    private static final Logger logger = LoggerFactory.getLogger(Publisher.class);

    private final Jedis publisherJedis;

    private final String channel;

    public Publisher(Jedis publisherJedis, String channel) {
        this.publisherJedis = publisherJedis;
        this.channel = channel;
    }

    public void start() {
        logger.info("Type your message (quit for terminate)");
        try {
            BufferedReader reader = new BufferedReader(new InputStreamReader(System.in));
            while (true) {
                String line = reader.readLine();
                if (!"quit".equals(line)) {
                    publisherJedis.publish(channel, line);
                } else {
                    break;
                }
            }

        } catch (IOException e) {
            logger.error("IO failure while reading input, e");
        }
    }
}

 

package practice.redis.pubsub;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisPool;
import redis.clients.jedis.JedisPoolConfig;

/**
 * 测试reids发布订阅
 * @author 15031483
 *
 */
public class TestPubSub {
	public static final String CHANNEL_NAME = "commonChannel";

	private static Logger logger = LoggerFactory.getLogger(TestPubSub.class);

	public static void main(String[] args) throws Exception {
		JedisPoolConfig poolConfig = new JedisPoolConfig();
		JedisPool jedisPool = new JedisPool(poolConfig, "localhost", 6379, 0);
		final Jedis subscriberJedis = jedisPool.getResource();
		final Subscriber subscriber = new Subscriber();

		new Thread(new Runnable() {
			@Override
			public void run() {
				try {
					logger.info("Subscribing to \"commonChannel\". This thread will be blocked.");
					subscriberJedis.subscribe(subscriber, CHANNEL_NAME);
					logger.info("Subscription ended.");
				} catch (Exception e) {
					logger.error("Subscribing failed.", e);
				}
			}
		}).start();

		Jedis publisherJedis = jedisPool.getResource();

		new Publisher(publisherJedis, CHANNEL_NAME).start();

		subscriber.unsubscribe();
		jedisPool.returnResource(subscriberJedis);
		jedisPool.returnResource(publisherJedis);
	}
}

 

 

• 要在单独的线程中订阅, 因为subscribe会阻塞当前线程的执行。 你可以使用一个PubSub实例来订阅多个Channel
• 注意Jedis实例是非线程安全的
• 当需要创建多个jedis实例时要使用JedisPool类，当使用完jedis对象时要放回JedisPool。