Redis实现消息队列

Jedis和RedisTemplate的区别

单机低并发Jedis
分布式高并发RedisTemplate

Redis实现消息队列

1. 通过List实现消息队列，lpush和lpop操作
消费者采用while，一直监听，浪费资源
使用brpop和blpop实现阻塞读取，会按照key的顺序进⾏读取，可以实现具有优先级的队列
优点：使用 List 实现消息队列的优点是消息可以被持久化，List 可以借助 Redis 本身的持久化功能，AOF 或者是 RDB 或混合持久化的方式，用于把数据保存至磁盘，这样当 Redis 重启之后，消息不会丢失。
缺点：但使用 List 同样存在一定的问题，比如消息不支持重复消费、没有按照主题订阅的功能、不支持消费消息确认等。

2. ZSet实现消息队列，它是利用 zadd 和 zrangebyscore 来实现存入和读取消息的。
优点：同样具备持久化的功能
缺点：List 存在的问题它也同样存在，不但如此，使用 ZSet 还不能存储相同元素的值。因为它是有序集合，有序集合的存储元素值是不能重复的，但分值可以重复，也就是说当消息值重复时，只能存储一条信息在 ZSet 中。
3. Redis发布订阅
此模式允许生产者只生产一次消息，由中间件负责将消息复制到多个消息队列，每个消息队列由对应的消费组消费。
优点：

可以按照主题订阅方式
典型的广播模式，一个消息可以发布到多个消费者。
多信道订阅，消费者可以同时订阅多个信道，从而接收多类消息。
缺点：
无法持久化保存消息，如果 Redis 服务器宕机或重启，那么所有的消息将会丢失；
发布订阅模式是“发后既忘”的工作模式，如果有订阅者离线重连之后就不能消费之前的历史消息；
不支持消费者确认机制，稳定性不能得到保证，例如当消费者获取到消息之后，还没来得及执行就宕机了。因为没有消费者确认机制，Redis 就会误以为消费者已经执行了，因此就不会重复发送未被正常消费的消息了，这样整体的 Redis 稳定性就被没有办法得到保障了。

可见，Pub/Sub 模式不适合做消息存储，消息积压类的业务，而是擅长处理广播，即时通讯，即时反馈的业务。

RedisTemplate实现发布订阅模式MQ

编写RedisConfig类

@Configuration
public class RedisConfig {
    
    @Bean
    MessageListenerAdapter listenerAdapter(RedisSubscribe redisSubscribe){
        return new MessageListenerAdapter(redisSubscribe,"receiveMsg");
    }

    @Bean
    RedisMessageListenerContainer container(RedisConnectionFactory redisConnectionFactory, MessageListenerAdapter listenerAdapter){
        RedisMessageListenerContainer container = new RedisMessageListenerContainer();
        container.setConnectionFactory(redisConnectionFactory);
        container.addMessageListener(listenerAdapter,new PatternTopic("redisPT"));
        container.addMessageListener(listenerAdapter,new ChannelTopic("redisCT"));
        return container;
    }

    @Bean
    public RedisTemplate<String, Object> redisTemplate(RedisConnectionFactory redisConnectionFactory) throws UnknownHostException {
        RedisTemplate<String, Object> template = new RedisTemplate();
        //连接工厂
        template.setConnectionFactory(redisConnectionFactory);
        template.afterPropertiesSet();
        return template;
    }
}

Producer

@Data
@Configuration
@EnableScheduling
public class RedisProducer {

    @Autowired
    private StringRedisTemplate redisTemplate;

    @Scheduled(fixedRate = 3000)
    private void SendMsg(){
        redisTemplate.convertAndSend("redisCT",String.valueOf(new Random().nextInt(100)));
    }

}

Subscriber

@Component
public class RedisSubscribe {
    public void receiveMsg(String msg){
        System.out.println("subsribe receive message:"+msg);
    }
}

执行程序：

最后编辑于：2022.06.29 13:56:25