Spring Boot 整合 Redis消息订阅与发布

Posted 2021-02-21 橘子味儿的猫

 

一、Redis 中的发布订阅功能

 

 

 

与 Redis 发布订阅相关的命令有 6 个，分别如下：

• PSUBSCRIBE pattern [pattern …]：订阅一个或者多个符合pattern格式的频道

• PUBLISH channel message：发布消息到chanel中

• PUBSUB subcommand [argument [argument …]]：查看订阅与发布系统状态

• PUNSUBSCRIBE [pattern [pattern …]]：退订所有符合格式的频道

• SUBSCRIBE channel [channel …]：订阅一个或者多个频道

• UNSUBSCRIBE [channel [channel …]]：取消订阅频道

 

而在 Jedis 中，也提供了一个类 JedisPubSub，用来对订阅的 channel 进行监听。

• onPMessage：监听到订阅模式接受到消息时的回调

• onMessage：监听到订阅频道接受到消息时的回调

• onSubscribe：订阅频道时的回调

• onUnsubscribe：取消订阅频道时的回调

• onPSubscribe：订阅频道模式时的回调

• onPUnsubscribe：取消订阅模式时的回调

 

二、项目具体实现

1. POM依赖包

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

 

2.消息推送

package com.zyu.controller;

import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;
import redis.clients.jedis.Jedis;

/**
 * @Auther: zyu
 * @Date: 2020/4/29 10:18
 * @Description:   发布订阅 SUB端
 */
@RestController
public class SubController {


    @GetMapping("/sub")
    public String messageSub(String message) {

        Jedis jedis = new Jedis();
        try{
            //发送广播
            jedis.publish("redisChat", message);
        }catch(Exception e){
            e.printStackTrace();
        }finally{
            jedis.close();
        }
        return "SUCCESS";
    }

}

 

3.消息接收

package com.zyu.task;

import org.springframework.stereotype.Component;
import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisPubSub;

/**
 * @Auther: zyu
 * @Date: 2020/4/29 09:03
 * @Description: Redis消息监听
 */
@Component
public class Receiver {

    public void receiveMessage() {

        Jedis jedis = new Jedis();
        JedisPubSub jedisPubSub = new JedisPubSub() {

            /**
             * 监听到订阅频道接受到消息时的回调
             * @param channel   频道
             * @param message   消息
             */
            @Override
            public void onMessage(String channel, String message) {
                System.out.println("Channel:" + channel);
                System.out.println("Message:" + message);
            }

            /**
             * 监听到订阅模式接受到消息时的回调
             * @param pattern
             * @param channel
             * @param message
             */
            @Override
            public void onPMessage(String pattern, String channel, String message) {
                // TODO Auto-generated method stub
                System.out.println(pattern + "," + channel + "," + message);
            }

            /**
             * 订阅频道时的回调
             * @param channel
             * @param subscribedChannels
             */
            @Override
            public void onSubscribe(String channel, int subscribedChannels) {
                // TODO Auto-generated method stub
                System.out.println("onSubscribe: channel[" + channel + "]," + "subscribedChannels[" + subscribedChannels + "]");
            }

            /**
             * 取消订阅频道时的回调
             * @param channel
             * @param subscribedChannels
             */
            @Override
            public void onUnsubscribe(String channel, int subscribedChannels) {
                // TODO Auto-generated method stub
                System.out.println(
                        "onUnsubscribe: channel[" + channel + "], " + "subscribedChannels[" + subscribedChannels + "]");
            }

            /**
             * 取消订阅模式时的回调
             * @param pattern
             * @param subscribedChannels
             */
            @Override
            public void onPUnsubscribe(String pattern, int subscribedChannels) {
                // TODO Auto-generated method stub
                System.out.println("onPUnsubscribe: pattern[" + pattern + "]," +
                        "subscribedChannels[" + subscribedChannels + "]");
            }

            /**
             * 订阅频道模式时的回调
             * @param pattern
             * @param subscribedChannels
             */
            @Override
            public void onPSubscribe(String pattern, int subscribedChannels) {
                System.out.println("onPSubscribe: pattern[" + pattern + "], " +
                        "subscribedChannels[" + subscribedChannels + "]");
            }

        };

        // 订阅一个或者多个符合pattern格式的频道
        String[] channels = {"redisChat"};
        jedis.subscribe(jedisPubSub, channels);
    }

}

 

4.Spring容器初始化时，将监听类注入

package com.zyu;

import com.zyu.task.Receiver;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.context.ConfigurableApplicationContext;

@SpringBootApplication
public class SpringbootRedisApplication {

    public static void main(String[] args) {
        ConfigurableApplicationContext run = SpringApplication.run(SpringbootRedisApplication.class, args);
        run.getBean(Receiver.class).receiveMessage();
    }

}