查看: 1428|回复: 0

[Java语言] 浅谈java实现redis的发布订阅(简单易懂)

发表于 2018-4-16 08:00:05

redis的应用场景实在太多了,现在介绍一下它的几大特性之一 发布订阅(pub/sub)。

特性介绍:

什么是redis的发布订阅(pub/sub)? Pub/Sub功能(means Publish, Subscribe)即发布及订阅功能。基于事件的系统中,Pub/Sub是目前广泛使用的通信模型,它采用事件作为基本的通信机制,提供大规模系统所要求的松散耦合的交互模式:订阅者(如客户端)以事件订阅的方式表达出它有兴趣接收的一个事件或一类事件;发布者(如服务器)可将订阅者感兴趣的事件随时通知相关订阅者。熟悉设计模式的朋友应该了解这与23种设计模式中的观察者模式极为相似。

同样,Redis的pub/sub是一种消息通信模式,主要的目的是解除消息发布者和消息订阅者之间的耦合, Redis作为一个pub/sub的server, 在订阅者和发布者之间起到了消息路由的功能。

如果没听懂上述的专业解释,没关系,其实我也没太听懂。

简单来讲,这里面还有个channel的概念,这里就是频道的意思,比如你订阅了银行的频道,当你的资金发生变动时,你就会接受到银行就会通过它的频道给你发送信息,在这里,你是属于被动接收的,而不是向银行索要信息,这个例子中,你就是sub(订阅者),而银行就是pub(发布者)。

项目运用场景:

一直都认为你会一样技术之前,都必须先明白这样一种技术在哪些地方会被用到,不能盲目的学东西。

看到发布订阅的特性,用来做一个简单的实时聊天系统再适合不过了。这是其中之一,当然这样的东西,我们开发中很少涉及到。再举一个常用的,在我们的分布式架构中,常常会遇到读写分离的场景,在写入的过程中,就可以使用redis发布订阅,使得写入值及时发布到各个读的程序中,就保证数据的完整一致性。再比如,在一个博客网站中,有100个粉丝订阅了你,当你发布新文章,就可以推送消息给粉丝们拉。总之场景很多,需要去挖掘。。

回顾java如何操作redis:

redis是一种缓存数据库,它也是C/S的结构,也就是客户端和服务端,一般来说,在java中,我们通常使用 jedis(客户端)去操作redis(服务端),这其中操作的时候,两者之间肯定要建立连接,就像数据库链接一样,在关系型数据库中,我们一般都维护一个连接池,以达到链接的复用,来省去建立连接和关闭连接的时间。所以在jedis中,同样也存在一个jedispool(jedis连接池)的概念,我们都是从池中去取连接使用。

上代码:

想使用jedis先引入依赖

  1. <dependency>
  2. <groupId>redis.clients</groupId>
  3. <artifactId>jedis</artifactId>
  4. <version>2.9.0</version>
  5. </dependency>
复制代码

建立一个Publisher (发布者)

  1. public class Publisher extends Thread{
  2. private final JedisPool jedisPool;
  3. public Publisher(JedisPool jedisPool) {
  4. this.jedisPool = jedisPool;
  5. }
  6. @Override
  7. public void run() {
  8. BufferedReader reader = new BufferedReader(new InputStreamReader(System.in));
  9. Jedis jedis = jedisPool.getResource(); //连接池中取出一个连接
  10. while (true) {
  11. String line = null;
  12. try {
  13. line = reader.readLine();
  14. if (!"quit".equals(line)) {
  15. jedis.publish("mychannel", line); //从 mychannel 的频道上推送消息
  16. } else {
  17. break;
  18. }
  19. } catch (IOException e) {
  20. e.printStackTrace();
  21. }
  22. }
  23. }
  24. }
复制代码

再建立一个订阅者

  1. public class Subscriber extends JedisPubSub {
  2. public Subscriber(){}
  3. @Override
  4. public void onMessage(String channel, String message) { //收到消息会调用
  5. System.out.println(String.format("receive redis published message, channel %s, message %s", channel, message));
  6. }
  7. @Override
  8. public void onSubscribe(String channel, int subscribedChannels) { //订阅了频道会调用
  9. System.out.println(String.format("subscribe redis channel success, channel %s, subscribedChannels %d",
  10. channel, subscribedChannels));
  11. }
  12. @Override
  13. public void onUnsubscribe(String channel, int subscribedChannels) { //取消订阅 会调用
  14. System.out.println(String.format("unsubscribe redis channel, channel %s, subscribedChannels %d",
  15. channel, subscribedChannels));
  16. }
  17. }
复制代码

这里订阅者需要继承JedisPubSub,来重写它的三个方法。用途 注释上已经写了,很简单。

我们这里只是定义了一个订阅者,下面去订阅频道。

  1. public class SubThread extends Thread {
  2. private final JedisPool jedisPool;
  3. private final Subscriber subscriber = new Subscriber();
  4. private final String channel = "mychannel";
  5. public SubThread(JedisPool jedisPool) {
  6. super("SubThread");
  7. this.jedisPool = jedisPool;
  8. }
  9. @Override
  10. public void run() {
  11. System.out.println(String.format("subscribe redis, channel %s, thread will be blocked", channel));
  12. Jedis jedis = null;
  13. try {
  14. jedis = jedisPool.getResource(); //取出一个连接
  15. jedis.subscribe(subscriber, channel); //通过subscribe 的api去订阅,入参是订阅者和频道名
  16. } catch (Exception e) {
  17. System.out.println(String.format("subsrcibe channel error, %s", e));
  18. } finally {
  19. if (jedis != null) {
  20. jedis.close();
  21. }
  22. }
  23. }
  24. }
复制代码

最后,再写一个测试类去跑一下。键盘输入消息,订阅者就会触发onMessage方法

  1. public class PubSubDemo {
  2. public static void main( String[] args )
  3. {
  4. // 连接redis服务端
  5. JedisPool jedisPool = new JedisPool(new JedisPoolConfig(), "127.0.0.1", 6379);
  6. System.out.println(String.format("redis pool is starting, redis ip %s, redis port %d", "127.0.0.1", 6379));
  7. SubThread subThread = new SubThread(jedisPool); //订阅者
  8. subThread.start();
  9. Publisher publisher = new Publisher(jedisPool); //发布者
  10. publisher.start();
  11. }
  12. }
复制代码

看打印结果

附上代码地址 https://github.com/fangyong1421/redis

以上就是本文的全部内容,希望对大家的学习有所帮助,也希望大家多多支持程序员之家。



回复

使用道具 举报