清理(删除）pika中的数据

本功能使用shell脚本实现，代码分为三个部分

1）java代码

2）redis连接池

3）shell脚本

1）删除代码

----------------------------------------------------------------------------------------

import com.xes.bdc.galaxy.util.jedis.RedisPoolFactory;
import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisPool;
import redis.clients.jedis.ScanParams;
import redis.clients.jedis.ScanResult;

import java.io.BufferedWriter;
import java.io.File;
import java.io.FileWriter;
import java.io.IOException;
import java.util.List;

/**
 * @author: create by maoxiangyi
 * @version: v1.0
 * @date:2019/7/4
 */
public class DeleteKey {
   public static void main(String[] args) throws InterruptedException, IOException {
      String hostname = args[0];
      String port = args[1];
      String condition = args[2];
      Long count = Long.parseLong(args[3]);

      String isDelete = args[4];
      String outputPath = args[5];

      Long cursor = 0l;

      JedisPool pool = RedisPoolFactory.getPool(hostname, Integer.parseInt(port));
      Jedis resource = pool.getResource();
      BufferedWriter bufferedWriter = null;
      if (!"delete".equalsIgnoreCase(isDelete)) {
         bufferedWriter = new BufferedWriter(new FileWriter(new File(outputPath)));
      }
      do {
         ScanParams scanParams = new ScanParams();
         scanParams.match(condition);
         scanParams.count(count.intValue());
         ScanResult<String> scan = resource.scan(cursor + "", scanParams);
         List<String> result = scan.getResult();
         for (String key : result) {
            if (bufferedWriter != null) {
               System.out.println("写入数据到文件:" + key);
               bufferedWriter.write(key);
               bufferedWriter.newLine();
               bufferedWriter.flush();
            } else {
               System.out.println("key即将被删除:" + key);
               resource.del(key);
            }
         }
         cursor = Long.parseLong(scan.getStringCursor());
      } while (cursor != 0);
      resource.close();
      if (bufferedWriter!=null) {
         bufferedWriter.flush();
         bufferedWriter.close();
      }
   }
}

Redis连接池管理

------------------------------------------------------------------------------

import redis.clients.jedis.JedisPool;
import redis.clients.jedis.JedisPoolConfig;

import java.util.HashMap;

/**
 * @author: create by maoxiangyi
 * @version: v1.0
 * @description: com.xes.bdc.galaxy.factory
 * @date:2019/5/5
 */
public class RedisPoolFactory {
   private static HashMap<String, JedisPool> poolFactory = new HashMap<String, JedisPool>();

   public static JedisPool getPool(String hostname, int port) {
      String key = hostname + port;
      if (!poolFactory.containsKey(key)) {
         synchronized (RedisPoolFactory.class) {
            if (!poolFactory.containsKey(key)) {
               JedisPoolConfig jedisPoolConfig = new JedisPoolConfig();
               //资源池中最大连接数
               jedisPoolConfig.setMaxTotal(200);
               //资源池允许最大空闲的连接数
               jedisPoolConfig.setMaxIdle(30);
               //默认值是-1 表示永不超时 不建议使用
               jedisPoolConfig.setMaxWaitMillis(10000);
               //返回连接时，是否提前进行 validate 操作
               jedisPoolConfig.setTestOnReturn( true );
               jedisPoolConfig.setTestWhileIdle( true );
               JedisPool jedisPool = new JedisPool(jedisPoolConfig, hostname, port, 3000);

               poolFactory.put(key, jedisPool);
            }
         }
      }
      return poolFactory.get(key);
   }
}

清除脚本 脚本随便起个名字

---------------------------------------------------------------------------------------------

host=data-stream-pika
port=19221
match=sk*
count=1000
isDel=no
output=/home/hadoop/flink_online_project/tool/redis_manager/keys

java -cp deletekey.jar com.xes.bdc.redis.tool.DeleteKey $host $port $match $count $isDel $output