使用redis作为消息队列的用法( 二 ) _redis

另一个进程使用RPOP往头部取出元素来执行，这一端也叫消费者。
如果仅仅是这种方式来实现队列，它就是需要进程不断地循环队列，判断队列是不是有新元素，有的话就取出来执行，没有的话，就继续循环，但是这个总有一个时间间隔，你总得规定每隔一段时间去循环，虽然这个时间很小，但总有延迟，这种方式叫作轮循。有没有一种方式就是让不断执行一个redis命令，而redis中的列队有值就会通过命令通知程序呢？有的，那就是阻塞操作的RPOP，它叫作BRPOP 。
我们来演示一下它是如何实现的。
$ redis-cli127.0.0.1:6379> BRPOP list1 0先执行BRPOP，假如队列list1没有值，它会返回nil，并且阻塞在那，在等另一个程序或进程往list1中填值。
我们开启另一个redis端终。
$ redis-cli127.0.0.1:6379> LPUSH list1 a(integer) 1我们再来看之前的结果。
127.0.0.1:6379> BRPOP list1 01) "list1"2) "a"(16.99s)这样就能把列表的值给取到了。
优点

能够实现持久化
采用 Master－Slave 数据复制模式。队列操作都是写操作，Master任务繁重，能让Slave分担的持久化工作，就不要Master做。RDB和AOF两种方法都用上，多重保险。
支持集群
接口使用简单

不足

Redis上消息只会被一个消费者消费，不会有多个订阅者消费同一个消息，简单一对一
生产者或者消费者崩溃后的处理机制，需要自己实现
生产者写入太快，消费者消费太慢，导致Redis的内存问题，处理机制需要自己实现

通过pub/sub来实现
实现机制
订阅，取消订阅和发布实现了发布/订阅消息范式，发布者不是计划发送消息给特定的订阅者。而是发布的消息分到不同的频道，不需要知道什么样的订阅者订阅。订阅者对一个或多个频道感兴趣，只需接收感兴趣的消息，不需要知道什么样的发布者发布的。
这是一种基于非持久化的消息机制，消息发布者和订阅者必须同时在线，否则一旦消息订阅者由于各种异常情况而被迫断开连接，在其重新连接后，其离线期间的消息是无法被重新通知的（即发即弃）。
Redis中的消息可以提供两种不同的功能。一类是基于Channel的消息，这一类消息和Redis中存储的Keys没有太多关联，也就是说即使不在Redis中存储任何Keys信息，这类消息也可以独立使用。另一类消息可以对（也可以不对）Redis中存储的Keys信息的变化事件进行通知，可以用来向订阅者通知Redis中符合订阅条件的Keys的各种事件。
通过springboot 构建redis消息队列
首先springboot配置文件配置如下：
spring.redis.host=localhostspring.redis.port=6379消息生产者，注入redisTemplate,用convertAndSend发送消息
import org.springframework.beans.factory.annotation.Autowired;import org.springframework.data.redis.core.StringRedisTemplate;import org.springframework.stereotype.Service;@Servicepublic class PublishService { @Autowired private StringRedisTemplate stringRedisTemplate; public void sendMsg(String channel, String msg) { stringRedisTemplate.convertAndSend(channel, msg); }}消费者：创建一个接收消息的类，继承MessageListener，也可以不继承
import lombok.extern.slf4j.Slf4j;import org.springframework.stereotype.Service;@Slf4j@Servicepublic class RedisReceiver { public void receiveMessage(String message) { log.info("receive message is {}",message); }}消息订阅者配置类：
import com.wuzy.queue.RedisReceiver;import org.springframework.context.annotation.Bean;import org.springframework.data.redis.connection.RedisConnectionFactory;import org.springframework.data.redis.core.StringRedisTemplate;import org.springframework.data.redis.listener.PatternTopic;import org.springframework.data.redis.listener.RedisMessageListenerContainer;import org.springframework.data.redis.listener.adapter.MessageListenerAdapter;import org.springframework.stereotype.Component;/** * redis 监听配置 */@Configurationpublic class RedisSubListenerConfig { /** * 初始化监听器 * * @param connectionFactory * @param listenerAdapter * @return */ @Bean RedisMessageListenerContainer container(RedisConnectionFactory connectionFactory, MessageListenerAdapter listenerAdapter) { RedisMessageListenerContainer container = new RedisMessageListenerContainer(); container.setConnectionFactory(connectionFactory); container.addMessageListener(listenerAdapter, new PatternTopic("channel_1")); // new PatternTopic("这里是监听的通道的名字") 通道要和发布者发布消息的通道一致 return container; } /** * 绑定消息监听者和接收监听的方法 * * @param redisReceiver * @return */ @Bean MessageListenerAdapter listenerAdapter(RedisReceiver redisReceiver) { // redisReceiver 消息接收者 // receiveMessage 消息接收后的方法 MessageListenerAdapter messageListenerAdapter = new MessageListenerAdapter(); messageListenerAdapter.setDefaultListenerMethod("receiveMessage"); messageListenerAdapter.setDelegate(redisReceiver); return messageListenerAdapter; } @Bean StringRedisTemplate template(RedisConnectionFactory connectionFactory) { return new StringRedisTemplate(connectionFactory); }}