Redis 发布订阅(pub/sub)是一种消息通信模式：发送者(pub)发送消息，订阅者(sub)接收消息。可以用于消息的传输，Redis的发布订阅机制包括三个部分，发布者，订阅者和Channel。适宜做在线聊天、消息推送等。
发布者和订阅者都是Redis客户端，Channel则为Redis服务器端，发布者将消息发送到某个的频道，订阅了这个频道的订阅者就能接收到这条消息，Redis 客户端可以订阅任意数量的频道。
下图展示了频道 channel1 ， 以及订阅这个频道的三个客户端 —— client2 、 client5 和 client1 之间的关系：
当有新消息通过 PUBLISH 命令发送给频道 channel1 时， 这个消息就会被发送给订阅它的三个客户端：
实例
以下实例演示了发布订阅是如何工作的。在我们实例中我们创建了订阅频道名为 redisChat:
先创建了订阅频道，才能创建发布频道进行添加
redis 127.0.0.1:6379> SUBSCRIBE redisChat
Reading messages... (press Ctrl-C to quit)
1) "subscribe"
2) "redisChat"
3) (integer) 1
现在，我们先重新开启个 redis 客户端，然后在同一个频道 redisChat 发布两次消息，订阅者就能接收到消息。
Redis 发布订阅命令
下表列出了 redis 发布订阅常用命令：
| PSUBSCRIBE pattern [pattern ...]
订阅一个或多个符合给定模式的频道。 |PUBSUB subcommand [argument [argument ...]]
查看订阅与发布系统状态。PUBLISH channel message
将信息发送到指定的频道。| PUNSUBSCRIBE [pattern [pattern ...]]
退订所有给定模式的频道。 || SUBSCRIBE channel [channel ...]
订阅给定的一个或多个频道的信息。 || UNSUBSCRIBE [channel [channel ...]]
指退订给定的频道。 |
java实现
可以通过spring-redis中的redisTemplate工具辅助实现
1.发布消息
/**
* redis发布消息
*
* @param channel
* @param message
*/
public void sendMessage(String channel, String message) {
redisTemplate.convertAndSend(channel, message);
}
直接使用convertAndSend方法即可向指定的通道发布消息
2.监听消息
监听消息需要两步,消息监听类并在xml中注册这个类
监听类有两种实现方式一种是实现org.springframework.data.redis.connection.MessageListener接口,实现onMessage方法示例代码如下:
@Component
public class RedisMessageListener implements MessageListener {
@Autowired
private RedisTemplate<String, String> redisTemplate;
private static Logger logger = Logger.getLogger(RedisMessageListener.class);
@Override
public void onMessage(Message message, byte[] pattern) {
byte[] body = message.getBody();// 请使用valueSerializer
byte[] channel = message.getChannel();
// 请参考配置文件，本例中key，value的序列化方式均为string。
// 其中key必须为stringSerializer。和redisTemplate.convertAndSend对应
String msgContent = (String) redisTemplate.getValueSerializer().deserialize(body);
String topic = (String) redisTemplate.getStringSerializer().deserialize(channel);
logger.info("redis--topic:" + topic + " body:" + msgContent);
}
}
也可以使用自己定义的类,方法名称自己定义,示例如下:
@Component
public class EventListener {
private static Logger logger = Logger.getLogger(EventListener.class);
public void getMessage(String message, String channel) {
logger.info(message);
}
}
这两中方式实现的不同在于注册监听器时的配置略有不同
redis配置文件:
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:p="http://www.springframework.org/schema/p" xmlns:redis="http://www.springframework.org/schema/redis"
xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-4.1.xsd http://www.springframework.org/schema/redis http://www.springframework.org/schema/redis/spring-redis-1.0.xsd">
<bean id="poolConfig" class="redis.clients.jedis.JedisPoolConfig">
<property name="maxIdle" value="${redis.maxIdle}" />
<!-- <property name="maxActive" value="${redis.maxActive}" /> -->
<property name="maxWaitMillis" value="${redis.maxWaitMillis}" />
<property name="testOnBorrow" value="${redis.testOnBorrow}" />
</bean>
<!--注意使用订阅发布时,此bean必须命名为redisCOnnectionFactory,否则需要在listener中指明连接工厂-->
<bean id="redisConnectionFactory" class="org.springframework.data.redis.connection.jedis.JedisConnectionFactory"
p:host-name="${redis.host}" p:port="${redis.port}" p:password="${redis.pass}"
p:pool-config-ref="poolConfig"/>
<bean id="redisTemplate" class="org.springframework.data.redis.core.StringRedisTemplate">
<property name="connectionFactory" ref="redisConnectionFactory"/>
</bean>
<bean id="stringRedisSerializer" class="org.springframework.data.redis.serializer.StringRedisSerializer"/>
<!--此处注册监听器,需要指定通道名称(topic)(可以使用正则表达式*_等等),第一种为实现MessageListener接口的监听器的注册,第二种为自己定义的类的注册需要制定处理方法名称(不制定的默认方法为handleMessage,如果你的方法是这个名称可以不指定)与序列化的方式,推荐使用第一种方式-->
<redis:listener-container>
<redis:listener ref="redisMessageListener" topic="talk"/>
<redis:listener ref="eventListener" topic="talk*" method="getMessage"
serializer="stringRedisSerializer"></redis:listener>
</redis:listener-container>
</beans>
redis消息队列
redis消息队列使用 redis中的list数据结构实现(左进右出)
/**
* 向指定的列表左边插入数据
*
* @param key
* @param value
* @return
*/
public void leftPush(String key, String value) {
redisTemplate.opsForList().leftPush(key, value);
}
这个代码即可向指定的list中的左边插入值
/**
* 弹出指定列表右边的数据(如果没有数据,在指定的时间内等待)
*
* @param key
* @param timeout
* @param unit
* @return
*/
public String rightPop(String key, long timeout, TimeUnit unit) {
return redisTemplate.opsForList().rightPop(key, timeout, unit);
}
以上代码可以从指定列表的右边弹出一个数据(如果没有,会等待指定时间返回空),只需要在工程中启动一个线程不停的使用这个方法即可实现消息队列的监听
@PostConstruct
public void messageListener() {
new Thread(new Runnable() {
@Override
public void run() {
while (true) {
rightPop(....)
....
}
}
}, "消息监听任务线程").start();
}
对于弹出方法,可以使用
/**
* 弹出指定列表右边,并向指定列表的左边插入(弹出列表如果没有元素,等待指定的时间)
*
* @param sourceKey
* @param destinationKey
* @param timeout
* @param unit
* @return
*/
public String rightPopAndLeftPush(String sourceKey, String destinationKey, long timeout, TimeUnit unit) {
return redisTemplate.opsForList().rightPopAndLeftPush(sourceKey, destinationKey, timeout, unit);
}
这个方法优化代码,实现弹出的同时插入一个处理队列(事务)
