Spring Data Redis 2 之消息订阅
Spring Data Redis是spring提供了操作Redis键值存储集成方案。简单回顾下 内容
[*] 抽象出RedisConnection ，RedisConnectionFactory概念，集成了4个redis客户端
[*] 提供的RedisTemplate，是一个高层次操作视图
[*] 主要提供操作视图，主要包括两大类目：*Operations，Bound*Operations。当然Bound*Operations是对*Operations的简单封装。
[*] 提供了Serializers序列工具，主要应用与key,value,hash方面。
本节主要内容，使用SDR工具，完成消息订阅与分发，事务管理，消息管道化。
1.消息订阅与分发
1.1 主要相关命令
http://img2.tuicool.com/uqENje.jpg%21web
命令参照：http://redis.readthedocs.io/en/2.4/pub_sub.html
1.2 Redis消息监听容器声明和消息监听器注册
为了订阅消息，需要实现MessageListener回调，每次新消息到达时，回调被调用，用户代码通过onMessage方法执行。
DefaultMessageListener.java 一个简单的MessageListener实现
public class DefaultMessageListenerimplements MessageListener {
protected final Logger LOGGER = LoggerFactory.getLogger(DefaultMessageListener.class);
@Override
public void onMessage(Message message, byte[] pattern) {
byte[] channel = message.getChannel();
byte[] body = message.getBody();
LOGGER.info(new String(channel) + "-->" + new String(body) + "-->" + new String(pattern));
}
}
考虑到“等待消息“过程是阻塞的，SDR提供了RedisMessageListenerContainer。
RedisMessageListenerContainer充当消息侦听器容器;它用于从Redis通道接收消息，并驱动注入它的MessageListener，RedisMessageListenerContainer负责消息接收和分派到侦听器中的所有线程的处理。
lettuce-context.xml
当然为了研究方便，当然使用下面的声明方式完全可以代替上面部分，完全等价。
至此，就完成了消息的订阅了。
1.3 MessageListenerAdapter
MessageListenerAdapter类是Spring的异步消息传递支持的最后一个组件：简而言之，它允许您将几乎任何类暴露为MDP。比如上面的DefaultMessageListener,毕竟还是实现了MessageListener接口，并没有完全实现了MDP。可以借助MessageListenerAdapter轻松实现。
简单展示下实现过程
1.3.1 一个完全符合MDP的接口
public interface MessageDelegate {
//void handleMessage(String message);
void handleMessage(Map message);
void handleMessage(byte[] message);
void handleMessage(Serializable message);
// pass the channel/pattern as well
void handleMessage(Serializable message, String channel);
}
1.3.2 MDP接口实现类
public class DefaultMessageDelegate implements MessageDelegate {
protected final Logger LOGGER = LoggerFactory.getLogger(DefaultMessageDelegate.class);
public void handleMessage(String message) {
LOGGER.info("--------handleMessage 1-----------");
LOGGER.info(message);
}
@Override
public void handleMessage(Map message) {
LOGGER.info("--------handleMessage 2-----------");
}
@Override
public void handleMessage(byte[] message) {
LOGGER.info("--------handleMessage 3-----------");
LOGGER.info(new String(message));
}
public void handleMessage(Serializable message) {
LOGGER.info("--------handleMessage 4-----------");
LOGGER.info(message.toString());
}
public void handleMessage(Serializable message, String channel) {
LOGGER.info("--------handleMessage 5-----------");
LOGGER.info(message + "------>" + channel);
}
// implementation elided for clarity...
}
1.3.3配置
1.4 总结
主要涉及4个组件
MessageListener:Redis中发布的消息的侦听器。
MessageListenerAdapter:消息侦听器适配器，通过反射将消息处理委托给目标侦听器方法，并进行灵活的消息类型转换
ChannelTopic:Channel topic
RedisMessageListenerContainer:Redis消息侦听器提供异步行为的容器。 处理侦听，转换和消息分派的低级细节
2. 事务管理
redis事务命令，具体见http://redis.readthedocs.io/en/2.4/transaction.html。
首先看一个测试用例
2.1 一个不能运行的事务代码
@Test
public void testTran(){
this.redisTemplate.multi();
try {
for (int i = 0; i < 1; i++) {
String key = "key" + i;
String value = "00000";
this.redisTemplate.opsForValue().set(key, value);
}
this.redisTemplate.exec();
}catch (Exception e){
this.redisTemplate.discard();
}
}
执行代码会报异常。具体异常
Caused by: com.lambdaworks.redis.RedisCommandExecutionException: ERR DISCARD without MULTI
at com.lambdaworks.redis.LettuceFutures.await(LettuceFutures.java:76)
at com.lambdaworks.redis.FutureSyncInvocationHandler.handleInvocation(FutureSyncInvocationHandler.java:59)
at com.google.common.reflect.AbstractInvocationHandler.invoke(AbstractInvocationHandler.java:87)
at com.sun.proxy.$Proxy15.discard(Unknown Source)
at org.springframework.data.redis.connection.lettuce.LettuceConnection.discard(LettuceConnection.java:833)
这是因为Redis通过multi，exec和discard命令为事务提供支持。 这些操作在RedisTemplate上可用，但是RedisTemplate不能保证在事务中使用相同的连接执行所有操作。
处理异常
通过设置setEnableTransactionSupport（true）显式为每个正在使用的RedisTemplate启用。 这将强制将正在使用的RedisConnection绑定到触发MULTI的当前线程。
修改后的代码如下
public void testTran(){
StringRedisSerializer serializer = new StringRedisSerializer();
redisTemplate.setKeySerializer(serializer);
redisTemplate.setValueSerializer(serializer);
this.redisTemplate.setEnableTransactionSupport(true);
this.redisTemplate.multi();
try {
for (int i = 0; i < 100; i++) {
String key = "key" + i;
String value = "00000";
this.redisTemplate.opsForValue().set(key, value);
}
this.redisTemplate.exec();
}catch (Exception e){
this.redisTemplate.discard();
}
}
运行结果
1) "key95"
2) "key82"
3) "key69"
4) "key90"
5) "key18"
6) "key1"
7) "key4"
8) "key63"
9) "key78"
执行结果，也简介认证了redisTemplate 执行结果无序。
2.2 正常事务代码
@Test
public void testTran3(){
StringRedisSerializer serializer = new StringRedisSerializer();
redisTemplate.setKeySerializer(serializer);
redisTemplate.setValueSerializer(serializer);
redisTemplate.execute(new SessionCallback() {
@Override
publicList execute(RedisOperations operations) throws DataAccessException {
operations.multi();
try {
for(int i=0;i5){
thrownew Exception();
}
}
return operations.exec();
} catch (Exception e) {
operations.discard();
}
// This will contain the results of all ops in the transaction
return Collections.emptyList();
}
});
}
}
3. 管道支持
Redis提供对流水线的支持，这涉及向服务器发送多个命令，而不必等待答复，然后在一个步骤中读取答复。当您需要在一行中发送多个命令时，流水线可以提高性能，例如向同一列表中添加许多元素。
/**
*在管道连接上执行给定的操作对象，返回结果。
*注意，回调不能返回非空值，因为它被管道覆盖。 此方法将使用默认序列化器反序列化结果
*
* @param action callback object to execute
* @return list of objects returned by the pipeline
*/
List executePipelined(RedisCallback action);
/**
*在流水线连接上执行给定的操作对象，使用专用的序列化程序返回结果
*
* @param action callback object to execute
* @param resultSerializer The Serializer to use for individual values or Collections of values. If any returned
* values are hashes, this serializer will be used to deserialize both the key and value
* @return list of objects returned by the pipeline
*/
List executePipelined(final RedisCallback action, final RedisSerializer resultSerializer);
/**
*在流水线连接上执行给定的Redis会话。 允许事务流水线化。
* 注意，回调不能返回非空值，因为它被管道覆盖。
* @param session Session callback
* @return list of objects returned by the pipeline
*/
List executePipelined(final SessionCallback session);
/**
* 在管道连接上执行给定的Redis会话，使用专用的序列化程序返回结果。 允许事务流水线化。
* 注意，回调不能返回非空值，因为它被管道覆盖。
*
* @param session Session callback
* @param resultSerializer
* @return list of objects returned by the pipeline
*/
List executePipelined(final SessionCallback session, final RedisSerializer resultSerializer);
页:
[1]