Java 类org.apache.curator.framework.recipes.cache.NodeCacheListener 实例源码

/**
 * @param cache    NodeCache
 * @param zkListen
 * @throws Exception
 * @Created 2016/9/20
 */
private static void runWatch(final NodeCache cache, final ZookeeperProcessListen zkListen)
        throws Exception {
    cache.getListenable().addListener(new NodeCacheListener() {

        @Override
        public void nodeChanged() {
            LOGGER.info("ZktoxmlMain runWatch  process path  event start ");
            String notPath = cache.getCurrentData().getPath();
            LOGGER.info("NodeCache changed, path is: " + notPath);
            // notify
            zkListen.notify(notPath);
            LOGGER.info("ZktoxmlMain runWatch  process path  event over");
        }
    });
}

@Test
public void test_zkNode() throws Exception {
    CuratorFramework zk = curator.usingNamespace("namespace-test");
    String groupPath = "/group-1/localhost:9001";
    String s = zk.create().creatingParentsIfNeeded().forPath(groupPath);
    Assert.assertEquals(s, "/group-1/localhost:9001");
    NodeCache nodeCache = new NodeCache(zk, "/group-1", true);
    nodeCache.start();
    nodeCache.getListenable().addListener(new NodeCacheListener() {

        @Override
        public void nodeChanged() throws Exception {
            logger.info("node cache change");
        }
    });
    zk.setData().forPath("/group-1", "test-0".getBytes());
    zk.setData().forPath("/group-1", "test-2".getBytes());
}

@Override
public void start() throws Exception {
    // Create the zookeeper node
    createNode();
    // Initial data load (avoid race conditions w/"NodeCache.start(true)")
    updateFromZkBytes(_curator.getData().forPath(_zkPath), _defaultValue);

    // Re-load the data and watch for changes.
    _nodeCache.getListenable().addListener(new NodeCacheListener() {
        @Override
        public void nodeChanged() throws Exception {
            ChildData childData = _nodeCache.getCurrentData();
            if (childData != null) {
                updateFromZkBytes(childData.getData(), _defaultValue);
            }
        }
    });
    _nodeCache.start();
}

public void start() throws Exception {
    started.set(true);
    if (curatorFramework.checkExists().forPath(path) == null) {
        try {
            curatorFramework.create().creatingParentsIfNeeded().forPath(path, "{}".getBytes());
        } catch (KeeperException.NodeExistsException ex) {
            log.info("Node exists on create, continuing");
        }
    }

    this.nodeCacheListener = new NodeCacheListener() {
        @Override
        public void nodeChanged() throws Exception {
            notifyListeners(get());
        }
    };

    this.nodeCache.getListenable().addListener(nodeCacheListener);
    this.nodeCache.start(true);
}

public static void main(String[] args) throws Exception {
    CuratorFramework framework = CuratorFrameworkFactory.builder()
            .connectString("localhost:2181").retryPolicy(new RetryNTimes(3, 2000)).build();
    try {
        framework.start();
        final NodeCache nodeCache = new NodeCache(framework, "/test");
        nodeCache.getListenable().addListener(new NodeCacheListener() {
            @Override
            public void nodeChanged() throws Exception {
                ChildData data = nodeCache.getCurrentData();
                System.out.println(new String(data.getData()) + " / " + data);
            }
        });
        nodeCache.start();

        Thread.sleep(30000);

        nodeCache.close();
    } finally {
        framework.close();
    }
}

/**
 * Start.
 *
 * @throws Exception the exception
 */
public void start() throws Exception { //NOSONAR
  LOG.info("Starting node tracker");
  zkClient.getUnhandledErrorListenable().addListener(errorsListener);
  if (createZkNode()) {
    controlCache = new NodeCache(zkClient, CONTROL_SERVER_NODE_PATH);
    controlCache.getListenable().addListener(new NodeCacheListener() {

      @Override
      public void nodeChanged() throws Exception {
        ChildData currentData = controlCache.getCurrentData();
        if (currentData == null) {
          LOG.warn("Control service node died!");
          onNoMaster();
        } else {
          LOG.warn("Control service node changed!");
          onMasterChange(currentData);
        }
      }
    });
    controlCache.start();
  } else {
    LOG.warn("Failed to create ZK node!");
  }
}

@Test
public void createValue_TriggersListenerOnPropagate() throws Exception {
    String value = "foo";

    final CountDownLatch latch = new CountDownLatch(1);

    container.addListener(new NodeCacheListener() {
        @Override
        public void nodeChanged() throws Exception {
            latch.countDown();
        }
    });

    framework.create().creatingParentsIfNeeded().forPath("/dev/test", value.getBytes(Charsets.UTF_8));

    latch.await();

    assertEquals(value, container.get());
}

@Override
protected void runOneIteration() {
    String newValue = PropertyReader.getProperty(propOverride);

    if (!Objects.equal(newValue, lastValue)) {
        for (final Map.Entry<NodeCacheListener, Executor> o : listeners.entrySet()) {
            o.getValue().execute(new Runnable() {
                @Override
                public void run() {
                    try {
                        o.getKey().nodeChanged();
                    } catch (Exception ex) {
                        LOG.warn("Exception when processing node change.", ex);
                    }
                }
            });
        }
    }
}

@Override
public void start() throws Exception {
    if (!isStarted.compareAndSet(false, true)) {
        throw new RuntimeException("Service already started!");
    }
    if (framework.getState() != CuratorFrameworkState.STARTED) {
        throw new RuntimeException("CuratorFramework is not started!");
    }
    nodeCache.getListenable().addListener(new NodeCacheListener() {
        @Override
        public void nodeChanged() throws Exception {
            build();
        }
    });
    nodeCache.start(true);
    build();
}

private void registerListener(final CuratorFramework curator, final String path, final ZookeeperConfigurationListener listener) throws Exception
{
    final NodeCache cache   = new NodeCache(curator, path, true);

    cache.start();
    cache.getListenable().addListener(new NodeCacheListener()
    {
        @Override
        public void nodeChanged() throws Exception
        {
            listener.onChange(curator, cache, path);
        }
    });

    childrens.add(cache);
    logger.info(String.format("Add listener %s for %s", listener, path));
}

@Override
protected void serviceStart() throws Exception {
    framework.start();
    nodeCache = new NodeCache(framework, zkMountTablePath, false);
    nodeCache.getListenable().addListener(new NodeCacheListener() {
        @Override
        public void nodeChanged() throws Exception {
            handleMountTableChange(nodeCache.getCurrentData().getData());
        }
    });
    nodeCache.start(false);
}

/**
 * Listen for a single node
 *
 * @param path
 * @param listener
 * @throws Exception
 */
public void listenForZNode(String path, ZookeeperEventListener listener) throws Exception {
    init();
    // TODO: simplify code of this method to:
    // ZNodeListener nodeListener = new ZNodeListener(path, listener, client);

    NodeCache cache = new NodeCache(client, path);
    cache.start();

    NodeCacheListener cacheListener = new ZNodeListener(
            listener.getHandler(), cache);

    cache.getListenable().addListener(cacheListener);
    cachePool.add(cache);
}

public void forcReadListenForZNode(String path, ZookeeperEventListener listener) throws Exception {
    init();
    NodeCache cache = new NodeCache(client, path);
    cache.start();

    NodeCacheListener cacheListener = new ZNodeListener(
            listener.getHandler(), cache);

    cache.getListenable().addListener(cacheListener);
    cacheListener.nodeChanged();
    cachePool.add(cache);
}

private void _initCacheWatcher() {
    cacheNodeWatcher = CacheBuilder.newBuilder()
            .concurrencyLevel(Runtime.getRuntime().availableProcessors()).maximumSize(10000)
            .expireAfterAccess(3600, TimeUnit.SECONDS)
            .removalListener(new RemovalListener<String, NodeCache>() {
                @Override
                public void onRemoval(RemovalNotification<String, NodeCache> event) {
                    try {
                        event.getValue().close();
                    } catch (IOException e) {
                        LOGGER.warn(e.getMessage(), e);
                    }
                }
            }).build(new CacheLoader<String, NodeCache>() {
                @Override
                public NodeCache load(final String path) throws Exception {
                    final NodeCache nodeCache = new NodeCache(curatorFramework, path);
                    nodeCache.getListenable().addListener(new NodeCacheListener() {
                        @Override
                        public void nodeChanged() throws Exception {
                            ChildData data = nodeCache.getCurrentData();
                            _invalidateCache(path, data != null ? data.getData() : null);
                        }
                    });
                    nodeCache.start();
                    return nodeCache;
                }
            });
}

public void registerUpdater(ConfigurationUpdater updater) {
  NodeCache cache = getOrCreateNodeCache(updater.getPath());

  cache
      .getListenable()
      .addListener(
          new NodeCacheListener() {
            @Override
            public void nodeChanged() {
              updater.update(cache.getCurrentData().getData());
            }
          });
}

private void tryAddListener(NodeCache cache) {
    if (!hasAddListener) {
        NodeCacheListener nodeCacheListener = () -> {
            T oldResource;
            synchronized (lock) {
                ChildData data = cache.getCurrentData();
                oldResource = resource;
                if (data != null && data.getData() != null) {
                    zkNodeExists = EXISTS;
                    ListenableFuture<T> future = refreshFactory.apply(data.getData(),
                            data.getStat());
                    addCallback(future, new FutureCallback<T>() {

                        @Override
                        public void onSuccess(@Nullable T result) {
                            resource = result;
                            cleanup(resource, oldResource, cache);
                        }

                        @Override
                        public void onFailure(Throwable t) {
                            factoryFailedListener.accept(t);
                            logger.error("", t);
                        }
                    }, directExecutor());
                } else {
                    zkNodeExists = NOT_EXISTS;
                    resource = null;
                    emptyLogged = false;
                    cleanup(resource, oldResource, cache);
                }
            }
        };
        cache.getListenable().addListener(nodeCacheListener);
        nodeCacheRemoveListener = () -> cache.getListenable().removeListener(nodeCacheListener);
        hasAddListener = true;
    }
}

private static void addListener(final NodeCache cache) {
    // a PathChildrenCacheListener is optional. Here, it's used just to log
    // changes
    NodeCacheListener listener = new NodeCacheListener() {

        @Override
        public void nodeChanged() throws Exception {
            if (cache.getCurrentData() != null)
                System.out.println("Node changed: " + cache.getCurrentData().getPath() + ", value: " + new String(cache.getCurrentData().getData()));
        }
    };
    cache.getListenable().addListener(listener);
}

@Override
public void init() {

    final NodeCache node = _configuration.getNode();
    if (node != null) {
        node.getListenable().addListener(new NodeCacheListener() {
            @Override
            public void nodeChanged() throws Exception {
                _configuration.reload();
            }
        });
    }
}

/**
 * Start watching the fluo app uuid. If it changes or goes away then halt the process.
 */
public static NodeCache startAppIdWatcher(Environment env) {
  try {
    CuratorFramework curator = env.getSharedResources().getCurator();

    byte[] uuidBytes = curator.getData().forPath(ZookeeperPath.CONFIG_FLUO_APPLICATION_ID);
    if (uuidBytes == null) {
      Halt.halt("Fluo Application UUID not found");
      throw new RuntimeException(); // make findbugs happy
    }

    final String uuid = new String(uuidBytes, StandardCharsets.UTF_8);

    final NodeCache nodeCache = new NodeCache(curator, ZookeeperPath.CONFIG_FLUO_APPLICATION_ID);
    nodeCache.getListenable().addListener(new NodeCacheListener() {
      @Override
      public void nodeChanged() throws Exception {
        ChildData node = nodeCache.getCurrentData();
        if (node == null || !uuid.equals(new String(node.getData(), StandardCharsets.UTF_8))) {
          Halt.halt("Fluo Application UUID has changed or disappeared");
        }
      }
    });
    nodeCache.start();
    return nodeCache;
  } catch (Exception e) {
    throw new RuntimeException(e);
  }
}

@Override
public void addListener(NodeCacheListener listener) {
    if (useCache || useCacheWhenNotConnectedToDataSource) {
        cache.getListenable().addListener(listener);
    }
}

@BeforeMethod(alwaysRun = true, timeOut = 30_000)
public void setup() throws Exception {
    // Get the zkConnection string from minicluster
    String zkConnection = "localhost:" + hBaseUtils.getZkCluster().getClientPort();

    zkClient = provideInitializedZookeeperClient(zkConnection);

    // Synchronize TSO start
    barrierTillTSOAddressPublication = new CountDownLatch(1);
    final NodeCache currentTSOZNode = new NodeCache(zkClient, CURRENT_TSO_PATH);
    currentTSOZNode.getListenable().addListener(new NodeCacheListener() {

        @Override
        public void nodeChanged() throws Exception {
            byte[] currentTSOAndEpochAsBytes = currentTSOZNode.getCurrentData().getData();
            String currentTSOAndEpoch = new String(currentTSOAndEpochAsBytes, Charsets.UTF_8);
            if (currentTSOAndEpoch.endsWith("#0")) { // Wait till a TSO instance publishes the epoch
                barrierTillTSOAddressPublication.countDown();
            }
        }

    });
    currentTSOZNode.start(true);

    // Configure TSO 1
    TSOServerConfig config1 = new TSOServerConfig();
    config1.setPort(TSO1_PORT);
    config1.setConflictMapSize(1000);
    config1.setLeaseModule(new TestHALeaseManagementModule(TEST_LEASE_PERIOD_MS, TSO_LEASE_PATH, CURRENT_TSO_PATH, zkConnection, NAMESPACE));
    Injector injector1 = Guice.createInjector(new TestTSOModule(hbaseConf, config1));
    LOG.info("===================== Starting TSO 1 =====================");
    tso1 = injector1.getInstance(TSOServer.class);
    leaseManager1 = (PausableLeaseManager) injector1.getInstance(LeaseManagement.class);
    tso1.startAndWait();
    TestUtils.waitForSocketListening("localhost", TSO1_PORT, 100);
    LOG.info("================ Finished loading TSO 1 ==================");

    // Configure TSO 2
    TSOServerConfig config2 = new TSOServerConfig();
    config2.setPort(TSO2_PORT);
    config2.setConflictMapSize(1000);
    config2.setLeaseModule(new TestHALeaseManagementModule(TEST_LEASE_PERIOD_MS, TSO_LEASE_PATH, CURRENT_TSO_PATH, zkConnection, NAMESPACE));
    Injector injector2 = Guice.createInjector(new TestTSOModule(hbaseConf, config2));
    LOG.info("===================== Starting TSO 2 =====================");
    tso2 = injector2.getInstance(TSOServer.class);
    injector2.getInstance(LeaseManagement.class);
    tso2.startAndWait();
    // Don't do this here: TestUtils.waitForSocketListening("localhost", 4321, 100);
    LOG.info("================ Finished loading TSO 2 ==================");

    // Wait till the master TSO is up
    barrierTillTSOAddressPublication.await();
    currentTSOZNode.close();

    // Configure HBase TM
    LOG.info("===================== Starting TM =====================");
    HBaseOmidClientConfiguration hbaseOmidClientConf = new HBaseOmidClientConfiguration();
    hbaseOmidClientConf.setConnectionType(HA);
    hbaseOmidClientConf.setConnectionString(zkConnection);
    hbaseOmidClientConf.getOmidClientConfiguration().setZkCurrentTsoPath(CURRENT_TSO_PATH);
    hbaseOmidClientConf.getOmidClientConfiguration().setZkNamespace(NAMESPACE);
    hbaseOmidClientConf.setHBaseConfiguration(hbaseConf);
    hbaseConf.setInt(HBASE_CLIENT_RETRIES_NUMBER, 3);
    tm = HBaseTransactionManager.builder(hbaseOmidClientConf).build();
    LOG.info("===================== TM Started =========================");
}

public void add(Path path, NodeCacheListener listener) {
    fileListeners.put(path, listener);
}

@Override
public void addListener(NodeCacheListener listener) {
    listeners.add(path, listener);
}

@Override
public void addListener(NodeCacheListener listener) {
    wrapped.getListenable().addListener(listener);

}

@Override
public void addListener(final NodeCacheListener listener) {
    this.addListener(listener, MoreExecutors.directExecutor());
}

@Override
public void addListener(final NodeCacheListener listener, final Executor executor) {
    listeners.put(listener, executor);
}

public ListenerContainer<NodeCacheListener> getListenable() {
    return delegate().getListenable();
}

@Override
public void addListener(final NodeCacheListener listener) {
    this.cache.getListenable().addListener(checkNotNull(listener));
}

@Override
public void addListener(final NodeCacheListener listener, final Executor executor) {
    this.cache.getListenable().addListener(checkNotNull(listener), checkNotNull(executor));
}

@Test
public void addListenerWithExecutor_NullListener_ThrowsNPE() {
    thrown.expect(NullPointerException.class);

    container.addListener((NodeCacheListener) null, executor);
}

/**
 * Add a listener to the NodeCache.
 */
void addListener(NodeCacheListener listener);

/**
 * Add a listener to the NodeCache to be run with the given executor.
 */
void addListener(NodeCacheListener listener, Executor executor);

void addListener(NodeCacheListener listener);

void addListener(NodeCacheListener listener);