一、背景

redis是缓存界的扛把子，而redisson又是java 链接redis 的一哥，每个项目基本都会用到redisson。
它的功能丰富让人羡慕，但是功能丰富极有可能导致使用姿势不正常，就爆发线上问题，这里说两个踩得雷以及解决方法。

二、问题、分析和方案

1、数据结合在高并发情况下的情况：RMap 和RSet 数据不一致问题

操作Rmap： 高并发时候，如果两个线程同时获取Rmap的数据，那原始数据进行操作后，最终结果是延迟放回去数据那个线程对应的结果。
原因：每个线程获取到的都是数据最原始的状态，并没有版本控制一说。

解决方案：
使用时候，根据场景评判是否需要加redisson的锁

2、 反序列化问题

操作对象：如果直接给Redis 中存放对象时候，一定要注意，在类有变化时候，如果使用FstCodec进行对象序列化，反序列化回来的对象会报错，
原因：redisson FstCodec 需要判定类模版

解决方案： 
使用jackson序列化， 或者存入对象时候，先转成json格式进行存储

3、 阻塞队列内存泄露OOM问题

操作对象：在 3.12.5 以下版本，使用 Redisson 的阻塞队列进行take操作，假设如果一直获取不到数据，会存在内存泄漏问题；
原因：使用阻塞队列时，由于连接长时间阻塞无数据，会导致当前连接的会一直进行ping探活，会一直新建CommandData，队列中数据CommandData无限增长最终出现内存泄漏

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
 //因为一直take，就会触发队列拿数据
public class PingConnectionHandler extends ChannelInboundHandlerAdapter {

  @Override
 public void channelActive(final ChannelHandlerContext ctx) throws Exception {
     RedisConnection connection = RedisConnection.getFrom(ctx.channel());
     connection.getConnectionPromise().onComplete((res, e) -> {
         if (e == null) {
             sendPing(ctx);
         }
     });
     ctx.fireChannelActive();
 }

 //启动定时，不停的执行sendping 方法，
 protected void sendPing(final ChannelHandlerContext ctx) {
     final RedisConnection connection = RedisConnection.getFrom(ctx.channel());
     final RFuture<String> future = connection.async(StringCodec.INSTANCE, RedisCommands.PING);
     
     config.getTimer().newTimeout(new TimerTask() {
         @Override
         public void run(Timeout timeout) throws Exception {
             CommandData<?, ?> commandData = connection.getCurrentCommand();
             if ((commandData == null || !commandData.isBlockingCommand()) 
                     && (future.cancel(false) || !future.isSuccess())) {
                 ctx.channel().close();
                 log.debug("channel: {} closed due to PING response timeout set in {} ms", ctx.channel(), config.getPingConnectionInterval());
             } else {
                 sendPing(ctx);
             }
         }
     }, config.getPingConnectionInterval(), TimeUnit.MILLISECONDS);
 }
}

// 因为上面会不停的执行，所以到了这块，就会不停的给queue中新增对象

 public class ChannelDuplexHandler extends ChannelInboundHandlerAdapter implements ChannelOutboundHandler 
 //写出时候，如果没有数据，则会在queue中新增一个QueueCommandHolder对象
 public class CommandsQueue extends ChannelDuplexHandler {
      @Override
     public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {
         if (msg instanceof QueueCommand) {
             QueueCommand data = (QueueCommand) msg;
             QueueCommandHolder holder = queue.peek();
             if (holder != null && holder.getCommand() == data) {
                 super.write(ctx, msg, promise);
             } else {
                 queue.add(new QueueCommandHolder(data, promise));
                 sendData(ctx.channel());
             }
         } else {
             super.write(ctx, msg, promise);
         }
     }
 }

3、 RMap存储对象多导致OOM问题

操作CacheMap ：如果使用cacheMap，如果放入太多的key，会导致线上OOM。
原因：java服务会启动过多的Task来定时执行检测Key的过期，
如下代码中的：EvictionTask 及根据不同的key 来起一个定时，如果key 多了，对象也会很多。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
public <K, V> RMapCache<K, V> getMapCache(String name, Codec codec) {
    return new RedissonMapCache<K, V>(codec, evictionScheduler, commandExecutor, name, this, null, null);
}

public RedissonMapCache(EvictionScheduler evictionScheduler, CommandAsyncExecutor commandExecutor,
                        String name, RedissonClient redisson, MapCacheOptions<K, V> options, WriteBehindService writeBehindService) {
    super(commandExecutor, name, redisson, options, writeBehindService);
    if (evictionScheduler != null) {
        evictionScheduler.schedule(getRawName(), getTimeoutSetName(), getIdleSetName(), getExpiredChannelName(), getLastAccessTimeSetName(), options);
    }
    this.evictionScheduler = evictionScheduler;
    this.publishCommand = commandExecutor.getConnectionManager().getSubscribeService().getPublishCommand();
} 

 public void schedule(String name, String timeoutSetName, String maxIdleSetName,
                     String expiredChannelName, String lastAccessTimeSetName, MapCacheOptions<?, ?> options) {
    boolean removeEmpty = false;
    if (options != null) {
        removeEmpty = options.isRemoveEmptyEvictionTask();
    }

    EvictionTask task = new MapCacheEvictionTask(name, timeoutSetName, maxIdleSetName, expiredChannelName, lastAccessTimeSetName,
            executor, removeEmpty, this);
    EvictionTask prevTask = tasks.putIfAbsent(name, task);
    if (prevTask == null) {
        task.schedule();
    }
}

解决方案：

使用redis 的原始map，同时使用expire 来让原始map过期，至于map中的详细数据，则自维护一个 过期策略，如下给出示例

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
   public class OwnLFU <K, V> extends LinkedHashMap<K, V> {
	private final int capacity;

	public OwnLFU(int capacity) {
		super(capacity, 0.75f, true);
		this.capacity = capacity;
	}

	@Override
	protected boolean removeEldestEntry(Map.Entry<K, V> eldest) {
		//这里也可以加上自己的逻辑,比如，如果往该LFU里面存入的是 时间戳之类的，就对key进行比对，返回是否删除
//		if((Long)eldest.getKey() > System.currentTimeMillis()){
//			return true;
//		}
		return  size() > capacity;
	}

   }