Java实现Redis的订阅发布功能，亲测可以！！！

向右看齐 2023-03-12 14:28 22阅读 0赞

### 文章目录 ###

*  简介
 *  发布消息
 *  订阅发布处理
 *  订阅频道
 *  测试

# 简介 #

首先我们得知道，当一个用户订阅了某个频道，那么他就会一直在那监听该频道是否发出消息，是一种阻塞的状态，所以，我们把订阅功能用多线程来实现，给每一个订阅者都分配一个线程。

# 发布消息 #

![在这里插入图片描述][20200509101242750.png]  
发布消息这个方法比较简单，我们只需要指定像哪个频道发布什么内容就行了，这里我把他们封装到了一个util类下，用静态方法，方便使用。

public static void publishMsg(String channel, String message) {
            // Jedis jedis = getJedis();
            Jedis jedis = new Jedis("127.0.0.1",6379);
            try {
                jedis.publish(channel, message);
            } catch (Exception e) {
                System.out.println(e.getMessage());
            } finally {
                jedis.close();
            }
        }

# 订阅发布处理 #

这个类是用来处理订阅发布的，也就是当我们订阅了某个频道做什么处理，取消订阅某个频道做什么处理，收到了某条消息做什么处理。因为我把他们都封装到util类下了，所以这里用到内部类来实现。

public static JedisPubSub jedisPubSub = new JedisPubSub() {
            @Override
            public void onMessage(String channel, String message) {
                System.out.println("收到了" + channel + "频道发来消息：" +  message);
                System.out.println(channel + ":" + message);
            }
            
            @Override
            public void onSubscribe(String channel, int subscribedChannels) {
                System.out.println("订阅了" + channel + "频道");
                System.out.println(channel + ":" + subscribedChannels);
            }
    
            @Override
            public void onUnsubscribe(String channel, int subscribedChannels) {
                System.out.println("取消订阅了" + channel + "频道");
                System.out.println(channel + ":" + subscribedChannels);
            }
        };

# 订阅频道 #

![在这里插入图片描述][20200509101217383.png]  
订阅频道，我们只需要指定订阅了哪个频道即可，这里需要注意的是，我们还需要指定，当做出与该频道相关的操作时（订阅、取消订阅、收到频道消息等）我们需要怎么处理，这也就用到了上面那个类，当然了，因为这是一个阻塞的线程，所以用多线程处理。

/**
         * 接收消息。
         * 该方法调用后，会一直执行下去。当有发布对应消息时，就会在jedisPubSub中接收到！
         * @param channels
         */
        public static void subscribeMsg(String channels) {
            // Jedis jedis = getJedis();
            Jedis jedis = new Jedis ("127.0.0.1", 6379);
            new Thread(()->{
                try {
                    jedis.subscribe(jedisPubSub, channels);
                } catch (Exception e) {
                    System.out.println(e.getMessage());
                } finally {
                    // jedis.close();
                }
            }).start();
        }

# 测试 #

为了保证发布在订阅之后，这里睡眠100毫秒

public static void main(String[] args) {
            JedisUtil.subscribeMsg("test");
            try {
                Thread.sleep(100);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            JedisUtil.publishMsg("test","hello");
            JedisUtil.publishMsg("test","hello aaa");
    }

![在这里插入图片描述][20200509101103294.png]

[20200509101242750.png]: /images/20230312/40ff9d40db7d4e14bc4ec37e3666c90b.png
[20200509101217383.png]: /images/20230312/5aa88647c32b4f64afb3fc2a866356ec.png
[20200509101103294.png]: /images/20230312/2607d99326504b73acdfac9d4f32e076.png