Spring Boot整合Memcached

前面已经提到，缓存是快速提升系统性能，缓解瓶颈的有效手段。

缓存的种类多种多样，小到CPU的缓存，大到静态生成的页面缓存。在本小节中，我们主要讨论在Spring Boot中整合如下两种缓存:

  • 本地缓存: 在内存中开辟一小块空间，用于缓存，速度很快，但容量受限。我们采用Gruva中的缓存实现。
  • 网络缓存: 同一微服务的不同节点间，通过网络共享，例如Memcached。

通用缓存接口

既然要在微服务中支持2种缓存，不妨设计一个较为通用的接口:

  1. public interface ICache<K, V> {
  3.     @Nullable
  4.     V get(K key);
  6.     @Nullable
  7.     default V cacheGet(K key, Function<K, V> func, int ttlSecs) {
  8.         V val = get(key);
  9.         if (val != null) {
  10.             return val;
  11.         } else {
  12.             val = func.apply(key);
  13.             put(key, val, ttlSecs);
  14.             return val;
  15.         }
  16.     }
  18.     default V cacheGet(K key, Function<K, V> func) {
  19.         return cacheGet(key, func, 0);
  20.     }
  22.     Map<K, V> batchGet(Collection<K> keys);
  24.     default Map<K, V> batchCacheGet(Collection<K> keys, Function<Collection<K>, Map<K, V>> func, int ttlSecs) {
  25.         // hit map
  26.         Map<K, V> hitMap = batchGet(keys);
  28.         // miss keys
  29.         Collection<K> missedKeys = null;
  30.         if (hitMap == null || hitMap.isEmpty()) {
  31.             missedKeys = keys;
  32.         } else {
  33.             missedKeys = keys.stream().filter(k -> !hitMap.containsKey(k)).collect(Collectors.toSet());
  34.         }
  36.         // check if no miss keys
  37.         if (missedKeys == null || missedKeys.isEmpty()) {
  38.             return hitMap;
  39.         }
  41.         // fetch miss key
  42.         Map<K, V> missMap = func.apply(missedKeys);
  43.         missMap.entrySet().forEach(e -> put(e.getKey(), e.getValue(), ttlSecs));
  45.         if (missMap == null || missMap.isEmpty()) {
  46.             // no miss map again
  47.             return hitMap;
  48.         } else {
  49.             // union & return
  50.             hitMap.putAll(missMap);
  51.             return hitMap;
  52.         }
  53.     }
  55.     default Map<K, V> batchCacheGet(Collection<K> keys, Function<Collection<K>, Map<K, V>> func) {
  56.         return batchCacheGet(keys, func, 0);
  57.     }
  59.     void put(K key, V value);
  61.     void put(K key, V value, int ttlSecs);
  63.     void del(K key);
  65.     default void batchDel(Collection<K> keys) {
  66.         if (keys != null) {
  67.             keys.stream().forEach(key -> del(key));
  68.         }
  69.     }
  71.     void clear();
  73. }

如上所示，我们定义了3大类Cache的基本操作:

  • get/batchGet: 从缓存中获取数据，支持单个或批量操作。
  • put/del: 向缓存中写入或删除，支持设置超时时间。
  • cacheGet: 若缓存中存在则直接返回，否则通过一个Function来生成结果，并写入缓存。类似的，支持单个和批量操作、支持设置超时时间。

LocalCache的实现

在设计Memcached缓存之前，先来看一下本地缓存实现：

  1. public class LocalCache<K, V> implements ICache<K, V> {
  3.     private Logger LOG = LoggerFactory.getLogger(getClass());
  5.     private Cache<K, V> gCache;
  7.     private long ttlSecs = 0;
  9.     public LocalCache(long capacity, long ttlSecs) {
  11.         CacheBuilder builder = CacheBuilder.newBuilder();
  12.         if (capacity > 0) {
  13.             builder.maximumSize(capacity);
  14.         }
  15.         if (ttlSecs > 0) {
  16.             this.ttlSecs = ttlSecs;
  17.             builder.expireAfterWrite(ttlSecs, TimeUnit.SECONDS);
  18.         }
  20.         this.gCache = builder.build();
  21.     }
  23.     @Nullable
  24.     @Override
  25.     public V get(K key) {
  26.         return gCache.getIfPresent(key);
  27.     }
  29.     @Override
  30.     public Map<K, V> batchGet(Collection<K> keys) {
  31.         if (keys == null || keys.isEmpty()) {
  32.             return new HashMap<>();
  33.         } else {
  34.             Map<K, V> result = new HashMap<>();
  35.             for (K key : keys) {
  36.                 V val = gCache.getIfPresent(key);
  37.                 if (val != null) {
  38.                     result.put(key, val);
  39.                 }
  40.             }
  41.             return result;
  42.         }
  43.     }
  45.     @Override
  46.     public void put(K key, V value) {
  47.         gCache.put(key, value);
  48.     }
  50.     @Override
  51.     public void put(K key, V value, int curTtlSecs) {
  52.         if (curTtlSecs != this.ttlSecs) {
  53.             LOG.error("not support per-put ttlSecs currently");
  54.         }
  55.         put(key, value);
  56.     }
  58.     @Override
  59.     public void del(K key) {
  60.         gCache.invalidate(key);
  61.     }
  63.     @Override
  64.     public void batchDel(Collection<K> keys) {
  65.         gCache.invalidateAll(keys);
  66.     }
  68.     @Override
  69.     public void clear() {
  70.         gCache.invalidateAll();
  71.     }
  72. }

如上所示，我们调用了Grava的缓存，来实现了本地缓存。

要说明的是，由于Grava的设计限制，目前TTL需要在创建缓存之初就设定好，并不支持per-key的ttl设定。

MemcachedCache的实现

在Spring Boot中集成Memcached，首先要选择一款基于Java的客户端，比较成熟的开源项目有:

  • spymemcached
  • XMemcached在本小节中，我们选择社区更为活跃的XMemcached，它支持线程池、一致性哈系等较为重要的特性。

首先来看一下客户端的构造：

  1. public class MemcachedClientBuilder2 {
  3.     public static MemcachedClient build(String serverList, int connPoolSize) throws IOException {
  4.         MemcachedClientBuilder builder = new XMemcachedClientBuilder(
  5.                 AddrUtil.getAddresses(serverList));
  6.         // conn pool
  7.         builder.setConnectionPoolSize(connPoolSize);
  8.         // consistent hash
  9.         builder.setSessionLocator(new KetamaMemcachedSessionLocator());
  10.         return builder.build();
  11.     }
  13. }

如上所示，Builder主要设定了两个参数:

  • serverList: 服务器列表（形如ip:port，若有多个可通过空格分割开）
  • connPoolSize: 线程池大小

如果你仔细观察ICache接口，可以发现它是泛型的ICache，K和V可以是任意类型。

然而，Memcached的设计较为轻量，Key必须是字符串，而Value则是byte数组。

所以，需要设计一种通用的方式，以方便泛型数据类型到Memcached的Key/Value转换。

我们将这一转换逻辑抽提成Key/Value的Transformer, Key的:

  1. public interface CacheKeyTransformer<T> {
  3.     String getKey(T t);
  5. }

和Value的

  1. public interface CacheValueTransformer<T> {
  3.     byte[] serialize(T obj);
  5.     T deserialize(byte[] bytes);
  7. }

这两个接口看起来很抽象，我们首先来看一下DefaultCacheKeyTransformer，实现了任何类型到String(Memcached Key类型)的转换:

  1. public class DefaultCacheKeyTransformer<T> implements CacheKeyTransformer<T> {
  3.     private String cacheType;
  5.     public DefaultCacheKeyTransformer(String cacheType) {
  6.         this.cacheType = cacheType;
  7.     }
  9.     @Override
  10.     public String getKey(T t) {
  11.         return cacheType + "#" + t.toString();
  12.     }
  14. }

一种很常见的场景，是将对象序列化为Json然后放到Memcached的Value中，JsonCacheValueTransformer完成了这一过程:

  1. public class JsonCacheValueTransformer<T> implements CacheValueTransformer<T> {
  3.     protected final Logger LOG = LoggerFactory.getLogger(getClass());
  5.     private ObjectMapper objectMapper;
  7.     private Class<T> cls;
  9.     public JsonCacheValueTransformer(Class<T> cls) {
  10.         this.objectMapper = new ObjectMapper();
  11.         this.cls = cls;
  12.     }
  14.     @Override
  15.     public byte[] serialize(T o) {
  16.         byte[] defReturn = new byte[1];
  17.         try {
  18.             if (o == null) {
  19.                 return defReturn;
  20.             }
  21.             return objectMapper.writeValueAsBytes(o);
  22.         } catch (Exception e) {
  23.             LOG.error("JsonCacheValueTransformer serialize exception", e);
  24.             return defReturn;
  25.         }
  26.     }
  28.     @Override
  29.     public T deserialize(byte[] bytes) {
  30.         try {
  31.             if (bytes == null) {
  32.                 return null;
  33.             }
  34.             return objectMapper.readValue(bytes, cls);
  35.         } catch (Exception e) {
  36.             LOG.error("JsonCacheValueTransformer deserialize exception", e);
  37.             return null;
  38.         }
  39.     }
  40. }

如上所示，我们应用了Jackson来实现了Json的序列化(反序列化)，并适配了byte数组到字符串的转换。

此外，Cache的Value中直接存储Integer/Long/String也较为常见，感兴趣的可以直接查看lmsia-cache项目的源代码，这里不再赘述。

实现了Key/Value的序列化之后，我们看一下具体的MemcachedCache实现:

  1. public abstract class AbstractMemcachedCache<K, V> implements ICache<K, V> {
  3.     protected final Logger LOG = LoggerFactory.getLogger(getClass());
  5.     private static final int connPoolSize = 16;
  7.     protected abstract MemcachedClient getMemcachedClient();
  9.     protected abstract CacheKeyTransformer<K> getKeyTransformer();
  11.     protected abstract CacheValueTransformer<V> getValueTransformer();
  13.     private Transcoder<byte[]> transcoder = new Transcoder<byte[]>() {
  15.         @Override
  16.         public void setPrimitiveAsString(boolean primitiveAsString) {
  17.         }
  19.         @Override
  20.         public void setPackZeros(boolean packZeros) {
  21.         }
  23.         @Override
  24.         public void setCompressionThreshold(int to) {
  25.         }
  27.         @Override
  28.         public void setCompressionMode(CompressionMode compressMode) {
  29.         }
  31.         @Override
  32.         public boolean isPrimitiveAsString() {
  33.             return false;
  34.         }
  36.         @Override
  37.         public boolean isPackZeros() {
  38.             return false;
  39.         }
  41.         @Override
  42.         public CachedData encode(byte[] o) {
  43.             return new CachedData(0, o);
  44.         }
  46.         @Override
  47.         public byte[] decode(CachedData d) {
  48.             if (d != null) {
  49.                 return d.getData();
  50.             } else {
  51.                 return null;
  52.             }
  53.         }
  55.     };
  57.     public void init() throws Exception {
  59.         // check
  60.         if (getKeyTransformer() == null) {
  61.             throw new RuntimeException("keyTransformer can not be null");
  62.         }
  64.         if (getValueTransformer() == null) {
  65.             throw new RuntimeException("valueTransformer can not be null");
  66.         }
  68.     }
  71.     @Nullable
  72.     @Override
  73.     public V get(K key) {
  74.         try {
  75.             byte[] bytes = getMemcachedClient().get(getKeyTransformer().getKey(key), transcoder);
  76.             if (bytes == null) {
  77.                 return null;
  78.             }
  79.             return getValueTransformer().deserialize(bytes);
  80.         } catch (Exception e) {
  81.             LOG.error("memcached get exception", e);
  82.             return null;
  83.         }
  84.     }
  86.     @Override
  87.     public Map<K, V> batchGet(Collection<K> keys) {
  88.         if (keys == null || keys.isEmpty()) {
  89.             return new HashMap<>();
  90.         }
  92.         Map<K, String> key2idMap = new HashMap<>();
  93.         for (K key : keys) {
  94.             key2idMap.put(key, getKeyTransformer().getKey(key));
  95.         }
  97.         Collection<String> ids = key2idMap.values();
  99.         try {
  100.             Map<String, byte[]> map = getMemcachedClient().get(ids, transcoder);
  101.             if (map == null || map.isEmpty()) {
  102.                 return new HashMap<>();
  103.             }
  105.             Map<K, V> result = new HashMap<>();
  106.             for (Entry<K, String> entry : key2idMap.entrySet()) {
  107.                 K key = entry.getKey();
  108.                 String id = entry.getValue();
  110.                 byte[] bytes = map.get(id);
  111.                 if (bytes != null) {
  112.                     result.put(key, getValueTransformer().deserialize(bytes));
  113.                 }
  114.             }
  115.             return result;
  117.         } catch (Exception e) {
  118.             LOG.error("batchGet exception", e);
  119.             return new HashMap<>();
  120.         }
  121.     }
  123.     @Override
  124.     public void put(K key, V value) {
  125.         put(key, value, 0);
  126.     }
  128.     @Override
  129.     public void put(K key, V value, int ttlSecs) {
  130.         try {
  131.             getMemcachedClient().add(
  132.                     getKeyTransformer().getKey(key),
  133.                     ttlSecs,
  134.                     getValueTransformer().serialize(value));
  135.         } catch (Exception e) {
  136.             LOG.error("memcached put exception", e);
  137.         }
  138.     }
  140.     @Override
  141.     public void del(K key) {
  142.         try {
  143.             getMemcachedClient().delete(getKeyTransformer().getKey(key));
  144.         } catch (Exception e) {
  145.             LOG.error("memcached del exception", e);
  146.         }
  147.     }
  149.     @Override
  150.     public void clear() {
  151.         try {
  152.             getMemcachedClient().flushAll();
  153.         } catch (Exception e) {
  154.             LOG.error("memcached flushAll exception", e);
  155.         }
  156.     }
  157. }

如上所示，AbstractMemcachedCache预留了3个抽象getter方法:

  • memcachedClient
  • keyTransfomer
  • valueTransfomer

实现者可以根据自己的需求来实现。

自动配置

前面已经提到，MemcachedCache依赖MemcachedClient的实例。

如果每次都要手动构造MemcachedClient，实在是有些繁琐，我们可以通过Spring Boot的自动配置来自动注入：

  1. @Configuration
  2. @ConfigurationProperties(prefix = "memcached")
  3. public class MemcachedClientAutoConfiguration {
  5.     // Server list seperate by space
  6.     private String serverList;
  8.     // Connection Pool Size, default 64
  9.     private int connPoolSize = 64;
  11.     public String getServerList() {
  12.         return serverList;
  13.     }
  15.     public void setServerList(String serverList) {
  16.         this.serverList = serverList;
  17.     }
  19.     public int getConnPoolSize() {
  20.         return connPoolSize;
  21.     }
  23.     public void setConnPoolSize(int connPoolSize) {
  24.         this.connPoolSize = connPoolSize;
  25.     }
  27.     @Bean
  28.     @ConditionalOnMissingBean(MemcachedClient.class)
  29.     public MemcachedClient createMemcachedClient() throws IOException {
  30.         return MemcachedClientBuilder2.build(serverList, connPoolSize);
  31.     }
  34. }

如上所示，上述自动配置会扫描配置文件：

  • 若发现”memcached”开头的配置，会尝试解析其serverList和connPoolSize字段。
  • 若解析成功，会调用之前介绍的Builder，自动生成一个MemcachedClient。

Memcached的应用案例

我们通过一个简单的案例来说明MemcachedCache的使用。

设计一个接口，返回10秒内每个用户的第一次访问的时间戳。

我们通过MemcachedCache来实现，首先定义Cache:

  1. @Service
  2. public class TimestampMemcachedCache extends AbstractMemcachedCache<Integer, Long> {
  4.     @Autowired
  5.     private MemcachedClient client;
  7.     private CacheKeyTransformer<Integer> keyTransformer = new DefaultCacheKeyTransformer<>("timestamp");
  9.     private CacheValueTransformer<Long> valueTransformer = new LongValueTransformer();
  11.     @Override
  12.     protected MemcachedClient getMemcachedClient() {
  13.         return client;
  14.     }
  16.     @Override
  17.     protected CacheKeyTransformer<Integer> getKeyTransformer() {
  18.         return keyTransformer;
  19.     }
  21.     @Override
  22.     protected CacheValueTransformer<Long> getValueTransformer() {
  23.         return valueTransformer;
  24.     }
  25. }

如上所示，我们定义了类型的Cache，其中Integer的Key表示用户Id，Long类型的Value表示时间戳。

上述的MemcachedClient是自动注入的，我们需要做一下配置：

  1. memcached.serverList: "127.0.0.1:11211"

在使用的Service中，如下使用:

  2.     @Autowired
  3.     private TimestampMemcachedCache cache;
  5.     @Override
  6.     public String getCacheTimestampByUserId(int usrId) {
  7.         return String.valueOf(cache.cacheGet(userId, key -> System.currentTimeMillis(), 10));
  8.     }

如上所示，我们用了cacheGet来分别缓存最新时间戳，过期时间设为10秒钟。

通过这个例子，你一定体会到了：有了ICache等封装后，Memcached的使用变得非常简单。

小结

在本节中，我们设计了通过了ICache接口，并实现了LocalCache、MemcachedCache两种不同的Cache。其中，我们重点探讨了MemcachedClient实现的细节，包括MemcachedClient的自动注入、Memcached数据类型的转换(Transfomer)。

拓展阅读

  1. 实际的应用中，缓存的更新是一个较为复杂的任务，建议阅读缓存更新的套路。