我正在尝试创建一个持久的 ZNode 并存储我已处理的特定文件的行数。创建工作正常,从节点读取数据也是如此,但如果在相同的代码中,删除则不起作用。我会解释我的意思。
我创建了函数:
setOrCreateFileCheckpoint(fileName: String, lineNumber: Int):- 检查 ZNode 是否存在,如果不存在则创建它,并将存储的值设置为 lineNumber getFileCheckpoint(fileName: String):- 返回存储在 ZNode 中的值 deleteFileCheckpoint(fileName: String):- 删除 ZNode
下面是所有三个的代码:
/*
updates or creates a checkpoint for a file being processed
*/
def setOrCreateFileCheckpoint(fileName: String, lineNumber: Int): Unit =
{
val fileCheckpointPath = checkpointPoolPath + "/" +fileName
val zk = getZookeeper
val zkCuratorClient = getZookeeperCuratorClient
if ( zk.exists(fileCheckpointPath, false) == null)
{
val node = new PersistentNode(zkCuratorClient, CreateMode.PERSISTENT, false, fileCheckpointPath, lineNumber.toString.getBytes())
node.start()
}
else
zk.setData(fileCheckpointPath, lineNumber.toString.getBytes(), -1)
}
/*
gets checkpoint for a file
*/
def getFileCheckpoint(fileName: String): Int =
{
val fileCheckpointPath = checkpointPoolPath + "/" +fileName
val zk = getZookeeper
val zkCuratorClient = getZookeeperCuratorClient
if ( zk.exists(fileCheckpointPath, false) != null)
new String(zk.getData(fileCheckpointPath, false, null)).toInt
else
0
}
/*
deletes the file checkpoint so that we don't keep accumulating zNodes on the zookeeper
*/
def deleteFileCheckpoint(fileName: String): Unit =
{
val fileCheckpointPath = checkpointPoolPath + "/" +fileName
val zk = getZookeeper
if ( zk.exists(fileCheckpointPath, false) == null)
{
throw RuntimeException("Trying to delete checkpoint that doesn't exist for file: " + fileName)
}
else
{
/*println(zk.exists(fileCheckpointPath, false).getVersion)
zk.delete(fileCheckpointPath, zk.exists(fileCheckpointPath, false).getVersion)*/
deleteChildren(zk, fileCheckpointPath, true)
}
}
以下是我正在测试且感到困惑的代码:
ZookeeperUtility.setOrCreateFileCheckpoint("file1", 2000) //let's call it cre1
println(ZookeeperUtility.getFileCheckpoint("file1")) //let's call it get1
ZookeeperUtility.deleteFileCheckpoint("file1") //let's call it del1
println("del1")
ZookeeperUtility.deleteFileCheckpoint("file1") //let's call in del2
println("del2")
运行 1:
第1步:我运行上面显示的代码
结果:del2 上遇到错误
第二步:注释掉cre1并再次运行代码
结果:已获取节点,并给出正确的值作为 del2 上遇到的结果错误。这真是令人难以置信。我不明白为什么。该节点应该被删除。
第三步:cre1仍被注释,与上一步相同,再次运行代码
结果:节点不存在在 get1 处给出 0,这意味着节点不存在。 del1 处遇到错误。这是第 2 步本身应该发生的事情
运行2:
Step1:注释掉del2,运行代码
结果:创建节点,获取正确数据,正常退出
Step2:注释掉cre1,运行代码
结果:从应该删除的节点中获取值 2000。正常退出
第三步:再次运行与第二步相同的代码
结果:获取 0,del1 上遇到错误。
如果我一次一步运行代码,如果我只在一次运行中创建,只在下一次运行中获取,并且只在之后的运行中删除,那么一切都会正常工作。我正处于拔头发的边缘。
附注代码是用 Scala 编写的,但我使用的是 Java API。 Scala 可以无缝地使用 Java 类。
如果你看看我注释掉了一部分的deleteFileCheckpoint函数,我也尝试过这种方法。它具有完全相同的行为。
最佳答案
This is mind boggling. I can't understand why. The node is supposed to be deleted.
我不知道你为什么感到惊讶。您正在创建一个PersistentNode
,它的存在是为了在节点被删除时自动重新创建该节点。事实上,所有周围的代码都非常令人费解。它复制了 PersistentNode 内部的功能。您不需要做所有其他事情。只需使用PersistentNode
即可。
此外,像 checkExists() 这样的代码,后跟基于结果的操作几乎永远不会在生产中工作。 ZooKeeper 是高度并发且最终一致的。这就是为什么您应该始终使用 Curator 的食谱而不是手动编码解决方案。
关于Java Zookeeper API 奇怪的 ZNode 行为。无法正确删除ZNode。它有意想不到的结果,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/61377683/