Guava Cache 数据变化实现回调的监听器RemovalListener

上一篇介绍了guava的使用，实现了项目第一个需求定期清理cache数据，第二个需求，我们需要在缓存被移除的时候，得到通知产生回调，并做一些额外处理工作。这个时候RemovalListener就派上用场了。

下面是获得所有数据改变的监听

[java]  view plain  copy

1. public class Main {  
2.   
3.     // 创建一个监听器  
4.     private static class MyRemovalListener implements RemovalListener<Integer, Integer> {  
5.     @Override  
6.     public void onRemoval(RemovalNotification<Integer, Integer> notification) {  
7.         String tips = String.format("key=%s,value=%s,reason=%s", notification.getKey(), notification.getValue(), notification.getCause());  
8.         System.out.println(tips);  
9.     }  
10.     }  
11.   
12.     public static void main(String[] args) {  
13.   
14.     // 创建一个带有RemovalListener监听的缓存  
15.     Cache<Integer, Integer> cache = CacheBuilder.newBuilder().removalListener(new MyRemovalListener()).build();  
16.   
17.     cache.put(1, 1);  
18.   
19.     // 手动清除  
20.     cache.invalidate(1);  
21.   
22.     System.out.println(cache.getIfPresent(1)); // null  
23.     }  
24.   
25. }  

使用invalidate()清除缓存数据之后，注册的回调被触发了。

下面是只有主动删除数据使的回调

public class CacheConnection {

  public static RemovalListener<String, Connection> myRemovalListener = new RemovalListener<String, Connection>(){
    @Override  
    public void onRemoval(RemovalNotification<String, Connection> notification) {  
        String tips = String.format("key=%s,value=%s,reason=%s in myRemovalListener", notification.getKey(), notification.getValue(), notification.getCause());  
        System.out.println(tips);
        //when expireAfterAccess to do
        if (notification.getCause().equals("EXPIRED") && notification.getValue() != null) {
          try {
            notification.getValue().close();
          } catch (SQLException e) {
            System.out.printf("Exception in myRemovalListener:\n");
            e.printStackTrace();
          }

          System.out.printf("Remove %s in cacheConnection", notification.getKey());
        }

    }
  };

  public static Cache<String, Connection> cacheConnection = CacheBuilder.newBuilder()  
      //设置cache的初始大小为20000，要合理设置该值  
      .initialCapacity(20000)  
      //设置并发数为5，即同一时间最多只能有5个线程往cache执行写入操作  
      .concurrencyLevel(100)  
      //设置cache中的数据在600秒没有被读写将自动删除  
      .expireAfterAccess(600, TimeUnit.SECONDS) 
      //设置监听，当出现自动删除时的回调
      .removalListener(myRemovalListener)
      //构建cache实例  
      .build();  

  public static Connection getCache(String key)  {
    try {
      Connection var = cacheConnection.getIfPresent(key);
      return var;
    } catch (Exception e) {
      // TODO: handle exception
      System.out.println("the value of cacheConnection is null");
      e.printStackTrace();
      return null;
    }
  }

   public static void putCache(String key, Connection value) {
     cacheConnection.put(key, value);
   }


}

RemovalNotification中包含了缓存的key、value以及被移除的原因RemovalCause。通过源码可以看出，移除原因与容量管理方式是相对应的。下面是具体的消息

[java]  view plain  copy

1. public enum RemovalCause {  
2.   /** 
3.    * The entry was manually removed by the user. This can result from the user invoking 
4.    * {@link Cache#invalidate}, {@link Cache#invalidateAll(Iterable)}, {@link Cache#invalidateAll()}, 
5.    * {@link Map#remove}, {@link ConcurrentMap#remove}, or {@link Iterator#remove}. 
6.    */  
7.   EXPLICIT {  
8.     @Override  
9.     boolean wasEvicted() {  
10.       return false;  
11.     }  
12.   },  
13.   
14.   /** 
15.    * The entry itself was not actually removed, but its value was replaced by the user. This can 
16.    * result from the user invoking {@link Cache#put}, {@link LoadingCache#refresh}, {@link Map#put}, 
17.    * {@link Map#putAll}, {@link ConcurrentMap#replace(Object, Object)}, or 
18.    * {@link ConcurrentMap#replace(Object, Object, Object)}. 
19.    */  
20.   REPLACED {  
21.     @Override  
22.     boolean wasEvicted() {  
23.       return false;  
24.     }  
25.   },  
26.   
27.   /** 
28.    * The entry was removed automatically because its key or value was garbage-collected. This 
29.    * can occur when using {@link CacheBuilder#weakKeys}, {@link CacheBuilder#weakValues}, or 
30.    * {@link CacheBuilder#softValues}. 
31.    */  
32.   COLLECTED {  
33.     @Override  
34.     boolean wasEvicted() {  
35.       return true;  
36.     }  
37.   },  
38.   
39.   /** 
40.    * The entry's expiration timestamp has passed. This can occur when using 
41.    * {@link CacheBuilder#expireAfterWrite} or {@link CacheBuilder#expireAfterAccess}. 
42.    */  
43.   EXPIRED {  
44.     @Override  
45.     boolean wasEvicted() {  
46.       return true;  
47.     }  
48.   },  
49.   
50.   /** 
51.    * The entry was evicted due to size constraints. This can occur when using 
52.    * {@link CacheBuilder#maximumSize} or {@link CacheBuilder#maximumWeight}. 
53.    */  
54.   SIZE {  
55.     @Override  
56.     boolean wasEvicted() {  
57.       return true;  
58.     }  
59.   };  
60.   
61.   /** 
62.    * Returns {@code true} if there was an automatic removal due to eviction (the cause is neither 
63.    * {@link #EXPLICIT} nor {@link #REPLACED}). 
64.    */  
65.   abstract boolean wasEvicted();  
66. }  

监听器使用很简单，有几个特点需要注意下：

1、默认情况下，监听器方法是被同步调用的（在移除缓存的那个线程中执行）。如果监听器方法比较耗时，会导致调用者线程阻塞时间变长。下面这段代码，由于监听器执行需要2s，所以main线程调用invalidate()要2s后才能返回。

[java]  view plain  copy

1. public class Main {  
2.   
3.     // 创建一个监听器  
4.     private static class MyRemovalListener implements RemovalListener<Integer, Integer> {  
5.         @Override  
6.         public void onRemoval(RemovalNotification<Integer, Integer> notification) {  
7.             String tips = String.format("key=%s,value=%s,reason=%s", notification.getKey(), notification.getValue(), notification.getCause());  
8.             System.out.println(tips);  
9.   
10.             try {  
11.                 // 模拟耗时  
12.                 Thread.sleep(2000);  
13.             } catch (InterruptedException e) {  
14.                 e.printStackTrace();  
15.             }  
16.   
17.         }  
18.     }  
19.   
20.     public static void main(String[] args) {  
21.   
22.         // 创建一个带有RemovalListener监听的缓存  
23.         final Cache<Integer, Integer> cache = CacheBuilder.newBuilder().removalListener(new MyRemovalListener()).build();  
24.         cache.put(1, 1);  
25.         cache.put(2, 2);  
26.   
27.         System.out.println("main...begin.");  
28.         cache.invalidate(1);// 耗时2s  
29.         System.out.println("main...over.");  
30.     }  
31.   
32. }  

解决这个问题的方法是：使用异步监听RemovalListeners.asynchronous(RemovalListener, Executor)。

[java]  view plain  copy

1. public class Main {  
2.   
3.     // 创建一个监听器  
4.     private static class MyRemovalListener implements RemovalListener<Integer, Integer> {  
5.         @Override  
6.         public void onRemoval(RemovalNotification<Integer, Integer> notification) {  
7.             String tips = String.format("key=%s,value=%s,reason=%s", notification.getKey(), notification.getValue(), notification.getCause());  
8.             System.out.println(tips);  
9.   
10.             try {  
11.                 // 模拟耗时  
12.                 Thread.sleep(2000);  
13.             } catch (InterruptedException e) {  
14.                 e.printStackTrace();  
15.             }  
16.   
17.         }  
18.     }  
19.   
20.     public static void main(String[] args) {  
21.   
22.         RemovalListener<Integer, Integer> async = RemovalListeners.asynchronous(new MyRemovalListener(), Executors.newSingleThreadExecutor());  
23.         // 创建一个带有RemovalListener监听的缓存  
24.         final Cache<Integer, Integer> cache = CacheBuilder.newBuilder().removalListener(async).build();  
25.         cache.put(1, 1);  
26.         cache.put(2, 2);  
27.   
28.         System.out.println("main...begin.");  
29.         cache.invalidate(1);// main线程立刻返回  
30.         System.out.println("main...over.");  
31.     }  
32.   
33. }  

2、创建cache的时候只能添加1个监听器，这个监听器对象会被多个线程共享，所以如果监听器需要操作共享资源，那么一定要做好同步控制。下面这段代码可以看出：2个线程会交替执行监听器的发方法。

[java]  view plain  copy

1. public class Main {  
2.   
3.     // 创建一个监听器  
4.     private static class MyRemovalListener implements RemovalListener<Integer, Integer> {  
5.         @Override  
6.         public void onRemoval(RemovalNotification<Integer, Integer> notification) {  
7.             String tips = String.format("key=%s,value=%s,reason=%s", notification.getKey(), notification.getValue(), notification.getCause());  
8.             System.out.println(tips);  
9.   
10.             try {  
11.                 // 模拟耗时  
12.                 Thread.sleep(2000);  
13.             } catch (InterruptedException e) {  
14.                 e.printStackTrace();  
15.             }  
16.   
17.             System.out.println("process over.");  
18.         }  
19.     }  
20.   
21.     public static void main(String[] args) {  
22.   
23.         // 创建一个带有RemovalListener监听的缓存  
24.         final Cache<Integer, Integer> cache = CacheBuilder.newBuilder().removalListener(new MyRemovalListener()).build();  
25.         cache.put(1, 1);  
26.         cache.put(2, 2);  
27.   
28.         new Thread(new Runnable() {  
29.             @Override  
30.             public void run() {  
31.                 System.out.println("thread1...trigger RemovalListener begin.");  
32.                 cache.invalidate(1);  
33.                 System.out.println("thread1...trigger RemovalListener over.");  
34.             }  
35.         }).start();  
36.   
37.         new Thread(new Runnable() {  
38.             @Override  
39.             public void run() {  
40.                 System.out.println("thread2...trigger RemovalListener begin.");  
41.                 cache.invalidate(2);  
42.                 System.out.println("thread2...trigger RemovalListener over.");  
43.             }  
44.         }).start();  
45.     }  
46.   
47. }  

3、监听器中抛出的任何异常，在被记录到日志后，会被guava丢弃，不会导致监听器不可用。下面这段代码可以看到：监听器中抛出的异常只是被记录了(打印到了控制台)，并没有导致JVM退出，之后缓存被移除一样可以再次触发。

[java]  view plain  copy

1. public class Main {  
2.   
3.     // 创建一个监听器  
4.     private static class MyRemovalListener implements RemovalListener<Integer, Integer> {  
5.         @Override  
6.         public void onRemoval(RemovalNotification<Integer, Integer> notification) {  
7.             String tips = String.format("key=%s,value=%s,reason=%s", notification.getKey(), notification.getValue(), notification.getCause());  
8.             System.out.println(tips);  
9.   
10.             throw new RuntimeException();  
11.         }  
12.     }  
13.   
14.     public static void main(String[] args) {  
15.   
16.         // 创建一个带有RemovalListener监听的缓存  
17.         final Cache<Integer, Integer> cache = CacheBuilder.newBuilder().removalListener(new MyRemovalListener()).build();  
18.         cache.put(1, 1);  
19.         cache.put(2, 2);  
20.   
21.         cache.invalidate(1);  
22.         cache.invalidate(2);  
23.     }  
24.   
25. }