Java实现redis消息订阅/发布（PubSub）

引入依赖

首先在你的java项目中引入

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

监听代码

订阅主题所需的类

import redis.clients.jedis.JedisPubSub;

public class ChannelListener extends JedisPubSub {
	public ChannelListener() {
	}

	@Override
	public void onMessage(String channel, String message) { // 收到消息会调用
		// 下面可以写你的业务处理代码
		System.out.println(String.format("收到消息成功！ channel： %s, message： %s", channel, message));
		if (message.equals("close"))
			this.unsubscribe("可填,不填就全部主题关闭订阅了");
	}

	@Override
	public void onSubscribe(String channel, int subscribedChannels) { // 订阅频道会调用
		System.out.println(String.format("订阅频道成功！ channel： %s, subscribedChannels %d", channel, subscribedChannels));
	}

	@Override
	public void onUnsubscribe(String channel, int subscribedChannels) { // 取消订阅会调用
		System.out.println(String.format("取消订阅频道！ channel： %s, subscribedChannels： %d", channel, subscribedChannels));

	}

}

测试代码

public static void main(String[] args) {
	JedisPoolConfig config = new JedisPoolConfig();
	config.setMaxTotal(1000);
	config.setMaxIdle(300);
	config.setMaxWaitMillis(1000);
	JedisPool jedisPool = new JedisPool(config, "ip", 端口, 可填, 密码==有?填:null);

	// 会阻塞,所以使用线程打开
	ExecutorService service = Executors.newCachedThreadPool();
	service.execute(new Runnable() {
		@Override
		public void run() {
			// TODO Auto-generated method stub
			Jedis jedis = jedisPool.getResource();
			ChannelListener c=	new ChannelListener();
			try {
				System.out.println("打开订阅线程,执行下句代码后此线程会阻塞");
				jedis.subscribe(c, "test");
				System.out.println("关闭订阅时才会打印");
			} finally {
				// TODO: handle finally clause
				System.out.println("线程关闭");
				//取消订阅
				jedis.close();
			}

		}
	});
	service.execute(new Runnable() {
		@Override
		public void run() {
			// TODO Auto-generated method stub
			Jedis jedis = jedisPool.getResource();
			try {
				// 每秒发送一条消息到对应主题
				for (int i = 0; i < 10; i++) {
					jedis.publish("test", String.valueOf(i));
					try {
						Thread.sleep(1000);
					} catch (InterruptedException e) {
						// TODO Auto-generated catch block
						e.printStackTrace();
					}
				}

			} finally {
				// TODO: handle finally clause
				System.out.println("关闭连接");
				jedis.close();
			}

		}
	});
	try {
		Thread.sleep(11000);
	} catch (InterruptedException e) {
		// TODO Auto-generated catch block
		e.printStackTrace();
	}
	// 关闭线程,即使你换成now来强制关闭,如果没有下面的通知订阅关闭代码,jvm是无法关闭的
	service.shutdown();
       //通知订阅关闭
	jedisPool.getResource().publish("test", "close");
	// 关闭线程池
	jedisPool.close();
}

输出结果

控制台:

)

服务器:

redis-cli后有密码就auth你的密码,没有就 subscribe 你需要订阅的主题

)

总结

订阅很愉快,写码需谨慎.里面有阻塞代码块在,需要做好此代码的处理,不然是无法优雅的关闭线程的.

赏

谢谢你请我吃糖果

支付宝

微信