springboot+redis实现发布订阅模式

引入依赖

  
      org.springframework.boot
      spring-boot-starter-data-redis
  

自定义监听器，处理订阅后收到的msg

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.connection.Message;
import org.springframework.data.redis.connection.MessageListener;
import org.springframework.data.redis.core.RedisTemplate;

public class ConsumerRedisListener implements MessageListener {


    @Autowired
    private RedisTemplate redisTemplate;

    @Override
    public void onMessage(Message message, byte[] pattern) {
        doBusiness(message);
    }

    
    public void doBusiness(Message message) {
        Object value = redisTemplate.getValueSerializer().deserialize(message.getBody());
        System.out.println("consumer message: " + value);
    }
}

定义一个top常量

public class RedisTopicConst {

    public static final String  SCHEDULE_TOPIC = "schedule-topic";

}

配置类

import com.xfr.consts.RedisTopicConst;
import com.xfr.listener.ConsumerRedisListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.redis.connection.lettuce.LettuceConnectionFactory;
import org.springframework.data.redis.listener.ChannelTopic;
import org.springframework.data.redis.listener.RedisMessageListenerContainer;


@Configuration
public class RedisConfig {


    
    @Autowired
    private LettuceConnectionFactory lettuceConnectionFactory;

    
    @Bean
    public ConsumerRedisListener consumerRedis() {
        return new ConsumerRedisListener();
    }

    
    @Bean
    public ChannelTopic topic() {
        return new ChannelTopic(RedisTopicConst.SCHEDULE_TOPIC);
    }

    
    @Bean
    public RedisMessageListenerContainer redisMessageListenerContainer() {
        RedisMessageListenerContainer container = new RedisMessageListenerContainer();
        container.setConnectionFactory(lettuceConnectionFactory);
        container.addMessageListener(consumerRedis(),topic());
        return container;

    }
}

测试一下，我这里是通过get请求跑接口测试的，也可以使用junit测试

import com.xfr.consts.RedisTopicConst;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;


@RestController
public class TestController {

    @Autowired
    private RedisTemplate redisTemplate;

    @GetMapping("test")
    public String testRedisPublishSubscribe(){
        // 发布msg
        redisTemplate.convertAndSend(RedisTopicConst.SCHEDULE_TOPIC, "5");
        return "success";
    }

}

最后看到控制台打印如下：
和我们自定义监听器那里对应起来

使用redis和mq做发布订阅模式的区别：

redis： 轻量级，低延迟，高并发，低可靠性；
rabbitmq：重量级，高可靠，异步，不保证实时；
具体看实际业务进行选择