Nifi自定义处理器如何获取redis的db连接并重用连接

Question

我正在使用 Nifi 查询 redis 服务器以获取地理空间数据。但是一旦我停止处理器，我正在查询的密钥就会被删除。

我想在处理器中创建和重用 redis 连接。

但代码在普通的 java 类中运行，如果我运行该代码，密钥不会被删除。

我不明白为什么要删除密钥。下面是 redis 地理空间数据的自定义处理器代码。

/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package com.test.redis;


import org.apache.nifi.components.PropertyDescriptor;
import org.apache.nifi.flowfile.FlowFile;
import org.apache.nifi.annotation.behavior.ReadsAttribute;
import org.apache.nifi.annotation.behavior.ReadsAttributes;
import org.apache.nifi.annotation.behavior.WritesAttribute;
import org.apache.nifi.annotation.behavior.WritesAttributes;
import org.apache.nifi.annotation.lifecycle.OnScheduled;
import org.apache.nifi.annotation.lifecycle.OnStopped;
import org.apache.nifi.annotation.documentation.CapabilityDescription;
import org.apache.nifi.annotation.documentation.SeeAlso;
import org.apache.nifi.annotation.documentation.Tags;
import org.apache.nifi.processor.exception.ProcessException;
import org.apache.nifi.processor.AbstractProcessor;
import org.apache.nifi.processor.ProcessContext;
import org.apache.nifi.processor.ProcessSession;
import org.apache.nifi.processor.ProcessorInitializationContext;
import org.apache.nifi.processor.Relationship;
import org.apache.nifi.processor.util.StandardValidators;

import redis.clients.jedis.GeoRadiusResponse;
import redis.clients.jedis.GeoUnit;
import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisPool;
import redis.clients.jedis.JedisPoolConfig;
import redis.clients.jedis.exceptions.JedisConnectionException;
import redis.clients.jedis.params.GeoRadiusParam;

import java.net.URI;
import java.net.URISyntaxException;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashSet;
import java.util.List;
import java.util.Set;

@Tags({"example"})
@CapabilityDescription("Provide a description")
@SeeAlso({})
@ReadsAttributes({@ReadsAttribute(attribute="", description="")})
@WritesAttributes({@WritesAttribute(attribute="", description="")})
public class IotHubRedis extends AbstractProcessor {

    private volatile Jedis jedisPool;


    public static final PropertyDescriptor ConnectionHost = new PropertyDescriptor
            .Builder().name("ConnectionHost")
            .displayName("ConnectionHost")
            .description("ConnectionHost")
            .required(true)
            .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
            .build();
    public static final PropertyDescriptor ConnectionPort = new PropertyDescriptor
            .Builder().name("ConnectionPort")
            .displayName("ConnectionPort")
            .description("ConnectionPort")
            .required(true)
            .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
            .build();

    public static final PropertyDescriptor Radius = new PropertyDescriptor
            .Builder().name("Radius")
            .displayName("Radius")
            .description("Radius")
            .required(true)
            .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
            .build();


      public static final PropertyDescriptor Lattitude = new PropertyDescriptor
      .Builder().name("Lattitude") .displayName("Lattitude")
      .description("Lattitude") .required(true)
      .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
      .expressionLanguageSupported(true) .build();


      public static final PropertyDescriptor Longitude = new PropertyDescriptor
      .Builder().name("Longitude") .displayName("Longitude")
      .description("Longitude") .required(true)
      .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
      .expressionLanguageSupported(true) .build();


    public static final PropertyDescriptor RedisKey = new PropertyDescriptor
            .Builder().name("RedisKey")
            .displayName("RedisKey")
            .description("RedisKey")
            .required(true)
            .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
            .build();

    public static final Relationship Success = new Relationship.Builder()
            .name("Success")
            .description("Success")
            .build();

    public static final Relationship Failure = new Relationship.Builder()
            .name("Failure")
            .description("Failure")
            .build();

    private List<PropertyDescriptor> descriptors;

    private Set<Relationship> relationships;

    @Override
    protected void init(final ProcessorInitializationContext context) {
        final List<PropertyDescriptor> descriptors = new ArrayList<PropertyDescriptor>();
        descriptors.add(ConnectionHost);
        descriptors.add(ConnectionPort);
        descriptors.add(Radius);
        descriptors.add(Lattitude);
        descriptors.add(Longitude);
        descriptors.add(RedisKey);
        this.descriptors = Collections.unmodifiableList(descriptors);

        final Set<Relationship> relationships = new HashSet<Relationship>();
        relationships.add(Success);
        relationships.add(Failure);
        this.relationships = Collections.unmodifiableSet(relationships);


    }

    @Override
    public Set<Relationship> getRelationships() {
        return this.relationships;
    }

    @Override
    public final List<PropertyDescriptor> getSupportedPropertyDescriptors() {
        return descriptors;
    }

    @OnScheduled
    public void onScheduled(final ProcessContext context) {
        try {
            //jedisPool = new Jedis(context.getProperty("ConnectionHost").toString(), Integer.parseInt(context.getProperty("ConnectionPort").toString()));
            //getLogger().info("----->Redis Connection is Successful"+jedisPool);
            jedisPool = new Jedis("192.168.8.214",6379);
        } catch (Exception e) {
            getLogger().error("Unable to establish Redis connection pool.");
        }

    }

    @OnStopped
    public void closeRedisPool(final ProcessContext context) {
        jedisPool.flushAll();
        jedisPool.close();
    }


    @Override
    public void onTrigger(final ProcessContext context, final ProcessSession session) throws ProcessException {
        FlowFile flowFile = session.get();
        Double lat=Double.parseDouble(context.getProperty("Lattitude").evaluateAttributeExpressions(flowFile).getValue().toString());
        Double lon=Double.parseDouble(context.getProperty("Longitude").evaluateAttributeExpressions(flowFile).getValue().toString());
        String location="NA";
        Double geoRadiusInKm=Double.parseDouble(context.getProperty("Radius").toString());
        String RedisKey=context.getProperty("RedisKey").toString();
        if ( flowFile == null ) {
            return;
        }

        else {
            try {

                if(!(lat==null || lat==null|| lat.equals("") || lon.equals(""))) {
                GeoRadiusParam param = GeoRadiusParam.geoRadiusParam();
                param.withDist().sortAscending().count(1);
                 List<GeoRadiusResponse> georadius2 = jedisPool.georadius(RedisKey.toString().getBytes(),lat, lon, geoRadiusInKm, GeoUnit.KM, param);
                 //flowFile = session.putAttribute(flowFile, "connection1", "------------>"+RedisKey+"=="+lat+"=="+lon+"=="+geoRadiusInKm+"=="+param);
                 if (!georadius2.isEmpty()) {
                    for (GeoRadiusResponse geoRadiusResponse : georadius2) {
                                    location = geoRadiusResponse.getMemberByString();
                    }
                    flowFile = session.putAttribute(flowFile, "msg", "------------>if block inside try");
                    flowFile = session.putAttribute(flowFile, "location", location);
                    session.transfer(flowFile, Success);
                }
                else {
                    flowFile = session.putAttribute(flowFile, "msg", "------------>else georadius2 is empty");
                    flowFile = session.putAttribute(flowFile, "location", location);
                    session.transfer(flowFile, Success);
                }
            }
                else {
                    flowFile = session.putAttribute(flowFile, "msg", "------------>lat lon null or empty");
                    flowFile = session.putAttribute(flowFile, "location", location);
                    session.transfer(flowFile, Success);
                }


        }
            catch (JedisConnectionException e) { 
                session.transfer(flowFile, Success);

            }
        // TODO implement
    }

}
}

Answer 1

在@OnStopped

你打电话给https://redis.io/commands/flushall

删除所有现有数据库的所有键...

所以，只需删除此命令

Answer 2

我知道这有点旧，但在你的代码中你有：

FlowFile flowFile = session.get();
// Doing something with flowfile
if ( flowFile == null ) {
    return;
}

你明白当 flowFile 为空时这可能/将会爆炸吗？