基于memcached for java 实现通用分布式缓存和集群分布式缓存

Dear 丶 2024-02-17 18:25 22阅读 0赞

本文参考借鉴：[http://guazi.iteye.com/blog/107164][http_guazi.iteye.com_blog_107164]

前提：基于memcached client for java 的基础进行的二次封装，实现缓存存储的两种模式：通用分布式缓存和集群分布式缓存。以下是对于memcached client for Java 二次封装的UML图。

![Center][]

对于memcached的客户端初始化在CacheFactory中通过读取配置文件cacheConfig.xml完成。通用分布式缓存，只是一个简单的封装，利用memcached client for java提供的分布式支持来实现，这里主要说一下clusterCache的实现思想：对存入的缓存对象的key值进行一次hash，找到对应的服务器存入，然后根据一定的规则再次进行hash，找到另外一个不同的服务器存入，取缓存时，先对要取的key值进行一次hash，找到主服务器，如果获取失败或者获取到的值为null，就对key进行再次hash，找到其从服务器，从这台服务器取缓存结果（如果取到结果就异步的更新到主服务器），这样就形成了主从式集群缓存。特点是：没有绝对的主节点和从节点，正常情况下所有服务器共同承担缓存服务器，在一台服务器出现异常时其他服务器共同承担增加的访问压力。

拓扑结构如下：

![Center 1][]

源代码

package com.yx.cache;
    
    public interface Cache<T> {
    	/**
    	 * 获取缓存中的数据
    	 * 
    	 * @param key
    	 * @return
    	 */
    	T get(String key);
    
    	/**
    	 * 把数据放入缓存 如果存在与key对应的值，则返回失败
    	 * 
    	 * @param key
    	 * @param value
    	 * @return
    	 */
    	boolean add(String key, T value);
    
    	/**
    	 * 把数据放入缓存 如果存在与key对应的值，则覆盖原有的值
    	 * 
    	 * @param key
    	 * @param value
    	 * @return
    	 */
    	boolean set(String key, T value);
    
    	/**
    	 * 缓存更新 如果不存在与key对应的缓存值，则不更新
    	 * 
    	 * @param key
    	 * @param value
    	 * @return
    	 */
    	boolean update(String key, T value);
    
    	/**
    	 * 删除缓存
    	 * 
    	 * @param key
    	 * @return
    	 */
    	boolean delete(String key);
    }

通用分布式缓存实现类：

package com.yx.cache;
    
    import com.danga.MemCached.MemCachedClient;
    
    public class CommonCache<T> implements Cache<T> {
    	private static MemCachedClient memCachedClient = null;
    
    	private String base = null;
    
    	CommonCache(Class<T> t, MemCachedClient client) {
    		memCachedClient = client;
    		base = t.getSimpleName() + "-";
    	}
    
    	public T get(String key) {
    		return (T) memCachedClient.get(base + key);
    	}
    
    	public boolean set(String key, T value) {
    		return memCachedClient.set(base + key, value);
    	}
    
    	@Override
    	public boolean update(String key, T value) {
    		return memCachedClient.replace(base + key, value);
    	}
    
    	@Override
    	public boolean delete(String key) {
    		return memCachedClient.delete(base + key);
    	}
    
    	@Override
    	public boolean add(String key, T value) {
    		return memCachedClient.add(base + key, value);
    	}
    }

集群分布式缓存实现类

package com.yx.cache;
    
    import com.danga.MemCached.MemCachedClient;
    import com.schooner.MemCached.SchoonerSockIOPool;
    import com.yx.cache.util.HashCodeUtil;
    import com.yx.task.ThreadPoolManager;
    
    public class ClusterCache<T> implements Cache<T> {
    	private static MemCachedClient memCachedClient = null;
    
    	private static ThreadPoolManager taskManager = ThreadPoolManager
    			.getInstance("cache");
    
    	private String base = null;
    
    	private SchoonerSockIOPool pool = SchoonerSockIOPool.getInstance();
    
    	ClusterCache(Class<T> t, MemCachedClient client) {
    		memCachedClient = client;
    		base = "i-" + t.getSimpleName() + "-";
    	}
    
    	@Override
    	public T get(String key) {
    		T value = null;
    		if (key == null) {
    			return null;
    		}
    		key = base + key;
    		if (pool.getServers().length < 2) {
    			value = (T) memCachedClient.get(key);
    		} else {
    			int hashCode = HashCodeUtil.getHash(key);
    
    			value = (T) memCachedClient.get(key, hashCode);
    			if (value == null) {
    				hashCode = this.getRehashCode(key, hashCode);
    
    				value = (T) memCachedClient.get(key, hashCode);
    				if (value != null) {// 如果在另外一台服务器上取到了缓存，则恢复第一台服务器
    					UpdateTask task = new UpdateTask(key, value);
    					taskManager.submit(task);
    				}
    
    			}
    		}
    		return value;
    	}
    
    	@Override
    	public boolean set(String key, T value) {
    		if (key == null) {
    			return false;
    		}
    		key = base + key;
    		boolean result = false;
    		if (pool.getServers().length < 2) {
    			result = memCachedClient.set(key, value);
    		} else {
    			int hashCode = HashCodeUtil.getHash(key);
    
    			result = memCachedClient.set(key, value, hashCode);
    			// if (result) {
    			hashCode = getRehashCode(key, hashCode);
    			memCachedClient.set(key, value, hashCode);
    
    			// }
    		}
    		return result;
    	}
    
    	private int getRehashCode(String key, int oldHashcode) {
    		String host = pool.getHost(key, oldHashcode);
    		int rehashTries = 0;
    		// if (result) {
    		int hashCode = HashCodeUtil.getHash(rehashTries + key);
    		while (host.equals(pool.getHost(key, hashCode))) {
    			rehashTries++;
    			hashCode = HashCodeUtil.getHash(rehashTries + key);
    		}
    		return hashCode;
    	}
    
    	@Override
    	public boolean update(String key, T value) {
    		if (key == null) {
    			return false;
    		}
    		key = base + key;
    		boolean result = false;
    		if (pool.getServers().length < 2) {
    			result = memCachedClient.replace(key, value);
    		} else {
    			int hashCode = HashCodeUtil.getHash(key);
    
    			result = memCachedClient.replace(key, value, hashCode);
    			// if (result) {
    			hashCode = getRehashCode(key, hashCode);
    			memCachedClient.replace(key, value, hashCode);
    
    			// }
    		}
    		return result;
    	}
    
    	@Override
    	public boolean delete(String key) {
    		if (key == null) {
    			return false;
    		}
    		key = base + key;
    		boolean result = false;
    		if (pool.getServers().length < 2) {
    			result = memCachedClient.delete(key);
    		} else {
    			int hashCode = HashCodeUtil.getHash(key);
    
    			result = memCachedClient.delete(key, hashCode, null);
    			// if (result) {
    			hashCode = this.getRehashCode(key, hashCode);
    
    			memCachedClient.delete(key, hashCode, null);
    
    			// }
    		}
    		return result;
    	}
    
    	@Override
    	public boolean add(String key, T value) {
    		if (key == null) {
    			return false;
    		}
    		key = base + key;
    		boolean result = false;
    		if (pool.getServers().length < 2) {
    			result = memCachedClient.add(key, value);
    		} else {
    			int hashCode = HashCodeUtil.getHash(key);
    
    			result = memCachedClient.add(key, value, hashCode);
    			// if (result) {
    			hashCode = getRehashCode(key, hashCode);
    			memCachedClient.add(key, value, hashCode);
    
    			// }
    		}
    		return result;
    	}
    
    	static class UpdateTask implements Runnable {
    
    		private String key;
    		private Object value;
    
    		UpdateTask(String key, Object value) {
    			this.key = key;
    			this.value = value;
    		}
    
    		@Override
    		public void run() {
    			memCachedClient.set(key, value, HashCodeUtil.getHash(key));
    		}
    
    	}
    }

基于工厂模式创建memcached 存储模式（通用模式还是集群模式）

package com.yx.cache;
    
    import java.util.ArrayList;
    import java.util.List;
    import java.util.Map;
    import java.util.concurrent.ConcurrentHashMap;
    
    import com.danga.MemCached.MemCachedClient;
    import com.danga.MemCached.SockIOPool;
    import com.yx.cache.util.ConfigUtil;
    
    public class CacheFactory {
    	private static MemCachedClient memCachedClient = null;
    
    	@SuppressWarnings("rawtypes")
    	private static final Map<String, Cache> map = new ConcurrentHashMap<String, Cache>();
    
    	static {
    		String serverStr = ConfigUtil.getConfigValue("servers", "");
    
    		List<String> servers = new ArrayList<String>();
    		for (String s : serverStr.split(",")) {
    			s = s.trim();
    			if (!"".equals(s)) {
    				servers.add(s);
    			}
    		}
    		if (servers.size() < 1) {
    			throw new RuntimeException("cache 初始化失败！");
    		}
    		SockIOPool pool = SockIOPool.getInstance();
    		pool.setServers(servers.toArray(new String[] {}));
    		pool.setFailover(Boolean.valueOf(ConfigUtil.getConfigValue("failover",
    				"true")));
    		pool.setInitConn(Integer.valueOf(ConfigUtil.getConfigValue("initConn",
    				"100")));
    		pool.setMinConn(Integer.valueOf(ConfigUtil.getConfigValue("minConn",
    				"25")));
    		pool.setMaxConn(Integer.valueOf(ConfigUtil.getConfigValue("maxConn",
    				"250")));
    		pool.setMaintSleep(Integer.valueOf(ConfigUtil.getConfigValue(
    				"maintSleep", "30")));
    		pool.setNagle(Boolean.valueOf(ConfigUtil.getConfigValue("nagle",
    				"false")));// 关闭nagle算法
    		pool.setSocketTO(Integer.valueOf(ConfigUtil.getConfigValue("socketTO",
    				"3000")));
    		pool.setAliveCheck(Boolean.valueOf(ConfigUtil.getConfigValue(
    				"aliveCheck", "true")));
    		pool.setHashingAlg(Integer.valueOf(ConfigUtil.getConfigValue(
    				"hashingAlg", "0")));
    		pool.setSocketConnectTO(Integer.valueOf(ConfigUtil.getConfigValue(
    				"socketConnectTO", "3000")));
    		String wStr = ConfigUtil.getConfigValue("weights", "");
    		List<Integer> weights = new ArrayList<Integer>();
    		for (String s : wStr.split(",")) {
    			s = s.trim();
    			if (!"".equals(s)) {
    				weights.add(Integer.valueOf(s));
    
    			}
    		}
    		if (weights.size() == servers.size()) {
    			pool.setWeights(weights.toArray(new Integer[] {}));
    		}
    		pool.initialize();
    		memCachedClient = new MemCachedClient();
    
    	}
    
    	public static <T> Cache<T> getCommonCache(Class<T> t) {
    		Cache<T> cache = map.get(t.getName());
    		if (cache == null) {
    			cache = createCommonCache(t);
    		}
    		return cache;
    	}
    
    	public static <T> Cache<T> getClusterCache(Class<T> t) {
    		Cache<T> cache = map.get("i-" + t.getName());
    		if (cache == null) {
    			cache = createClusterCache(t);
    		}
    		return cache;
    	}
    
    	private static synchronized <T> Cache<T> createCommonCache(Class<T> t) {
    		Cache<T> cache = map.get(t.getName());
    		if (cache == null) {
    			cache = new CommonCache<T>(t, memCachedClient);
    			map.put(t.getName(), cache);
    		}
    		return cache;
    	}
    
    	private static synchronized <T> Cache<T> createClusterCache(Class<T> t) {
    		Cache<T> cache = map.get(t.getName());
    		if (cache == null) {
    			cache = new ClusterCache<T>(t, memCachedClient);
    			map.put(t.getName(), cache);
    		}
    		return cache;
    	}
    }

读取配置文件工具类封装和生成Hash代码工具类

ConfigUtil.Java和HashCodeUtil.java

package com.yx.cache.util;
    
    import java.io.IOException;
    import java.io.InputStream;
    import java.util.HashMap;
    import java.util.Iterator;
    import java.util.Map;
    
    import org.dom4j.Document;
    import org.dom4j.DocumentException;
    import org.dom4j.Element;
    import org.dom4j.io.SAXReader;
    
    public class ConfigUtil {
    
    	private static final String CONFILE = "cacheConfig.xml";
    	private static final Map<String, String> map = new HashMap<String, String>();
    
    	static {
    		SAXReader saxReader = new SAXReader();
    		InputStream ins = ConfigUtil.class.getClassLoader()
    				.getResourceAsStream(CONFILE);
    		try {
    			if (ins != null) {
    				Document doc = saxReader.read(ins);
    				Element root = doc.getRootElement();
    				Iterator<Element> iter = root.elementIterator();
    				while (iter.hasNext()) {
    					Element e = iter.next();
    					map.put(e.getName(), e.getTextTrim());
    				}
    			}
    		} catch (DocumentException e) {
    			// TODO Auto-generated catch block
    			e.printStackTrace();
    			throw new RuntimeException("找不到配置文件：" + CONFILE);
    		} finally {
    			try {
    				if (ins != null) {
    					ins.close();
    				} else {
    					throw new RuntimeException("找不到配置文件：" + CONFILE);
    				}
    			} catch (IOException e) {
    				// TODO Auto-generated catch block
    				e.printStackTrace();
    			}
    		}
    	}
    
    	public static String getConfigValue(String key, String defaultValue) {
    		String tmp = map.get(key);
    		return isEmpty(tmp) ? defaultValue : tmp;
    	}
    
    	public static void main(String[] args) {
    		System.out.println(map);
    	}
    
    	private static boolean isEmpty(String str) {
    		if (str == null || "".equals(str)) {
    			return true;
    		}
    		return false;
    	}
    }

package com.yx.cache.util;
    
    import java.security.MessageDigest;
    import java.security.NoSuchAlgorithmException;
    import java.util.zip.CRC32;
    
    import com.schooner.MemCached.SchoonerSockIOPool;
    
    public class HashCodeUtil {
    
    	public static final int NATIVE_HASH = 0; // native String.hashCode();
    	public static final int OLD_COMPAT_HASH = 1; // original compatibility
    	public static final int NEW_COMPAT_HASH = 2; // new CRC32 based
    	public static final int CONSISTENT_HASH = 3; // MD5 Based -- Stops
    
    	private static int hashingAlg = SchoonerSockIOPool.getInstance()
    			.getHashingAlg();
    
    	/**
    	 * Returns a bucket to check for a given key.
    	 * 
    	 * @param key
    	 *            String key cache is stored under
    	 * @return int bucket
    	 */
    	public static final int getHash(String key) {
    
    		switch (hashingAlg) {
    		case NATIVE_HASH:
    			return key.hashCode();
    		case OLD_COMPAT_HASH:
    			return origCompatHashingAlg(key);
    		case NEW_COMPAT_HASH:
    			return newCompatHashingAlg(key);
    		case CONSISTENT_HASH:
    			return md5HashingAlg(key);
    		default:
    			// use the native hash as a default
    			hashingAlg = NATIVE_HASH;
    			return key.hashCode();
    		}
    	}
    
    	private static int origCompatHashingAlg(String key) {
    		int hash = 0;
    		char[] cArr = key.toCharArray();
    
    		for (int i = 0; i < cArr.length; ++i) {
    			hash = (hash * 33) + cArr[i];
    		}
    
    		return hash;
    	}
    
    	private static int newCompatHashingAlg(String key) {
    		CRC32 checksum = new CRC32();
    		checksum.update(key.getBytes());
    		int crc = (int) checksum.getValue();
    		return (crc >> 16) & 0x7fff;
    	}
    
    	private static int md5HashingAlg(String key) {
    		MessageDigest md5 = MD5.get();
    		md5.reset();
    		md5.update(key.getBytes());
    		byte[] bKey = md5.digest();
    		int res = ((bKey[3] & 0xFF) << 24) | ((bKey[2] & 0xFF) << 16)
    				| ((bKey[1] & 0xFF) << 8) | (bKey[0] & 0xFF);
    		return res;
    	}
    
    	private static ThreadLocal<MessageDigest> MD5 = new ThreadLocal<MessageDigest>() {
    		@Override
    		protected final MessageDigest initialValue() {
    			try {
    				return MessageDigest.getInstance("MD5");
    			} catch (NoSuchAlgorithmException e) {
    				throw new IllegalStateException(" no md5 algorythm found");
    			}
    		}
    	};
    
    }

对于集群分布式缓存还缺少一个工具类ThreadPoolManage.java

package com.yx.task;
    
    import java.util.HashMap;
    import java.util.Map;
    import java.util.concurrent.Executors;
    import java.util.concurrent.ThreadPoolExecutor;
    
    /**
     * @author liuyuxiao
     * @Date 2011-5-30 下午04:34:16
     */
    public class ThreadPoolManager {
    
    	private static final Map<String, ThreadPoolManager> map = new HashMap<String, ThreadPoolManager>();
    
    	final int CORE_SIZE = 5;
    
    	private ThreadPoolExecutor executor = (ThreadPoolExecutor) Executors
    			.newFixedThreadPool(CORE_SIZE);
    
    	public void submit(Runnable task) {
    		executor.submit(task);
    	}
    
    	public boolean finished() {
    		return executor.getCompletedTaskCount() == executor.getTaskCount();
    	}
    
    	private ThreadPoolManager() {
    
    	}
    
    	public static synchronized ThreadPoolManager getInstance(String key) {
    		ThreadPoolManager t = map.get(key);
    		if (t == null) {
    			t = new ThreadPoolManager();
    			map.put(key, t);
    		}
    		return t;
    	}
    }

对于集群缓存模式和通用缓存模式测试：

package com.yx.cache.test;
    
    import com.yx.cache.Cache;
    import com.yx.cache.CacheFactory;
    
    public class TestCommonCache {
    
    	/**
    	 * @param args
    	 */
    	public static void main(String[] args) {
    
    		Cache<String> cache = CacheFactory.getCommonCache(String.class);
    		int count = 0;
    		for (int i = 0; i < 100; i++) {
    
    			// cache.set("" + i, "Hello!" + i);
    
    			String result = cache.get("" + i);
    			// System.out.println(String.format("set( %d ): %s", i, success));
    			if (result == null) {
    				count++;
    			}
    			System.out.println(String.format("get( %d ): %s", i, result));
    		}
    
    		System.out.println(count);
    		// for (int i = 0; i < 500; i++) {
    		// MemTask task = new MemTask();
    		// Thread t = new Thread(task);
    		// t.start();
    		// }
    
    	}
    }

package com.yx.cache.test;
    
    import com.yx.cache.Cache;
    import com.yx.cache.CacheFactory;
    
    public class TestClusterCache {
    
    	public static void main(String[] args) {
    		Cache<String> cache = CacheFactory.getClusterCache(String.class);
    		int count = 0;
    		for (int i = 0; i < 100; i++) {
    
    			// cache.set("" + i, "Hello!" + i);
    
    			String result = cache.get("" + i);
    			// System.out.println(String.format("set( %d ): %s", i, success));
    			if (result == null) {
    				count++;
    			}
    			System.out.println(String.format("get( %d ): %s", i, result));
    		}
    	}
    }

[http_guazi.iteye.com_blog_107164]: http://guazi.iteye.com/blog/1071646
[Center]: https://image.dandelioncloud.cn/pgy_files/images/2024/01/29/897ffcaa805a44c291b94ceacce10055.png
[Center 1]: https://image.dandelioncloud.cn/pgy_files/images/2024/01/29/de2a830216594e89bdb2ec8cc1891294.png