在Apache NiFi中,可以使用Wait/Notify模式来实现暂停流程以重新验证令牌。以下是一个示例解决方法:
Wait处理器,将其配置为等待某个条件满足。可以使用UpdateAttribute处理器来设置条件,例如在属性token_valid上设置一个布尔值来指示令牌是否有效。Wait处理器的属性中,将Signal Counter Name设置为一个自定义的名称,例如token_validation。Notify处理器,将其配置为发送一个通知信号以解除等待。在Signal Counter Name属性中,设置为与Wait处理器中的相同名称(即token_validation)。RouteOnAttribute处理器,将其配置为根据令牌的有效性进行分流。
token_valid属性等于true的条件,将流程路由到继续处理令牌的处理器。token_valid属性等于false的条件,将流程路由到重新验证令牌的处理器。def token = flowFile.getAttribute('token')
def isValid = // 自定义验证逻辑,返回布尔值
if (isValid) {
// 令牌有效,将属性 token_valid 设置为 true
flowFile = session.putAttribute(flowFile, 'token_valid', 'true')
} else {
// 令牌无效,将属性 token_valid 设置为 false
flowFile = session.putAttribute(flowFile, 'token_valid', 'false')
}
session.transfer(flowFile, REL_SUCCESS)
在上述代码中,首先从流文件的属性中获取令牌值。然后,根据自定义的验证逻辑,确定令牌是否有效。根据验证结果,将属性token_valid设置为true或false。最后,使用session.transfer方法将流文件转移到下一个处理器。
使用上述解决方法,在流程中的任何位置都可以将流程暂停,等待令牌验证后再继续处理。