问题

我正在使用Nifi，尝试创建一个Apache Nifi处理器，但它没有响应！处理器已构建，在处理器列表中显示并且工作，但没有任何输入！！它没有显示任何内容。<br>
以下是我的代码：

@Override
public void onTrigger(final ProcessContext context, final ProcessSession session) throws ProcessException {
    final AtomicReference<String> value = new AtomicReference<>();
    Integer Somme = new Integer(1254 + 5520);

    FlowFile flowFile = session.get();
    session.read(flowFile, new InputStreamCallback() {
        @Override
        public void process(InputStream in) throws IOException {
            HttpGet request = new HttpGet(URL);

            HttpHost target = new HttpHost(localhost, 8080, "http");
            CredentialsProvider provider = new BasicCredentialsProvider();
            provider.setCredentials(
                new AuthScope(target.getHostName(), target.getPort()),
                new UsernamePasswordCredentials(Username, paswd)
            );

            AuthCache authCache = new BasicAuthCache();
            authCache.put(target, new BasicScheme());

            HttpClientContext localContext = HttpClientContext.create();
            localContext.setAuthCache(authCache);

            try (CloseableHttpClient httpClient = HttpClientBuilder.create()
                    .setDefaultCredentialsProvider(provider)
                    .build();
                CloseableHttpResponse response = httpClient.execute(target, request, localContext)) {

                // 401 if wrong user/password
                System.out.println(response.getStatusLine().getStatusCode());

                HttpEntity entity = response.getEntity();
                if (entity != null) {
                    // return it as a String
                    String result = EntityUtils.toString(entity);
                    System.out.println(result.getBytes());
                    value.set(result);
                }
            }
        }
    });

    // Write the results to an attribute
    String results = value.get();
    if (results != null && !results.isEmpty()) {
        flowFile = session.putAttribute(flowFile, "match", results);
    }

    // To write the results back out ot flow file
    flowFile = session.write(flowFile, new OutputStreamCallback() {
        @Override
        public void process(OutputStream out) throws IOException {
            out.write(value.get().getBytes());
        }
    });

    session.transfer(flowFile, Success);
}

任何帮助将不胜感激！已经过了几天，但似乎一切都不对

    @Override
public void onTrigger(final ProcessContext context, final ProcessSession session) throws ProcessException {
final AtomicReference<String> value = new AtomicReference<>();	  
Integer Somme=new Integer(1254+5520);
FlowFile flowFile = session.get();
session.read(flowFile, new InputStreamCallback() {
@Override
public void process(InputStream in) throws IOException {
HttpGet request = new HttpGet(URL);
HttpHost target = new HttpHost(localhost, 8080, "http");
CredentialsProvider provider = new BasicCredentialsProvider();
provider.setCredentials(
new AuthScope(target.getHostName(), target.getPort()),
new UsernamePasswordCredentials(Username, paswd)
);
AuthCache authCache = new BasicAuthCache();
authCache.put(target, new BasicScheme());
HttpClientContext localContext = HttpClientContext.create();
localContext.setAuthCache(authCache);
try (CloseableHttpClient httpClient = HttpClientBuilder.create()
.setDefaultCredentialsProvider(provider)
.build();
CloseableHttpResponse response = httpClient.execute(target, request, localContext)) {
// 401 if wrong user/password
System.out.println(response.getStatusLine().getStatusCode());
HttpEntity entity = response.getEntity();
if (entity != null) {
// return it as a String
String result = EntityUtils.toString(entity);
System.out.println(result.getBytes());
value.set(result);
}
}
}
});
// Write the results to an attribute
String results = value.get();
if(results != null && !results.isEmpty()){
flowFile = session.putAttribute(flowFile, "match", results);
}
// To write the results back out ot flow file
flowFile = session.write(flowFile, new OutputStreamCallback() {
@Override
public void process(OutputStream out) throws IOException {
out.write(value.get().getBytes());
}
});
session.transfer(flowFile, Success);
}
}

答案1

得分: 1

当实现源处理器（表示没有输入）时，您希望使用session.create()来创建全新的流文件。使用session.get()将尝试从传入的队列中获取流文件，但由于没有队列，它将始终为空。