home/pinpoint
2
0
Fork 0
mirror of https://github.com/wahyd4/pinpoint.git synced 2026-08-24 12:16:33 +10:00
Code Issues Packages Projects Releases Wiki Activity

#320 change deprecated HTablePool api to HConnection

Browse Source
This commit is contained in:
Woonduk Kang
2015-04-10 20:32:29 +09:00
parent c51a62ce20
commit 30a9424e19
1 changed files with 41 additions and 11 deletions
Download Patch File Download Diff File Expand all files Collapse all files
commons/src/main/java/com/navercorp/pinpoint/common/hbase/PooledHTableFactory.java
+41 -11
View File
@@ -16,17 +16,22 @@
package com.navercorp.pinpoint.common.hbase;
import com.navercorp.pinpoint.common.util.ExecutorFactory;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.client.HTableInterface;
import org.apache.hadoop.hbase.client.HTableInterfaceFactory;
import org.apache.hadoop.hbase.client.HConnection;
import org.apache.hadoop.hbase.client.HConnectionManager;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.data.hadoop.hbase.HbaseSystemException;
import java.io.IOException;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
/**
* HTableInterfaceFactory based on HTablePool.
@@ -34,12 +39,24 @@ import java.util.concurrent.Executors;
*/
public class PooledHTableFactory implements HTableInterfaceFactory, DisposableBean {
private final Logger logger = LoggerFactory.getLogger(this.getClass());
public static final int DEFAULT_POOL_SIZE = 256;
public static final int DEFAULT_WORKER_QUEUE_SIZE = 1024*5;
private ExecutorService executor;
private HConnection connection;
public static final int DEFAULT_POOL_SIZE = 256;
public PooledHTableFactory(Configuration config) {
this.executor = Executors.newFixedThreadPool(DEFAULT_POOL_SIZE);
this(config, DEFAULT_POOL_SIZE, DEFAULT_WORKER_QUEUE_SIZE);
}
public PooledHTableFactory(Configuration config, int poolSize) {
this(config, poolSize, DEFAULT_WORKER_QUEUE_SIZE);
}
public PooledHTableFactory(Configuration config, int poolSize, int workerQueueSize) {
this.executor = getExecutorService(poolSize, workerQueueSize);
try {
this.connection = HConnectionManager.createConnection(config, executor);
} catch (IOException e) {
@@ -47,13 +64,13 @@ public class PooledHTableFactory implements HTableInterfaceFactory, DisposableBe
}
}
public PooledHTableFactory(Configuration config, int poolSize) {
this.executor = Executors.newFixedThreadPool(poolSize);
try {
this.connection = HConnectionManager.createConnection(config, executor);
} catch (IOException e) {
throw new HbaseSystemException(e);
}
private ExecutorService getExecutorService(int poolSize, int workQueueMaxSize) {
logger.info("create HConnectionThreadPoolExecutor poolSize:{}, workerQueueMaxSize:{}", poolSize, workQueueMaxSize);
ThreadPoolExecutor threadPoolExecutor = ExecutorFactory.newFixedThreadPool(poolSize, workQueueMaxSize, "Pinpoint-HConnectionExecutor", true);
threadPoolExecutor.prestartAllCoreThreads();
return threadPoolExecutor;
}
@@ -62,7 +79,7 @@ public class PooledHTableFactory implements HTableInterfaceFactory, DisposableBe
try {
return connection.getTable(tableName, executor);
} catch (IOException e) {
return null;
throw new HbaseSystemException(e);
}
}
@@ -79,5 +96,18 @@ public class PooledHTableFactory implements HTableInterfaceFactory, DisposableBe
if (connection != null) {
this.connection.close();
}
if (this.executor != null) {
this.executor.shutdown();
try {
final boolean shutdown = executor.awaitTermination(1000 * 5, TimeUnit.MILLISECONDS);
if (!shutdown) {
final List<Runnable> discardTask = this.executor.shutdownNow();
logger.warn("discard task size:{}", discardTask.size());
}
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
}
}
}