当前位置：首页 > news >正文

NIFI 使用HTTP 作为数据源接收数据

news 2026/3/27 3:25:26

NIFI 使用HTTP 作为数据源接收数据

1.概述

在NIFI 中，可以 ListenHTTP 组件启动一个HTTP服务，通过HTTP 服务接收客户端发送的信息，后续可以增加处理器，对请求进行处理。
我做了一个示例

通过 ListenHTTP 接收信息
通过 ExecuteGroovyScript 对数据进行处理
通过 PutFile 对数据进行验证，查看数据是否有效

2.配置

2.1 配置ListenHTTP 组件

配置后，我们就可以通过将数据 post 到 http://ip:8080/data 上进行数据接收

2.2 ExecuteGroovyScript 处理脚本

import groovy.json.JsonSlurper
import com.fasterxml.jackson.databind.ObjectMapper
import com.fasterxml.jackson.databind.json.JsonMapper
import com.fasterxml.jackson.core.JsonGenerator
import org.apache.commons.io.IOUtils
import java.nio.charset.StandardCharsetsdef flowFile = session.get()
if (!flowFile) returntry {def text = ''session.read(flowFile, { inputStream ->text = IOUtils.toString(inputStream, StandardCharsets.UTF_8)} as InputStreamCallback)// 用 Groovy 解析 JSON（支持宽松语法）def jsonSlurper = new JsonSlurper()def data = jsonSlurper.parseText(text)data['age'] = 20// 使用 Jackson 序列化，不转义非 ASCII 字符（保留中文）def mapper = new ObjectMapper()mapper.getFactory().setCodec(mapper)// 关键：禁用 Unicode 转义def writer = mapper.writer().without(JsonGenerator.Feature.ESCAPE_NON_ASCII)def updatedJson = writer.writeValueAsString(data)flowFile = session.write(flowFile, { outputStream ->outputStream.write(updatedJson.getBytes(StandardCharsets.UTF_8))} as OutputStreamCallback)flowFile = session.putAttribute(flowFile, "mime.type", "application/json; charset=utf-8")session.transfer(flowFile, REL_SUCCESS)} catch (Exception e) {log.error("处理FlowFile时发生错误", e)session.transfer(flowFile, REL_FAILURE)
}

这个代码的作用是读取上游的数据，并通过脚本修改这个数据，再投放给下游。