第三集：分布式Ehcache緩存改造
前言
? 好久沒有寫博客了，大有半途而廢的趨勢。忙不是借口，這個好習慣還是要繼續堅持。前面我承諾的第一期的DIY分布式，是時候上終篇了---DIY分布式緩存。
探索之路
? 在前面的文章中，我給大家大致說過項目背景：項目中的緩存使用的是Ehcache。因為前面使用Ehcache的應用就一臺，所以這種單機的Ehcache并不會有什么問題。現在分布式部署之后，如果各個應用之間的緩存不能共享，那么其實各自就是一個孤島。可能在一個業務跑下來，請求了不同的應用，結果在緩存中取出來的值不一樣，
造成數據不一致。所以需要重新設計緩存的實現。
? 因為盡量不要引入新的中間件，所以改造仍然是圍繞Ehcache來進行的。搜集了各種資料之后，發現Ehcache實現分布式緩存基本有以下兩種思路：
客戶端實現分布式算法: 在使用Ehcache的客戶端自己實現分布式算法。
算法的基本思路就是取模：即假設有三臺應用（編號假設分別為0,1,2），對于一個要緩存的對象，首先計算其key的hash值，然后將hash值模3，得到的余數是幾，就將數據緩存到哪臺機器。
同步冗余數據: Ehcache是支持集群配置的，集群的各個節點之間支持按照一定的協議進行數據同步。這樣每臺應用其實緩存了一整份數據，不同節點之間的數據是一致的。
? 雖然冗余的辦法顯得有點浪費資源，但是我最終還是選擇了冗余。具體原因有以下幾點：
- 分布式算法的復雜性: 前面所講的分布式算法只是最基本的實現。事實上實現要比這個復雜的多。需要考慮增加或者刪除節點的情況，需要使用更加復雜的一致性hash算法
- 可能導致整個應用不可用： 當刪除節點之后，如果算法不能夠感知進行自動調整，仍然去請求那個已經被刪除的節點，可能導致整個系統不可用。
Demo
? 最終我的實現采用RMI的方式進行同步
配置ehcache
? spring-ehcache-cache.xml
<?xml version="1.0" encoding="UTF-8"?>
<ehcache xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"xsi:noNamespaceSchemaLocation="http://ehcache.org/ehcache.xsd" name="businessCaches"><diskStore path="java.io.tmpdir/ehcache"/><cache name="business1Cache"maxElementsInMemory="10000000"eternal="true"overflowToDisk="false"memoryStoreEvictionPolicy="LRU"><cacheEventListenerFactoryclass="net.sf.ehcache.distribution.RMICacheReplicatorFactory"/></cache><cache name="business2Cache"maxElementsInMemory="100"eternal="true"overflowToDisk="false"memoryStoreEvictionPolicy="LRU"><cacheEventListenerFactoryclass="net.sf.ehcache.distribution.RMICacheReplicatorFactory"/></cache><!-- cache發布信息配置，人工發現peerDiscovery=manual,cacheNames可配置多個緩存名稱，以|分割 ） --><cacheManagerPeerProviderFactoryclass="com.rampage.cache.distribute.factory.DisRMICacheManagerPeerProviderFactory"properties="peerDiscovery=manual, cacheNames=business1Cache|business2Cache" /><!-- 接收同步cache信息的地址 --><cacheManagerPeerListenerFactoryclass="com.rampage.cache.distribute.factory.DisRMICacheManagerPeerListenerFactory"properties="socketTimeoutMillis=2000" />
</ehcache>? spring-cache.xml
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"xmlns:cache="http://www.springframework.org/schema/cache"xmlns:context="http://www.springframework.org/schema/context"xmlns:aop="http://www.springframework.org/schema/aop"xsi:schemaLocation="http://www.springframework.org/schema/cache http://www.springframework.org/schema/cache/spring-cache.xsdhttp://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsdhttp://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context-4.2.xsdhttp://www.springframework.org/schema/aop http://www.springframework.org/schema/aop/spring-aop-4.2.xsd"default-autowire="byName"><!-- 包掃描 --><context:component-scan base-package="com.rampage.cache" /><!-- 啟用Cache注解 --><cache:annotation-driven cache-manager="cacheManager"key-generator="keyGenerator" proxy-target-class="true" /><!-- 自定義的緩存key生成類，需實現org.springframework.cache.interceptor.KeyGenerator接口 --><bean id="keyGenerator" class="com.rampage.cache.support.CustomKeyGenerator" /><!-- 替換slite的ehcache實現 --><bean id="ehCacheManagerFactory" class="org.springframework.cache.ehcache.EhCacheManagerFactoryBean"><property name="configLocation" value="classpath:spring/cache/sppay-ehcache-cache.xml"/><!-- value對應前面ehcache文件定義的manager名稱 --><property name="cacheManagerName" value="businessCaches" /></bean><bean id="ehCacheManager" class="org.springframework.cache.ehcache.EhCacheCacheManager"><property name="cacheManager" ref="ehCacheManagerFactory"/></bean><bean id="cacheManager" class="org.springframework.cache.support.CompositeCacheManager"><property name="cacheManagers"><list><ref bean="ehCacheManager" /></list></property><property name="fallbackToNoOpCache" value="true" /></bean>
</beans>實現自定義轉發和監聽
? 細心的讀者應該不難發現，前面xml配置中cacheManagerPeerProviderFactory和cacheManagerPeerListenerFactory我使用的都是自定義的類。之所以使用自定義的類，是為了在初始化的時候發布的地址和端口，監聽的地址端口可以在配置文件配置。具體類的實現如下:
/*** 分布式EhCache監聽工廠* @author secondWorld**/
public class DisRMICacheManagerPeerListenerFactory extends RMICacheManagerPeerListenerFactory {private static final Logger LOGGER = LoggerFactory.getLogger(DisRMICacheManagerPeerListenerFactory.class);/*** 配置文件中配置的監聽地址，可以不配置，默認為本機地址*/private static final String LISTEN_HOST = "distribute.ehcache.listenIP";/*** 配置文件中配置的監聽端口*/private static final String LISTEN_PORT = "distribute.ehache.listenPort";@Overrideprotected CacheManagerPeerListener doCreateCachePeerListener(String hostName, Integer port,Integer remoteObjectPort, CacheManager cacheManager, Integer socketTimeoutMillis) {// xml中hostName為空，則讀取配置文件(app-config.properties)中的值if (StringUtils.isEmpty(hostName)) {String propHost = AppConfigPropertyUtils.get(LISTEN_HOST);if (StringUtils.isNotEmpty(propHost)) {hostName = propHost;}}// 端口采用默認端口0，則去讀取配置文件（app-config.properties）中的值if (port != null && port == 0) {Integer propPort = null;try {propPort = Integer.parseInt(AppConfigPropertyUtils.get(LISTEN_PORT));} catch (NumberFormatException e) {}if (propPort != null) {port = propPort;}}LOGGER.info("Initiliazing DisRMICacheManagerPeerListenerFactory:cacheManager[{}], hostName[{}], port[{}], remoteObjectPort[{}], socketTimeoutMillis[{}]......",cacheManager, hostName, port, remoteObjectPort, socketTimeoutMillis);return super.doCreateCachePeerListener(hostName, port, remoteObjectPort, cacheManager, socketTimeoutMillis);}
}/*** 分布式EhCache發布工廠* * @author secondWorld**/
public class DisRMICacheManagerPeerProviderFactory extends RMICacheManagerPeerProviderFactory {private static final Logger LOGGER = LoggerFactory.getLogger(DisRMICacheManagerPeerProviderFactory.class);private static final String CACHENAME_DELIMITER = "|";private static final String PROVIDER_ADDRESSES = "distribute.ehcache.providerAddresses";private static final String CACHE_NAMES = "cacheNames";/*** rmi地址格式： //127.0.0.1:4447/Cache1|//127.0.0.1:4447/Cache2*/@Overrideprotected CacheManagerPeerProvider createManuallyConfiguredCachePeerProvider(Properties properties) {// 從app-config.properties中讀取發布地址列表String providerAddresses = AppConfigPropertyUtils.get(PROVIDER_ADDRESSES, StringUtils.EMPTY);// 從ehcache配置文件讀取緩存名稱String cacheNames = PropertyUtil.extractAndLogProperty(CACHE_NAMES, properties);// 參數校驗，這里發布地址和緩存名稱都不能為空if (StringUtils.isEmpty(providerAddresses) || StringUtils.isEmpty(cacheNames)) {throw new IllegalArgumentException("Elements \"providerAddresses\" and \"cacheNames\" are needed!");}// 解析地址列表List<String> cachesNameList = getCacheNameList(cacheNames);List<String> providerAddressList = getProviderAddressList(providerAddresses);// 注冊發布節點RMICacheManagerPeerProvider rmiPeerProvider = new ManualRMICacheManagerPeerProvider();StringBuilder sb = new StringBuilder();for (String cacheName : cachesNameList) {for (String providerAddress : providerAddressList) {sb.setLength(0);sb.append("//").append(providerAddress).append("/").append(cacheName);rmiPeerProvider.registerPeer(sb.toString());LOGGER.info("Registering peer provider [{}]", sb);}}return rmiPeerProvider;}/*** 得到發布地址列表* @param providerAddresses 發布地址字符串* @return 發布地址列表*/private List<String> getProviderAddressList(String providerAddresses) {StringTokenizer stringTokenizer = new StringTokenizer(providerAddresses,AppConfigPropertyUtils.APP_ITEM_DELIMITER);List<String> ProviderAddressList = new ArrayList<String>(stringTokenizer.countTokens());while (stringTokenizer.hasMoreTokens()) {String providerAddress = stringTokenizer.nextToken();providerAddress = providerAddress.trim();ProviderAddressList.add(providerAddress);}return ProviderAddressList;}/*** 得到緩存名稱列表* @param cacheNames 緩存名稱字符串* @return 緩存名稱列表*/private List<String> getCacheNameList(String cacheNames) {StringTokenizer stringTokenizer = new StringTokenizer(cacheNames, CACHENAME_DELIMITER);List<String> cacheNameList = new ArrayList<String>(stringTokenizer.countTokens());while (stringTokenizer.hasMoreTokens()) {String cacheName = stringTokenizer.nextToken();cacheName = cacheName.trim();cacheNameList.add(cacheName);}return cacheNameList;}@Overrideprotected CacheManagerPeerProvider createAutomaticallyConfiguredCachePeerProvider(CacheManager cacheManager,Properties properties) throws IOException {throw new UnsupportedOperationException("Not supported automatic distribute cache!");}
}
配置
? 假設有三臺機器，則他們分別得配置如下:
#應用1，在4447端口監聽
#緩存同步消息發送地址（如果同步到多臺需要配置多臺地址，多臺地址用英文逗號分隔）
distribute.ehcache.providerAddresses=127.0.0.1:4446,127.0.0.1:4448
#緩存同步監聽端口和IP
distribute.ehache.listenPort=4447
distribute.ehcache.listenIP=localhost#應用2，在4448端口監聽
#緩存同步消息發送地址（如果同步到多臺需要配置多臺地址，多臺地址用英文逗號分隔）
distribute.ehcache.providerAddresses=127.0.0.1:4446,127.0.0.1:4447
#緩存同步監聽端口和IP
distribute.ehache.listenPort=4448
distribute.ehcache.listenIP=localhost#應用3，在4446端口監聽
#緩存同步消息發送地址（如果同步到多臺需要配置多臺地址，多臺地址用英文逗號分隔）
distribute.ehcache.providerAddresses=127.0.0.1:4447,127.0.0.1:4448
#緩存同步監聽端口和IP
distribute.ehache.listenPort=4446
distribute.ehcache.listenIP=localhost使用
? 使用的時候直接通過Spring的緩存注解即可。簡單的示例如下:
@CacheConfig("business1Cache")
@Component
public class Business1 {@Cacheablepublic String getData(String key) {// TODO:...}
}說明
? 前面的實現是通過RMI的方式來實現緩存同步的，相對來說RMI的效率還是很快的。所以如果不需要實時的緩存一致性，允許少許延遲，那么這種方式的實現足夠。
總結
? 到這篇完成，分布式改造的第一章算是告一段落了。對于分布式，如果可以選擇，必然要選擇現在成熟的框架。但是項目有很多時候，由于各種歷史原因，必須要在原來的基礎上改造。這個時候，希望我寫的這個系列對大家有所幫助。造輪子有時候就是這么簡單。
相關鏈接
黎明前最黑暗，成功前最絕望！
- https://www.cnblogs.com/Kidezyq/p/8748961.html
- https://www.cnblogs.com/Kidezyq/p/8977750.html
