Java 类com.google.common.cache.RemovalNotification 实例源码

public AccessTokenJob() {
    logger.info("init");
    accessTokenCache = CacheBuilder.newBuilder()
            // 设置并发级别为200，并发级别是指可以同时写缓存的线程数
            .concurrencyLevel(200)
            // 设置写缓存后1分钟过期
            .expireAfterWrite(90, TimeUnit.MINUTES).initialCapacity(10).maximumSize(100)
            // 设置要统计缓存的命中率
            .recordStats()
            // 设置缓存的移除通知
            .removalListener(new RemovalListener<AppIdSecret, String>() {
                @Override
                public void onRemoval(RemovalNotification<AppIdSecret, String> notification) {
                    logger.info(notification.getKey() + " was removed, cause by " + notification.getCause());
                }
            }).build(new CacheLoader<AppIdSecret, String>() {
                // build方法中可以指定CacheLoader，在缓存不存在时通过CacheLoader的实现自动加载缓存
                @Override
                public String load(AppIdSecret appIdSecret) throws Exception {
                    Token token = CommonUtil.getAccessToken(appIdSecret.getAppId(), appIdSecret.getAppSecret());
                    return token.getToken();
                }
            });
}

/**
 * Create a new KMSAudit.
 *
 * @param windowMs Duplicate events within the aggregation window are quashed
 *                 to reduce log traffic. A single message for aggregated
 *                 events is printed at the end of the window, along with a
 *                 count of the number of aggregated events.
 */
KMSAudit(long windowMs) {
  cache = CacheBuilder.newBuilder()
      .expireAfterWrite(windowMs, TimeUnit.MILLISECONDS)
      .removalListener(
          new RemovalListener<String, AuditEvent>() {
            @Override
            public void onRemoval(
                RemovalNotification<String, AuditEvent> entry) {
              AuditEvent event = entry.getValue();
              if (event.getAccessCount().get() > 0) {
                KMSAudit.this.logEvent(event);
                event.getAccessCount().set(0);
                KMSAudit.this.cache.put(entry.getKey(), event);
              }
            }
          }).build();
  executor = Executors.newScheduledThreadPool(1, new ThreadFactoryBuilder()
      .setDaemon(true).setNameFormat(KMS_LOGGER_NAME + "_thread").build());
  executor.scheduleAtFixedRate(new Runnable() {
    @Override
    public void run() {
      cache.cleanUp();
    }
  }, windowMs / 10, windowMs / 10, TimeUnit.MILLISECONDS);
}

/**
 * Create a new KMSAudit.
 *
 * @param windowMs Duplicate events within the aggregation window are quashed
 *                 to reduce log traffic. A single message for aggregated
 *                 events is printed at the end of the window, along with a
 *                 count of the number of aggregated events.
 */
RENAudit(long windowMs) {
  cache = CacheBuilder.newBuilder()
      .expireAfterWrite(windowMs, TimeUnit.MILLISECONDS)
      .removalListener(
          new RemovalListener<String, AuditEvent>() {
            @Override
            public void onRemoval(
                RemovalNotification<String, AuditEvent> entry) {
              AuditEvent event = entry.getValue();
              if (event.getAccessCount().get() > 0) {
                RENAudit.this.logEvent(event);
                event.getAccessCount().set(0);
                RENAudit.this.cache.put(entry.getKey(), event);
              }
            }
          }).build();
  executor = Executors.newScheduledThreadPool(1, new ThreadFactoryBuilder()
      .setDaemon(true).setNameFormat(REN_LOGGER_NAME + "_thread").build());
  executor.scheduleAtFixedRate(new Runnable() {
    @Override
    public void run() {
      cache.cleanUp();
    }
  }, windowMs / 10, windowMs / 10, TimeUnit.MILLISECONDS);
}

/**
 * 
 */
public AuthorizationController() {
    cache = CacheBuilder.newBuilder()
            // 设置并发级别为200，并发级别是指可以同时写缓存的线程数
            .concurrencyLevel(200)
            // 设置写缓存后1分钟过期
            .expireAfterWrite(2, TimeUnit.MINUTES).initialCapacity(10).maximumSize(100)
            // 设置要统计缓存的命中率
            .recordStats()
            // 设置缓存的移除通知
            .removalListener(new RemovalListener<String, SNSUserInfo>() {
                @Override
                public void onRemoval(RemovalNotification<String, SNSUserInfo> notification) {
                    log.info(notification.getKey() + " was removed, cause by " + notification.getCause());
                }
            }).build(new CacheLoader<String, SNSUserInfo>() {
                // build方法中可以指定CacheLoader，在缓存不存在时通过CacheLoader的实现自动加载缓存
                @Override
                public SNSUserInfo load(String appIdSecret) throws Exception {
                    return userInfoCache.get(appIdSecret);
                }
            });
}

@Override
public void onRemoval(RemovalNotification<CacheKey, CompiledScript> notification) {
    if (logger.isDebugEnabled()) {
        logger.debug("notifying script services of script removal due to: [{}]", notification.getCause());
    }
    scriptMetrics.onCacheEviction();
    for (ScriptEngineService service : scriptEngines) {
        try {
            service.scriptRemoved(notification.getValue());
        } catch (Exception e) {
            logger.warn("exception calling script removal listener for script service", e);
            // We don't rethrow because Guava would just catch the
            // exception and log it, which we have already done
        }
    }
}

@Override
public void onRemoval(RemovalNotification<Object, Cache<Query, Value>> notification) {
    Object key = notification.getKey();
    if (key == null) {
        return;
    }

    Cache<Query, Value> valueCache = notification.getValue();
    if (valueCache == null) {
        return;
    }

    for (Value value : valueCache.asMap().values()) {
        listener.onRemoval(value.shardId, value.bitset);
        // if null then this means the shard has already been removed and the stats are 0 anyway for the shard this key belongs to
    }
}

public KeyProviderCache(long expiryMs) {
  cache = CacheBuilder.newBuilder()
      .expireAfterAccess(expiryMs, TimeUnit.MILLISECONDS)
      .removalListener(new RemovalListener<URI, KeyProvider>() {
        @Override
        public void onRemoval(
            RemovalNotification<URI, KeyProvider> notification) {
          try {
            notification.getValue().close();
          } catch (Throwable e) {
            LOG.error(
                "Error closing KeyProvider with uri ["
                    + notification.getKey() + "]", e);
            ;
          }
        }
      })
      .build();
}

/**
 * Create a new KMSAudit.
 *
 * @param windowMs Duplicate events within the aggregation window are quashed
 *                 to reduce log traffic. A single message for aggregated
 *                 events is printed at the end of the window, along with a
 *                 count of the number of aggregated events.
 */
KMSAudit(long windowMs) {
  cache = CacheBuilder.newBuilder()
      .expireAfterWrite(windowMs, TimeUnit.MILLISECONDS)
      .removalListener(
          new RemovalListener<String, AuditEvent>() {
            @Override
            public void onRemoval(
                RemovalNotification<String, AuditEvent> entry) {
              AuditEvent event = entry.getValue();
              if (event.getAccessCount().get() > 0) {
                KMSAudit.this.logEvent(event);
                event.getAccessCount().set(0);
                KMSAudit.this.cache.put(entry.getKey(), event);
              }
            }
          }).build();
  executor = Executors.newScheduledThreadPool(1, new ThreadFactoryBuilder()
      .setDaemon(true).setNameFormat(KMS_LOGGER_NAME + "_thread").build());
  executor.scheduleAtFixedRate(new Runnable() {
    @Override
    public void run() {
      cache.cleanUp();
    }
  }, windowMs / 10, windowMs / 10, TimeUnit.MILLISECONDS);
}

@Override
public void init(DeviceId deviceId, PipelinerContext context) {
    this.serviceDirectory = context.directory();
    this.deviceId = deviceId;

    flowRuleService = serviceDirectory.get(FlowRuleService.class);
    flowObjectiveStore = serviceDirectory.get(FlowObjectiveStore.class);

    pendingNext = CacheBuilder.newBuilder()
            .expireAfterWrite(20, TimeUnit.SECONDS)
            .removalListener((RemovalNotification<Integer, NextObjective> notification) -> {
                if (notification.getCause() == RemovalCause.EXPIRED) {
                    notification.getValue().context()
                            .ifPresent(c -> c.onError(notification.getValue(),
                                    ObjectiveError.FLOWINSTALLATIONFAILED));
                }
            }).build();
}

@Activate
public void activate() {
    providerService = providerRegistry.register(this);

    pendingOperations = CacheBuilder.newBuilder()
            .expireAfterWrite(TIMEOUT, TimeUnit.SECONDS)
            .removalListener((RemovalNotification<Long, MeterOperation> notification) -> {
                if (notification.getCause() == RemovalCause.EXPIRED) {
                    providerService.meterOperationFailed(notification.getValue(),
                                                         MeterFailReason.TIMEOUT);
                }
            }).build();

    controller.addEventListener(listener);
    controller.addListener(listener);

    controller.getSwitches().forEach((sw -> createStatsCollection(sw)));
}

public void onRemoval(RemovalNotification<Integer, StatementInfo> notification) {
  Integer stmtId = notification.getKey();
  StatementInfo doomed = notification.getValue();
  if (doomed == null) {
    // log/throw?
    return;
  }
  LOG.debug("Expiring statement {} because {}", stmtId, notification.getCause());
  try {
    if (doomed.getResultSet() != null) {
      doomed.getResultSet().close();
    }
    if (doomed.statement != null) {
      doomed.statement.close();
    }
  } catch (Throwable t) {
    LOG.info("Exception thrown while expiring statement {}", stmtId, t);
  }
}

private static LoadingCache<DatanodeInfo, DatanodeInfo> initExcludedNodes(
    long excludedNodesCacheExpiry) {
  return CacheBuilder.newBuilder()
      .expireAfterWrite(excludedNodesCacheExpiry, TimeUnit.MILLISECONDS)
      .removalListener(new RemovalListener<DatanodeInfo, DatanodeInfo>() {
        @Override
        public void onRemoval(
            @Nonnull RemovalNotification<DatanodeInfo, DatanodeInfo>
                notification) {
          LOG.info("Removing node " + notification.getKey()
              + " from the excluded nodes list");
        }
      }).build(new CacheLoader<DatanodeInfo, DatanodeInfo>() {
        @Override
        public DatanodeInfo load(DatanodeInfo key) throws Exception {
          return key;
        }
      });
}

/**
 * Create a new KMSAudit.
 *
 * @param windowMs Duplicate events within the aggregation window are quashed
 *                 to reduce log traffic. A single message for aggregated
 *                 events is printed at the end of the window, along with a
 *                 count of the number of aggregated events.
 */
KMSAudit(long windowMs) {
  cache = CacheBuilder.newBuilder()
      .expireAfterWrite(windowMs, TimeUnit.MILLISECONDS)
      .removalListener(
          new RemovalListener<String, AuditEvent>() {
            @Override
            public void onRemoval(
                RemovalNotification<String, AuditEvent> entry) {
              AuditEvent event = entry.getValue();
              if (event.getAccessCount().get() > 0) {
                KMSAudit.this.logEvent(event);
                event.getAccessCount().set(0);
                KMSAudit.this.cache.put(entry.getKey(), event);
              }
            }
          }).build();
  executor = Executors.newScheduledThreadPool(1, new ThreadFactoryBuilder()
      .setDaemon(true).setNameFormat(KMS_LOGGER_NAME + "_thread").build());
  executor.scheduleAtFixedRate(new Runnable() {
    @Override
    public void run() {
      cache.cleanUp();
    }
  }, windowMs / 10, windowMs / 10, TimeUnit.MILLISECONDS);
}

/**
 * Creates a {@link Cache} configured by this instance.
 *
 * @param <T>
 *            the type of the value stored in the Cache
 * @param out
 *            a concurrent {@code Deque} to which the cached values are
 *            added as they are removed from the cache
 * @param ticker
 *            the time source used to determine expiration
 * @return a {@link Cache} corresponding to this instance's values or
 *         {@code null} unless {@code #numEntries} is positive.
 */
@Nullable
public <T> Cache<String, T> createCache(final ConcurrentLinkedDeque<T> out, Ticker ticker) {
  Preconditions.checkNotNull(out, "The out deque cannot be null");
  Preconditions.checkNotNull(ticker, "The ticker cannot be null");
  if (numEntries <= 0) {
    return null;
  }
  final RemovalListener<String, T> listener = new RemovalListener<String, T>() {
    @Override
    public void onRemoval(RemovalNotification<String, T> notification) {
      out.addFirst(notification.getValue());
    }
  };
  CacheBuilder<String, T> b = CacheBuilder.newBuilder().maximumSize(numEntries).ticker(ticker)
      .removalListener(listener);
  if (expirationMillis >= 0) {
    b.expireAfterWrite(expirationMillis, TimeUnit.MILLISECONDS);
  }
  return b.build();
}

/**
 * Creates a {@link Cache} configured by this instance.
 *
 * @param <T>
 *            the type of the value stored in the Cache
 * @param out
 *            a concurrent {@code Deque} to which cached values are added as
 *            they are removed from the cache
 * @param ticker
 *            the time source used to determine expiration
 * @return a {@link Cache} corresponding to this instance's values or
 *         {@code null} unless {@code #numEntries} is positive.
 */
@Nullable
public <T> Cache<String, T> createCache(final ConcurrentLinkedDeque<T> out, Ticker ticker) {
  Preconditions.checkNotNull(out, "The out deque cannot be null");
  Preconditions.checkNotNull(ticker, "The ticker cannot be null");
  if (numEntries <= 0) {
    return null;
  }
  final RemovalListener<String, T> listener = new RemovalListener<String, T>() {
    @Override
    public void onRemoval(RemovalNotification<String, T> notification) {
      out.addFirst(notification.getValue());
    }
  };
  CacheBuilder<String, T> b = CacheBuilder.newBuilder().maximumSize(numEntries).ticker(ticker)
      .removalListener(listener);
  if (flushCacheEntryIntervalMillis >= 0) {
    b.expireAfterWrite(flushCacheEntryIntervalMillis, TimeUnit.MILLISECONDS);
  }
  return b.build();
}

@Override
public void onRemoval(RemovalNotification<K, V> notification)
{
    switch (notification.getCause())
    {
        case EXPIRED:
            notifyListeners(new GuavaCacheEntryEvent<>(this, EventType.EXPIRED, notification));
            break;

        case EXPLICIT:
            notifyListeners(new GuavaCacheEntryEvent<>(this, EventType.REMOVED, notification));
            break;

        case REPLACED:
            notifyListeners(new GuavaCacheEntryEvent<>(this, EventType.UPDATED, notification));
            break;
    }
}

CircuitBreakerRegistryImpl(Vertx vertx, CircuitBreakerRegistryOptions options) {
  this.vertx = vertx;
  this.options = options;
  this.cache = CacheBuilder.newBuilder()
          .expireAfterAccess(options.getCacheExpires(), TimeUnit.SECONDS)
          .removalListener(new RemovalListener<String, CircuitBreaker>() {
            @Override
            public void onRemoval(RemovalNotification<String, CircuitBreaker> notification) {
              Log.create(LOGGER)
                      .setLogType(LogType.LOG)
                      .setModule("CircuitBreaker")
                      .setEvent("cache.removed")
                      .addData("key", notification.getKey())
                      .setMessage("cause by: {}")
                      .addArg(notification.getCause())
                      .info();
            }
          })
          .build(new CacheLoader<String, CircuitBreaker>() {
            @Override
            public CircuitBreaker load(String circuitBreakerName) throws Exception {
              return create(circuitBreakerName);
            }
          });
}

public PlansCache(long maxBytes) {

        LOG.log(Level.INFO, "Max query plan cache size: {0} bytes", maxBytes + "");

        this.cache = CacheBuilder
            .newBuilder()
            .recordStats()
            .weigher((String sql, ExecutionPlanContainer plan) -> {
                return plan.weight;
            })
            .maximumWeight(maxBytes)
            .removalListener((RemovalNotification<String, ExecutionPlanContainer> notification) -> {
                LOG.log(Level.FINE, "Removed query {0} -> {1} size {2} bytes", new Object[]{notification.getCause(),
                    notification.getKey(), notification.getValue().weight});
            })
            .build();

    }

/**
 * Constructs ZKDiscoveryService using the provided zookeeper client for storing service registry under namespace.
 * @param zkClient of zookeeper quorum
 * @param namespace under which the service registered would be stored in zookeeper.
 *                  If namespace is {@code null}, no namespace will be used.
 */
public ZKDiscoveryService(ZKClient zkClient, String namespace) {
  this.closed = new AtomicBoolean();
  this.discoverables = HashMultimap.create();
  this.lock = new ReentrantLock();
  this.retryExecutor = Executors.newSingleThreadScheduledExecutor(
    Threads.createDaemonThreadFactory("zk-discovery-retry"));
  this.zkClient = namespace == null ? zkClient : ZKClients.namespace(zkClient, namespace);
  this.services = CacheBuilder.newBuilder()
    .removalListener(new RemovalListener<String, ServiceDiscoveredCacheEntry>() {
      @Override
      public void onRemoval(RemovalNotification<String, ServiceDiscoveredCacheEntry> notification) {
        ServiceDiscoveredCacheEntry entry = notification.getValue();
        if (entry != null) {
          entry.cancel();
        }
      }
    })
    .build(createServiceLoader());
  this.watcherCancellable = this.zkClient.addConnectionWatcher(createConnectionWatcher());
}

public KeyProviderCache(long expiryMs) {
  cache = CacheBuilder.newBuilder()
      .expireAfterAccess(expiryMs, TimeUnit.MILLISECONDS)
      .removalListener(new RemovalListener<URI, KeyProvider>() {
        @Override
        public void onRemoval(
            RemovalNotification<URI, KeyProvider> notification) {
          try {
            notification.getValue().close();
          } catch (Throwable e) {
            LOG.error(
                "Error closing KeyProvider with uri ["
                    + notification.getKey() + "]", e);
            ;
          }
        }
      })
      .build();
}

/**
 * Create a new KMSAudit.
 *
 * @param windowMs Duplicate events within the aggregation window are quashed
 *                 to reduce log traffic. A single message for aggregated
 *                 events is printed at the end of the window, along with a
 *                 count of the number of aggregated events.
 */
KMSAudit(long windowMs) {
  cache = CacheBuilder.newBuilder()
      .expireAfterWrite(windowMs, TimeUnit.MILLISECONDS)
      .removalListener(
          new RemovalListener<String, AuditEvent>() {
            @Override
            public void onRemoval(
                RemovalNotification<String, AuditEvent> entry) {
              AuditEvent event = entry.getValue();
              if (event.getAccessCount().get() > 0) {
                KMSAudit.this.logEvent(event);
                event.getAccessCount().set(0);
                KMSAudit.this.cache.put(entry.getKey(), event);
              }
            }
          }).build();
  executor = Executors.newScheduledThreadPool(1, new ThreadFactoryBuilder()
      .setDaemon(true).setNameFormat(KMS_LOGGER_NAME + "_thread").build());
  executor.scheduleAtFixedRate(new Runnable() {
    @Override
    public void run() {
      cache.cleanUp();
    }
  }, windowMs / 10, windowMs / 10, TimeUnit.MILLISECONDS);
}

public JmxConnectionCache( int pollRateMs ) {
    cache = CacheBuilder.newBuilder().expireAfterAccess( pollRateMs * 3, TimeUnit.MILLISECONDS ).removalListener(
      new RemovalListener<VirtualMachineConnector, JmxConnection>() {
          @Override
          public void onRemoval( RemovalNotification<VirtualMachineConnector, JmxConnection> notification ) {
              try {
                  if( notification.getValue() != null ) {
                      logger.info( "Removing idle connection to {}", notification.getKey() );
                      notification.getValue().getConnector().close();
                  }
              } catch( IOException e ) {
                  logger.error( "Error closing connection to {}", notification.getKey(), e );
              }
          }
      } ).build( new CacheLoader<VirtualMachineConnector, JmxConnection>() {
        @Override
        public JmxConnection load( VirtualMachineConnector key ) throws Exception {
            return key.connect();
        }
    } );
}

@Test
public void testCacheExpire() throws InterruptedException {
    int total = 10;
    final Map<String, String> expired = new HashMap<>();
    RemovalListener removalListener = new RemovalListener<String, String>() {
        @Override
        public void onRemoval(RemovalNotification<String, String> notification) {
            if(RemovalCause.EXPIRED == notification.getCause()) {
                expired.put(notification.getKey(), notification.getValue());
            }
        }
    };
    Cache<String, String> myCache = CacheBuilder.newBuilder()
            .expireAfterWrite(2, TimeUnit.MILLISECONDS)
            .removalListener(removalListener)
            .build();
    for(int i = 0; i < total; i++) {
        myCache.put("key_" + i, "val_" + i);
    }
    Thread.sleep(10);
    myCache.cleanUp();
    assertEquals(total, expired.size());
}

ShardWriterCache() {
  this.cache = CacheBuilder
      .newBuilder()
      .expireAfterWrite(IDLE_TIMEOUT_MS, TimeUnit.MILLISECONDS)
      .removalListener(new RemovalListener<Integer, ShardWriter<K, V>>() {
        @Override
        public void onRemoval(RemovalNotification<Integer, ShardWriter<K, V>> notification) {
          if (notification.getCause() != RemovalCause.EXPLICIT) {
            ShardWriter writer = notification.getValue();
            LOG.info("{} : Closing idle shard writer {} after 1 minute of idle time.",
                     writer.shard, writer.producerName);
            writer.producer.close();
          }
        }
      }).build();

  // run cache.cleanUp() every 10 seconds.
  SCHEDULED_CLEAN_UP_THREAD.scheduleAtFixedRate(
      new Runnable() {
        @Override
        public void run() {
          cache.cleanUp();
        }
      },
      CLEAN_UP_CHECK_INTERVAL_MS, CLEAN_UP_CHECK_INTERVAL_MS, TimeUnit.MILLISECONDS);
}

public KeyProviderCache(long expiryMs) {
  cache = CacheBuilder.newBuilder()
      .expireAfterAccess(expiryMs, TimeUnit.MILLISECONDS)
      .removalListener(new RemovalListener<URI, KeyProvider>() {
        @Override
        public void onRemoval(
            RemovalNotification<URI, KeyProvider> notification) {
          try {
            notification.getValue().close();
          } catch (Throwable e) {
            LOG.error(
                "Error closing KeyProvider with uri ["
                    + notification.getKey() + "]", e);
            ;
          }
        }
      })
      .build();
}

/**
 * Create a new KMSAudit.
 *
 * @param windowMs Duplicate events within the aggregation window are quashed
 *                 to reduce log traffic. A single message for aggregated
 *                 events is printed at the end of the window, along with a
 *                 count of the number of aggregated events.
 */
KMSAudit(long windowMs) {
  cache = CacheBuilder.newBuilder()
      .expireAfterWrite(windowMs, TimeUnit.MILLISECONDS)
      .removalListener(
          new RemovalListener<String, AuditEvent>() {
            @Override
            public void onRemoval(
                RemovalNotification<String, AuditEvent> entry) {
              AuditEvent event = entry.getValue();
              if (event.getAccessCount().get() > 0) {
                KMSAudit.this.logEvent(event);
                event.getAccessCount().set(0);
                KMSAudit.this.cache.put(entry.getKey(), event);
              }
            }
          }).build();
  executor = Executors.newScheduledThreadPool(1, new ThreadFactoryBuilder()
      .setDaemon(true).setNameFormat(KMS_LOGGER_NAME + "_thread").build());
  executor.scheduleAtFixedRate(new Runnable() {
    @Override
    public void run() {
      cache.cleanUp();
    }
  }, windowMs / 10, windowMs / 10, TimeUnit.MILLISECONDS);
}

@Override
public void onRemoval(final RemovalNotification<BufferedIncrement, Amount> entry) {
  final Amount amount = entry.getValue();
  final long raw = amount.getRawAndInvalidate();
  final long delta = Amount.amount(raw);
  if (Amount.numUpdatesLeft(raw) == Loader.MAX_UPDATES) {
    // This amount was never incremented, because the number of updates
    // left is still the original number.  Therefore this is an Amount
    // that has been evicted before anyone could attach any update to
    // it, so the delta must be 0, and we don't need to send this RPC.
    assert delta == 0 : "WTF? Pristine Amount with non-0 delta: " + amount;
    return;
  }
  final BufferedIncrement incr = entry.getKey();
  final AtomicIncrementRequest req =
    new AtomicIncrementRequest(incr.table, incr.key, incr.family,
                               incr.qualifier, delta);
  client.atomicIncrement(req).chain(amount.deferred);
}

@Activate
public void activate() {
    providerService = providerRegistry.register(this);
    controller.addListener(listener);
    controller.addEventListener(listener);

    pendingBatches = CacheBuilder.newBuilder()
            .expireAfterWrite(10, TimeUnit.SECONDS)
            .removalListener((RemovalNotification<Long, InternalCacheEntry> notification) -> {
                if (notification.getCause() == RemovalCause.EXPIRED) {
                    providerService.batchOperationCompleted(notification.getKey(),
                                                            notification.getValue().failedCompletion());
                }
            }).build();


    for (OpenFlowSwitch sw : controller.getSwitches()) {
        FlowStatsCollector fsc = new FlowStatsCollector(sw, POLL_INTERVAL);
        fsc.start();
        collectors.put(new Dpid(sw.getId()), fsc);
    }


    log.info("Started");
}

protected FileCacheService()
{
    cache = CacheBuilder.<String, Queue<RandomAccessReader>>newBuilder()
                        .expireAfterAccess(AFTER_ACCESS_EXPIRATION, TimeUnit.MILLISECONDS)
                        .concurrencyLevel(DatabaseDescriptor.getConcurrentReaders())
                        .removalListener(new RemovalListener<String, Queue<RandomAccessReader>>()
                        {
                            @Override
                            public void onRemoval(RemovalNotification<String, Queue<RandomAccessReader>> notification)
                            {
                                Queue<RandomAccessReader> cachedInstances = notification.getValue();

                                if (cachedInstances == null)
                                    return;

                                for (RandomAccessReader reader : cachedInstances)
                                    reader.deallocate();
                            }
                        })
                        .build();
}

@Override
public void onRemoval(RemovalNotification<K, V> notification) {
    TransactionBoundCacheEntry transactionBoundCacheEntry = transactionBoundCacheEntries.get();
    if (transactionBoundCacheEntry == null) {
        log.error(cacheName + ": Weird. We receive removal event, but no transaction-bound cache exists");
        return;
    }
    if (transactionBoundCacheEntry == markerUseGlobalCache) {
        log.error(cacheName + ": Weird. We received removal event, while trasaction is completting");
        return;
    }

    if (log.isTraceEnabled()) {
        log.trace(cacheName + ": Element is removed from local cache: " + notification.getKey());
    }

    transactionBoundCacheEntry.transactionBoundRemovals.add(notification.getKey());
}

private LoadingCache<String, PrintWriter> buildPrintWriterCache()
{
    return CacheBuilder.newBuilder()
            .maximumSize(20)
            .removalListener((RemovalNotification<String, PrintWriter> rn) -> rn.getValue().close())
            .build(new CacheLoader<String, PrintWriter>()
            {
                @Override
                public PrintWriter load(String fileName)
                        throws Exception
                {
                    File file = new File(fileName);
                    createParentDirs(file);
                    return new PrintWriter(new FileOutputStream(file, true));
                }
            });
}

public CodeAssist() {
  // todo configure expire time
  cache =
      CacheBuilder.newBuilder()
          .expireAfterWrite(15, TimeUnit.MINUTES)
          .removalListener(
              new RemovalListener<String, CodeAssistContext>() {
                @Override
                public void onRemoval(
                    RemovalNotification<String, CodeAssistContext> notification) {
                  if (notification.getValue() != null) {
                    notification.getValue().clean();
                  }
                }
              })
          .build();
}

public RefactoringManager() {
  sessions =
      CacheBuilder.newBuilder()
          .expireAfterAccess(15, TimeUnit.MINUTES)
          .removalListener(
              new RemovalListener<String, RefactoringSession>() {
                @Override
                public void onRemoval(
                    RemovalNotification<String, RefactoringSession> notification) {
                  RefactoringSession value = notification.getValue();
                  if (value != null) {
                    value.dispose();
                  }
                }
              })
          .build();
}

public ExpiringSet(long duration, @Nonnull TimeUnit unit,
                   @Nullable final RemovalListener<E> removalListener) {
  Preconditions.checkNotNull(unit, "unit cannot be null.");
  Preconditions.checkArgument(duration > 0, "duration must be positive: %s %s", duration, unit);

  CacheBuilder<Object, Object> builder = CacheBuilder.newBuilder()
      .expireAfterWrite(duration, unit);

  if (removalListener != null) {
    builder.removalListener(new com.google.common.cache.RemovalListener<E, Object>() {
      @Override public void onRemoval(@Nonnull RemovalNotification<E, Object> notification) {
        removalListener.onRemoval(notification.getKey());
      }
    });
  }

  cache = builder.build();
}

private RemovalListener<Object, Boolean> buildRemovalListener() {
    return new RemovalListener<Object, Boolean>() {
        @Override
        public void onRemoval(RemovalNotification<Object, Boolean> notification) {
            // if there are no more entries in the user cache - delete the data
            // this might be not a very good idea to do it like that
            // beacause we have no syncronization with loading/unloading mechanism
            // TODO: ideally we should be running some RxJava observable,
            //       which should only emit a value where there were no onRemoval() and load()
            //       calls for, say, 500ms.
            if (cache.size() == 0) {
                data = null;
            }
        }
    };
}

public void onRemoval(RemovalNotification<Integer, StatementInfo> notification) {
  Integer stmtId = notification.getKey();
  StatementInfo doomed = notification.getValue();
  if (doomed == null) {
    // log/throw?
    return;
  }
  if (LOG.isDebugEnabled()) {
    LOG.debug("Expiring statement " + stmtId + " because "
        + notification.getCause());
  }
  try {
    if (doomed.resultSet != null) {
      doomed.resultSet.close();
    }
    if (doomed.statement != null) {
      doomed.statement.close();
    }
  } catch (Throwable t) {
    LOG.info("Exception thrown while expiring statement " + stmtId);
  }
}

private RemovalListener<String, DFSClient> clientRemovalListener() {
  return new RemovalListener<String, DFSClient>() {
    @Override
    public void onRemoval(
        RemovalNotification<String, DFSClient> notification) {
      DFSClient client = notification.getValue();
      try {
        client.close();
      } catch (IOException e) {
        LOG.warn(String
            .format("IOException when closing the DFSClient(%s), cause: %s",
                client, e));
      }
    }
  };
}

/**
 * Create a new KMSAudit.
 *
 * @param windowMs Duplicate events within the aggregation window are quashed
 *                 to reduce log traffic. A single message for aggregated
 *                 events is printed at the end of the window, along with a
 *                 count of the number of aggregated events.
 */
KMSAudit(long windowMs) {
  cache = CacheBuilder.newBuilder()
      .expireAfterWrite(windowMs, TimeUnit.MILLISECONDS)
      .removalListener(
          new RemovalListener<String, AuditEvent>() {
            @Override
            public void onRemoval(
                RemovalNotification<String, AuditEvent> entry) {
              AuditEvent event = entry.getValue();
              if (event.getAccessCount().get() > 0) {
                KMSAudit.this.logEvent(event);
                event.getAccessCount().set(0);
                KMSAudit.this.cache.put(entry.getKey(), event);
              }
            }
          }).build();
  executor = Executors.newScheduledThreadPool(1, new ThreadFactoryBuilder()
      .setDaemon(true).setNameFormat(KMS_LOGGER_NAME + "_thread").build());
  executor.scheduleAtFixedRate(new Runnable() {
    @Override
    public void run() {
      cache.cleanUp();
    }
  }, windowMs / 10, windowMs / 10, TimeUnit.MILLISECONDS);
}

private SnapshotManager(KylinConfig config) {
    this.config = config;
    this.snapshotCache = CacheBuilder.newBuilder().removalListener(new RemovalListener<String, SnapshotTable>() {
        @Override
        public void onRemoval(RemovalNotification<String, SnapshotTable> notification) {
            SnapshotManager.logger.info("Snapshot with resource path " + notification.getKey() + " is removed due to " + notification.getCause());
        }
    }).maximumSize(config.getCachedSnapshotMaxEntrySize())//
            .expireAfterWrite(1, TimeUnit.DAYS).build(new CacheLoader<String, SnapshotTable>() {
                @Override
                public SnapshotTable load(String key) throws Exception {
                    SnapshotTable snapshotTable = SnapshotManager.this.load(key, true);
                    return snapshotTable;
                }
            });
}

private DictionaryManager(KylinConfig config) {
    this.config = config;
    this.dictCache = CacheBuilder.newBuilder()//
            .softValues()//
            .removalListener(new RemovalListener<String, DictionaryInfo>() {
                @Override
                public void onRemoval(RemovalNotification<String, DictionaryInfo> notification) {
                    DictionaryManager.logger.info("Dict with resource path " + notification.getKey() + " is removed due to " + notification.getCause());
                }
            })//
            .maximumSize(config.getCachedDictMaxEntrySize())//
            .expireAfterWrite(1, TimeUnit.DAYS).build(new CacheLoader<String, DictionaryInfo>() {
                @Override
                public DictionaryInfo load(String key) throws Exception {
                    DictionaryInfo dictInfo = DictionaryManager.this.load(key, true);
                    if (dictInfo == null) {
                        return NONE_INDICATOR;
                    } else {
                        return dictInfo;
                    }
                }
            });
}