java延迟map, 自定义延迟map, 过期清理map,map能力扩展。如何设置map数据过期,改造map适配数据过期
1. 功能:
map 线程安全,能够对存入的数据设置过期,或者自定义删除
2. aliyun代码看到的一个对象正好符合上述需求
出处是aliyun sdk core jar包的一个类。感兴趣可以去下载下jar查看
下面是源码:
package com.aliyuncs.policy.cache;import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Future;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import org.apache.commons.lang3.concurrent.BasicThreadFactory;public class ThrottlingPool {private static final Map<String, Entity> map = new ConcurrentHashMap();private static final ScheduledExecutorService executor = new ScheduledThreadPoolExecutor(1, (new BasicThreadFactory.Builder()).namingPattern("throttling-pool-%d").daemon(true).build());public ThrottlingPool() {}public static synchronized void put(String key, Object data) {put(key, data, -1);}public static synchronized void put(final String key, Object data, int expire) {remove(key);if (data != null) {if (expire >= 0) {Future future = executor.schedule(new Runnable() {public void run() {synchronized(ThrottlingPool.class) {ThrottlingPool.map.remove(key);}}}, (long)expire, TimeUnit.MILLISECONDS);map.put(key, new Entity(data, expire, future));} else {map.put(key, new Entity(data, expire, (Future)null));}}}public static synchronized Object get(String key) {Entity entity = (Entity)map.get(key);return entity != null ? entity.getValue() : null;}public static synchronized <T> T get(String key, Class<T> clazz) {return (T)clazz.cast(get(key));}public static synchronized int getExpire(String key) {Entity entity = (Entity)map.get(key);return entity != null ? entity.getExpire() : 0;}public static synchronized Object remove(String key) {Entity entity = (Entity)map.remove(key);if (entity == null) {return null;} else {Future future = entity.getFuture();if (future != null) {future.cancel(true);}return entity.getValue();}}public static synchronized int size() {return map.size();}public static synchronized void clear() {for(Entity entity : map.values()) {if (entity != null) {Future future = entity.getFuture();if (future != null) {future.cancel(true);}}}map.clear();}public static synchronized Map<String, Entity> getPool() {return map;}private static class Entity {private Object value;private int expire;private Future future;public Entity(Object value, int expire, Future future) {this.value = value;this.expire = expire;this.future = future;}public Object getValue() {return this.value;}public int getExpire() {return this.expire;}public Future getFuture() {return this.future;}}
}
2. 但是有个问题,如果数据量大,且都设置有过期时间,容易过期不及时!单线程处理不过来
3. 下面代码采用延迟队列一版:
import java.util.concurrent.*;public class ThrottlingPool {private static final ConcurrentHashMap<String, Object> map = new ConcurrentHashMap<>();private static final DelayQueue<DelayedCacheEntry> delayQueue = new DelayQueue<>();private static final ExecutorService executor = Executors.newSingleThreadExecutor(r -> {Thread t = new Thread(r);t.setDaemon(true);return t;});public static void put(String key, Object data, long expireMs) {long expirationTime = System.currentTimeMillis() + expireMs;delayQueue.removeIf(entry -> entry.getKey().equals(key));map.put(key, data);delayQueue.offer(new DelayedCacheEntry(key, expirationTime));}public static Object get(String key) {return map.get(key);}public static void remove(String key) {map.remove(key);}public static int size() {return map.size();}// 启动一个后台线程处理过期任务static {executor.execute(() -> {while (!Thread.currentThread().isInterrupted()) {try {DelayedCacheEntry entry = delayQueue.take();
// synchronized (ThrottlingPool.class) {map.remove(entry.getKey());
// }} catch (InterruptedException ex) {ex.printStackTrace();Thread.currentThread().interrupt();break;}}});// 钩子Runtime.getRuntime().addShutdownHook(new Thread(() -> {executor.shutdown();}));
// Thread cleanupThread = new Thread(() -> {
// while (true) {
// try {
// DelayedCacheEntry entry = delayQueue.take();
// synchronized (ThrottlingPool.class) {
// map.remove(entry.getKey());
// }
// } catch (InterruptedException e) {
// Thread.currentThread().interrupt();
// break;
// }
// }
// });
// cleanupThread.setDaemon(true);
// cleanupThread.start();}private static class DelayedCacheEntry implements Delayed {private final String key;private final long expirationTime;public DelayedCacheEntry(String key, long expirationTime) {this.key = key;this.expirationTime = expirationTime;}@Overridepublic long getDelay(TimeUnit unit) {long diff = expirationTime - System.currentTimeMillis();return unit.convert(diff, TimeUnit.MILLISECONDS);}@Overridepublic int compareTo(Delayed o) {return Long.compare(this.expirationTime, ((DelayedCacheEntry) o).expirationTime);}public String getKey() {return key;}}}
4.本人水平有限,如有问题,欢迎指正