专栏首页扎心了老铁commons-pool与commons-pool2连接池(Hadoop连接池)

commons-pool与commons-pool2连接池(Hadoop连接池)

commons-pool和commons-pool2是用来建立对象池的框架,提供了一些将对象池化必须要实现的接口和一些默认动作。对象池化之后可以通过pool的概念去管理其生命周期,例如对象的创建,使用,销毁等。例如我们通常使用的连接池,连接池可以有效管理连接的数量和状态,保证连接资源的情况而且避免并发场景下连接的频繁建立和释放。

我们这里来讲述如何使用commons-pool2来池化对象。我们以池化hadoop连接为例。

1、先解决依赖

     <dependency>
            <groupId>org.apache.commons</groupId>
            <artifactId>commons-pool2</artifactId>
            <version>2.3</version>
        </dependency>

2、如何使用连接池

我们是在spingboot框架中池化hadoop集群连接,先看一下池化之后的效果。

下面是我们池化之后的hadoop集群客户端。可以看到我们可以通过连接池的方式管理hadoo集群的链接。

1)配置连接池

最大连接数maxTotal

最大空闲连接数maxIdle

最小空闲连接数minIdle

获取连接的最大等待时间maxWait

可以看到传入这些配置的时候我们使用了一个config对象JHadoopPoolConfig,后面我们将说明这个config对象如何实现。

2)管理连接池

我们以三个函数说明了如何去连接池中申请连接,使用连接和释放链接资源。

申请资源pool.getResource()

释放资源pool.returnBrokenResource()和pool.returnResource()

这里要注意的是,一定要在catch和finally中成功释放资源,不然会导致could not get a Resource from the Pool的异常

package com.xiaoju.dqa.jazz.hadoop.client;

import org.apache.hadoop.fs.FileStatus;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.text.SimpleDateFormat;
import java.util.Date;


public class JHadoopClient {
    protected final Logger logger = LoggerFactory.getLogger(this.getClass());

    private JHadoopPool jHadoopPool;
    private String coreResource;
    private String hdfsResource;
    private int maxTotal;
    private int maxIdle;
    private int minIdle;
    private int maxWaitMillis;

    public String getCoreResource() {
        return coreResource;
    }
    public void setCoreResource(String coreResource) {
        this.coreResource = coreResource;
    }
    public String getHdfsResource() {
        return hdfsResource;
    }
    public void setHdfsResource(String hdfsResource) {
        this.hdfsResource = hdfsResource;
    }
    public int getMaxTotal() {
        return maxTotal;
    }
    public void setMaxTotal(int maxTotal) {
        this.maxTotal = maxTotal;
    }
    public int getMaxIdle() {
        return maxIdle;
    }
    public void setMaxIdle(int maxIdle) {
        this.maxIdle = maxIdle;
    }
    public int getMaxWaitMillis() {
        return maxWaitMillis;
    }
    public void setMaxWaitMillis(int maxWaitMillis) {
        this.maxWaitMillis = maxWaitMillis;
    }
    public int getMinIdle() {
        return minIdle;
    }
    public void setMinIdle(int minIdle) {
        this.minIdle = minIdle;
    }

    public void init() {
        try {
            JHadoopPoolConfig conf = new JHadoopPoolConfig();
            conf.setMaxTotal(maxTotal);
            conf.setMaxIdle(maxIdle);
            conf.setMinIdle(minIdle);
            conf.setMaxWaitMillis(maxWaitMillis);
            jHadoopPool = new JHadoopPool(conf, coreResource, hdfsResource);
            logger.info("[HDFS]初始化JHadoopClient成功");
        } catch (Exception ex) {
            logger.error("[HDFS]初始化JHadoopClient失败", ex);
        }
    }

    public void stop() {
        try {
            jHadoopPool.destroy();
        } catch(Exception e) {
        }
    }

    public boolean exists(String path) throws Exception {
        JHadoop jHadoop = null;
        boolean broken = false;
        try {
            jHadoop = jHadoopPool.getResource();
            return jHadoop.exists(path);
        } catch (Exception e) {
            broken = true;
            jHadoopPool.returnBrokenResource(jHadoop);
            logger.error("[HDFS]判断文件是否存在失败", e);
            throw e;
        } finally {
            if (null != jHadoop && !broken) {
                jHadoopPool.returnResource(jHadoop);
            }
        }
    }

    public String getModificationTime(String path) throws Exception {
        JHadoop jHadoop = null;
        boolean broken = false;
        try {
            jHadoop = jHadoopPool.getResource();
            FileStatus fileStatus = jHadoop.getFileStatus(path);
            long modifyTimestamp = fileStatus.getModificationTime();
            SimpleDateFormat simpleDateFormat = new SimpleDateFormat("yyyyMMddHHmmss");
            Date date = new Date(modifyTimestamp);
            return simpleDateFormat.format(date);
        } catch (Exception e) {
            broken = true;
            jHadoopPool.returnBrokenResource(jHadoop);
            logger.error("[HDFS]获取最近修改时间失败", e);
            throw e;
        } finally {
            if (null != jHadoop && !broken) {
                jHadoopPool.returnResource(jHadoop);
            }
        }
    }

    public long getPathSize(String path) throws Exception {
        JHadoop jHadoop = null;
        boolean broken = false;
        try {
            jHadoop = jHadoopPool.getResource();
            return jHadoop.getContentSummary(path).getLength();
        } catch (Exception e) {
            broken = true;
            jHadoopPool.returnBrokenResource(jHadoop);
            logger.error("[HDFS]获取路径大小失败", e);
            throw e;
        } finally {
            if (null != jHadoop && !broken) {
                jHadoopPool.returnResource(jHadoop);
            }
        }
    }

}

3)注册成bean

通过配置文件传入链接池相应的配置。

package com.xiaoju.dqa.jazz.hadoop.configuration;

import com.xiaoju.dqa.jazz.hadoop.client.JHadoopClient;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
public class HadoopConfig {

    @Value("${hadoop.core.resource}")
    private String coreResource;
    @Value("${hadoop.hdfs.resource}")
    private String hdfsResource;
    @Value("${hadoop.pool.maxTotal}")
    private int maxTotal;
    @Value("${hadoop.pool.maxIdle}")
    private int maxIdle;
    @Value("${hadoop.pool.minIdle}")
    private int minIdle;
    @Value("${hadoop.pool.maxWaitMillis}")
    private int maxWaitMillis;

    @Bean(initMethod = "init", destroyMethod = "stop")
    public JHadoopClient jHadoopClient() {
        JHadoopClient jHadoopClient = new JHadoopClient();
        jHadoopClient.setMaxTotal(maxTotal);
        jHadoopClient.setMaxIdle(maxIdle);
        jHadoopClient.setMinIdle(minIdle);
        jHadoopClient.setMaxWaitMillis(maxWaitMillis);
        jHadoopClient.setCoreResource(coreResource);
        jHadoopClient.setHdfsResource(hdfsResource);
        return jHadoopClient;
    }
}

4)config对象如何实现

我们这里要说明一下下面这些参数的含义:

1)setTestWhileConfig - 在空闲时检查有效性, 默认false

2)setMinEvictableIdleTimeMillis - 逐出连接的最小空闲时间

3)setTimeBetweenEvictionRunsMillis - 逐出扫描的时间间隔(毫秒) 如果为负数则不运行逐出线程,默认-1

4)setNumTestsPerEvictionRun - 每次逐出检查时 逐出的最大数目

package com.xiaoju.dqa.jazz.hadoop.client;


import org.apache.commons.pool2.impl.GenericObjectPoolConfig;


public class JHadoopPoolConfig extends GenericObjectPoolConfig {
    public JHadoopPoolConfig() {
        this.setTestWhileIdle(true);
        this.setMinEvictableIdleTimeMillis(60000L);
        this.setTimeBetweenEvictionRunsMillis(30000L);
        this.setNumTestsPerEvictionRun(-1);
    }
}

3、连接池JHadoopPool

这个类继承了Pool<JHadoop>,用来初始化连接池对象。而JHadoop是Pool要管理的连接对象。

可以看到JHadoopPool在初始化的时候传入了一个JHadoopFactory的实例。这个实例将会以工厂模式来创建实际的JHadoop

JHadoopPool代码

package com.xiaoju.dqa.jazz.hadoop.client;

import org.apache.commons.pool2.impl.GenericObjectPoolConfig;

public class JHadoopPool extends Pool<JHadoop> {
    public JHadoopPool(GenericObjectPoolConfig poolConfig, String coreResource, String hdfsResource) {
        super(poolConfig, new JHadoopFactory(coreResource, hdfsResource));
    }

    public JHadoopPool(GenericObjectPoolConfig poolConfig) {
        super(poolConfig, new JHadoopFactory());
    }

}

JHadoop代码

JHadoop实现了hadoop.fs中的方法调用。

我这里只给出了几个函数的简单封装,你可以根据具体的需要进行增加。

package com.xiaoju.dqa.jazz.hadoop.client;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.ContentSummary;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;


public class JHadoop {
    protected final Logger logger = LoggerFactory.getLogger(this.getClass());
    private FileSystem fs;
    private String coreResource;
    private String hdfsResource;

    public JHadoop(String coreResource, String hdfsResource) {
        this.coreResource = coreResource;
        this.hdfsResource = hdfsResource;
    }

    public String getCoreResource() {
        return coreResource;
    }
    public void setCoreResource(String coreResource) {
        this.coreResource = coreResource;
    }
    public String getHdfsResource() {
        return hdfsResource;
    }
    public void setHdfsResource(String hdfsResource) {
        this.hdfsResource = hdfsResource;
    }


    public void open() {
        try {
            Configuration conf = new Configuration();
            conf.addResource(coreResource);
            conf.addResource(hdfsResource);
            fs = FileSystem.get(conf);
            logger.info("[JHadoop]创建实例成功");
        } catch (Exception e) {
            logger.error("[JHadoop]创建实例失败", e);
        }
    }

    public void close() {
        try {
            if (null != fs) {
                fs.close();
                logger.info("[JHadoop]关闭实例成功");
            }
        } catch(Exception e) {
            logger.error("[JHadoop]关闭实例失败", e);
        }
    }

    public boolean isConnected() throws IOException {
        fs.exists(new Path("/forTest"));
        return true;
    }

    public boolean exists(String path) throws IOException {
        Path hdfsPath = new Path(path);
        return fs.exists(hdfsPath);
    }

    public FileStatus getFileStatus(String path) throws IOException {
        Path hdfsPath = new Path(path);
        return fs.getFileStatus(hdfsPath);
    }

    public ContentSummary getContentSummary(String path) throws IOException {
        ContentSummary contentSummary = null;
        Path hdfsPath = new Path(path);
        if (fs.exists(hdfsPath)) {
            contentSummary = fs.getContentSummary(hdfsPath);
        }
        return contentSummary;
    }

}

4、连接工厂类JHadoopFactory

JHadoopFactory这个类管理着连接对象的创建,销毁,验证等动作

package com.xiaoju.dqa.jazz.hadoop.client;

import org.apache.commons.pool2.PooledObject;
import org.apache.commons.pool2.PooledObjectFactory;
import org.apache.commons.pool2.impl.DefaultPooledObject;


public class JHadoopFactory implements PooledObjectFactory<JHadoop> {
    private final String coreResource;
    private final String hdfsResource;

    public JHadoopFactory() {
        this.coreResource = "core-site.xml";
        this.hdfsResource = "hdfs-site.xml";
    }
    public JHadoopFactory(String coreResource, String hdfsResource) {
        this.coreResource = coreResource;
        this.hdfsResource = hdfsResource;
    }

    @Override
    public PooledObject<JHadoop> makeObject() throws Exception {
        JHadoop jHadoop = new JHadoop(coreResource, hdfsResource);
        jHadoop.open();
        return new DefaultPooledObject<JHadoop>(jHadoop);
    }

    @Override
    public void destroyObject(PooledObject<JHadoop> pooledJHadoop) throws Exception {
        JHadoop jHadoop = pooledJHadoop.getObject();
        jHadoop.close();
    }

    @Override
    public boolean validateObject(PooledObject<JHadoop> pooledJHadoop) {
        JHadoop jHadoop = pooledJHadoop.getObject();
        try {
            return jHadoop.isConnected();
        } catch (Exception e) {
            return false;
        }
    }

    @Override
    public void activateObject(PooledObject<JHadoop> pooledObject) throws Exception {

    }

    @Override
    public void passivateObject(PooledObject<JHadoop> pooledObject) throws Exception {

    }
}

本文参与腾讯云自媒体分享计划,欢迎正在阅读的你也加入,一起分享。

我来说两句

0 条评论
登录 后参与评论

相关文章

  • 大数据算法设计模式(2) - 左外链接(leftOuterJoin) spark实现

    左外链接(leftOuterJoin) spark实现 package com.kangaroo.studio.algorithms.join; impor...

    用户1225216
  • Elasticsearch强制重置未分配的分片(unassigned)

    强制重置未分片的分片,这个问题源自于Elasticsearch维护中,Node意外退出的场景。 意外退出后Elasticsearch由于网络原因或者jvm性能压...

    用户1225216
  • 使用HDFS客户端java api读取hadoop集群上的信息

    本文介绍使用hdfs java api的配置方法。 1、先解决依赖,pom <dependency> <groupId>org.apac...

    用户1225216
  • Android监听手机短信的示例代码

    方式1:定义为Activity的内部类,直接在onChang()方法中处理业务逻辑,比如收到短信之后设置给EditText

    砸漏
  • 锻造正则神兵之Java源码分析器-V0.01

    张风捷特烈
  • guava 集合和函数接口妙用

    用户5166556
  • Effective Java通俗理解(下)

    第31条:用实例域代替序数   枚举类型有一个ordinal方法,它范围该常量的序数从0开始,不建议使用这个方法,因为这不能很好地对枚举进行维护,正确应该是利用...

    用户1148394
  • Java内存模型—JMMhappens-before规则

     有时候编译器、处理器的优化会导致runtime与我们设想的不一样,为此Java对编译器和处理器做了一些限制,JAVA内存模型(JMM)将这些抽象出来,这样编写...

    用户1174983
  • 基础篇-Java三大特性之封装

    写了这么长时间的代码,现在感觉代码设计真的是一门艺术,虽然是一个非常简单的增删改查的需求,但是1000个人可能有1000个不同的设计方法,怎么设计的更为优雅、美...

    一个程序员的成长
  • Java爬虫可以非常溜

    xxl-crawler是 许雪里 大佬开源的一个java爬虫,熟悉java语言的用起来可以非常顺手。

    java乐园

扫码关注云+社区

领取腾讯云代金券