我们一起学ignite之CacheInterceptor

一起学 ignite 拦截器 源码分析

如题，ignite的cache也有拦截功能，其接口定义为CacheInterceptor，设置缓存的拦截处理器后我们就可以对缓存数据操作前后加上一些特定的逻辑，这个功能就是缓存拦截器。

ignite的缓存默认是没有添加任何拦截功能，不在操作缓存的前后做任何处理逻辑，当我们需要添加一些功能的时候，可以定义自己的拦截器，ignite默认为我们提供了一个拦截适配器，CacheInterceptorAdapter，代码如下：

public class CacheInterceptorAdapter<K, V> implements CacheInterceptor<K, V> {
    private static final long serialVersionUID = 0L;
    @Nullable @Override public V onGet(K key, V val) {
        return val;
    }
    @Nullable @Override public V onBeforePut(Cache.Entry<K, V> entry, V newVal) {
        return newVal;
    }
    @Override public void onAfterPut(Cache.Entry<K, V> entry) {
        // No-op.
    }
    @Nullable @Override public IgniteBiTuple<Boolean, V> onBeforeRemove(Cache.Entry<K, V> entry) {
        return new IgniteBiTuple<>(false, entry.getValue());
    }
    @Override public void onAfterRemove(Cache.Entry<K, V> entry) {
        // No-op.
            }
}

可以看到默认的拦截适配器没有做任何处理。
我们可以自己定义个一拦截器，可以在里面添加下对缓存操作的日志。如下

package com.github.aloxc.ignite.benchmark;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.ignite.Ignite;
import org.apache.ignite.IgniteCache;
import org.apache.ignite.Ignition;
import org.apache.ignite.cache.CacheInterceptor;
import org.apache.ignite.cache.CacheMemoryMode;
import org.apache.ignite.cache.CacheMode;
import org.apache.ignite.cache.eviction.lru.LruEvictionPolicy;
import org.apache.ignite.configuration.CacheConfiguration;
import org.apache.ignite.configuration.DeploymentMode;
import org.apache.ignite.configuration.IgniteConfiguration;
import org.apache.ignite.lang.IgniteBiTuple;
import org.jetbrains.annotations.Nullable;
import javax.cache.Cache;
import java.sql.Timestamp;
import java.util.Random;
/**
 * Created by Administrator on 2016/12/27.
 */
public class BenchmarkUserInterceptor implements CacheInterceptor<Integer,BenchmarkUser> {
    private static Log logger = LogFactory.getLog(BenchmarkUserInterceptor.class);
    @Nullable
    @Override
    public BenchmarkUser onGet(Integer key, @Nullable BenchmarkUser value) {
        logger.error(Thread.currentThread().getStackTrace()[1] + "被调用!key = " + key + ",value = " + value);
        return value;
    }
    @Nullable
    @Override
    public BenchmarkUser onBeforePut(Cache.Entry<Integer, BenchmarkUser> entry, BenchmarkUser newVal) {
        logger.error(Thread.currentThread().getStackTrace()[1] + "被调用!entry = " + entry + ",newVal = " + newVal);
        return newVal;
    }
    @Override
    public void onAfterPut(Cache.Entry<Integer, BenchmarkUser> entry) {
        logger.error(Thread.currentThread().getStackTrace()[1] + "被调用!entry = " + entry);
    }
    @Nullable
    @Override
    public IgniteBiTuple<Boolean, BenchmarkUser> onBeforeRemove(Cache.Entry<Integer, BenchmarkUser> entry) {
        logger.error(Thread.currentThread().getStackTrace()[1] + "被调用! entry = " + entry);
        return new IgniteBiTuple<>(false, entry.getValue());
    }
    @Override
    public void onAfterRemove(Cache.Entry<Integer, BenchmarkUser> entry) {
        logger.error(Thread.currentThread().getStackTrace()[1] + "被调用! entry = " + entry);
    }
    private static final String CACHE_NAME = "benchmark4";
    private static Ignite ignite;
    private static IgniteCache<Integer, BenchmarkUser> cache;
    public static void main(String[] args){
        logger.error("开始插入测试数据");
        try {
                IgniteConfiguration cfg = new IgniteConfiguration();
                cfg.setClientMode(false);
                CacheConfiguration cacheConfiguration = new CacheConfiguration<String, BenchmarkUser>();
                cacheConfiguration.setCacheMode(CacheMode.PARTITIONED);//分区
                cacheConfiguration.setBackups(0);//关闭备份
                cacheConfiguration.setMemoryMode(CacheMemoryMode.OFFHEAP_TIERED);
                cacheConfiguration.setOffHeapMaxMemory(524 * 1000 * 1000);//限制堆外内存大小4G
                cacheConfiguration.setEvictionPolicy(new LruEvictionPolicy(100001));
                cacheConfiguration.setStartSize(3 * 1000 * 1000);//初始化缓存容量大小，类似于hashmap的size
                cacheConfiguration.setSwapEnabled(false);//禁用交换存储
                cacheConfiguration.setInterceptor(new BenchmarkUserInterceptor());
                cfg.setCacheConfiguration(cacheConfiguration);
                cfg.setDeploymentMode(DeploymentMode.SHARED);
                cfg.setPeerClassLoadingEnabled(true);
                ignite = Ignition.start(cfg);
                cache = ignite.getOrCreateCache(CACHE_NAME);
            logger.error("资源初始化完毕");
            Random r = new Random();
            for (int id = 0; id < 3; id++) {
                BenchmarkUser user = new BenchmarkUser();
                user.setId(id);
                user.setName("user-" + id);
                user.setCommentCount(id);
                user.setComposeCount(id);
                user.setReplyCount(id);
                user.setIp("ip" + id);
                user.setLoginCount(id);
                user.setPermission(r.nextBoolean());
                user.setLoginTime(new Timestamp(System.currentTimeMillis()));
                user.setRegisterTime(new Timestamp(System.currentTimeMillis()));
                cache.put(id,user);
                cache.remove(id);
            }
        } catch (Throwable e) {
            logger.error("发生异常",e);
        } finally {
            logger.error("程序运行完毕");
//            ignite.destroyCache(CACHE_NAME);
            ignite.close();
            System.exit(1);
        }
    }
}

允许程序后通过日志中可以看到一些内容，

onBeforePut(BenchmarkUserInterceptor.java:39)被调用!entry = CacheLazyEntry [key=null, val=null, keepBinary=false, updateCntr=null],newVal = {"id":0,"name":"user-0","ip":"ip0","loginCount":0,"composeCount":0,"replyCount":0,"commentCount":0,"permission":true,"loginTime":1482823782805,"registerTime":1482823782805}
onAfterPut(BenchmarkUserInterceptor.java:45)被调用!entry = CacheLazyEntry [key=null, val={"id":0,"name":"user-0","ip":"ip0","loginCount":0,"composeCount":0,"replyCount":0,"commentCount":0,"permission":true,"loginTime":1482823782805,"registerTime":1482823782805}, keepBinary=false, updateCntr=1]
onBeforeRemove(BenchmarkUserInterceptor.java:51)被调用! entry = CacheLazyEntry [key=null, val=null, keepBinary=false, updateCntr=null]
onAfterRemove(BenchmarkUserInterceptor.java:57)被调用! entry = CacheLazyEntry [key=null, val=null, keepBinary=false, updateCntr=2]
onBeforePut(BenchmarkUserInterceptor.java:39)被调用!entry = CacheLazyEntry [key=null, val=null, keepBinary=false, updateCntr=null],newVal = {"id":1,"name":"user-1","ip":"ip1","loginCount":1,"composeCount":1,"replyCount":1,"commentCount":1,"permission":true,"loginTime":1482823783157,"registerTime":1482823783157}
onAfterPut(BenchmarkUserInterceptor.java:45)被调用!entry = CacheLazyEntry [key=null, val={"id":1,"name":"user-1","ip":"ip1","loginCount":1,"composeCount":1,"replyCount":1,"commentCount":1,"permission":true,"loginTime":1482823783157,"registerTime":1482823783157}, keepBinary=false, updateCntr=1]
onBeforeRemove(BenchmarkUserInterceptor.java:51)被调用! entry = CacheLazyEntry [key=null, val=null, keepBinary=false, updateCntr=null]
onAfterRemove(BenchmarkUserInterceptor.java:57)被调用! entry = CacheLazyEntry [key=null, val=null, keepBinary=false, updateCntr=2]
onBeforePut(BenchmarkUserInterceptor.java:39)被调用!entry = CacheLazyEntry [key=null, val=null, keepBinary=false, updateCntr=null],newVal = {"id":2,"name":"user-2","ip":"ip2","loginCount":2,"composeCount":2,"replyCount":2,"commentCount":2,"permission":false,"loginTime":1482823783161,"registerTime":1482823783161}
onAfterPut(BenchmarkUserInterceptor.java:45)被调用!entry = CacheLazyEntry [key=null, val={"id":2,"name":"user-2","ip":"ip2","loginCount":2,"composeCount":2,"replyCount":2,"commentCount":2,"permission":false,"loginTime":1482823783161,"registerTime":1482823783161}, keepBinary=false, updateCntr=1]
onBeforeRemove(BenchmarkUserInterceptor.java:51)被调用! entry = CacheLazyEntry [key=null, val=null, keepBinary=false, updateCntr=null]
onAfterRemove(BenchmarkUserInterceptor.java:57)被调用! entry = CacheLazyEntry [key=null, val=null, keepBinary=false, updateCntr=2]

从日志中看到，已经执行我们的拦截逻辑（打印操作日志）了，并且从日中还看到了CacheLazyEntry，通过查看源码我们可以得到一些方法，
getPartitionUpdateCounter()：key的更新次数，
getValue()：类型为我们cache中保存的数据类型，其值为当前操作的value，