Hbase的Endpoint实现
Posted 修合无人见 存心有天知 -
tags:
篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了Hbase的Endpoint实现相关的知识，希望对你有一定的参考价值。
protobuf在windows环境下的安装
1、所需包下载地址：http://code.google.com/p/protobuf/downloads/list 。protobuf-x.x.x.tar.gz和protobuf-x.x.x-win32.zip 64位一样。
2、将protobuf-x.x.x-win32.zip 解压到 System32 文件夹下
3、解压protobuf-x.x.x-tar.gz到任意目录下，并复制protobuf-x.x.x-win32.zip的解压文件到src目录下
4、使用dos窗口进入到protobuf-x.x.x-tar.gz的解压路径下的java路径中，使用maven命令： mvn package 编辑该包。
生成protobuf-java-x.x.x.jar
5、使用自带的example测试环境是否安装成功，执行编译命令
protoc -I=<.proto文件路径> --java_out=<编译后的文件路径> xxxx.proto
.proto文件简介
/*全局属性定义，编译后的java类名，包路径*/
option java_package = "com.theta.hbase.coprocessor.protos";option java_outer_classname = "CountProtos";option java_generic_services = true;option java_generate_equals_and_hash = true;option optimize_for = SPEED;/*具体的消息*每个message域对应一个内部类，该内部类还包含一个Builder内部类*域内字段会生成对应的 setter和getter方法*使用Builder内部类来对字段赋值**/message CountRequest {required string startKey = 1;required string endKey = 2;}message CountResponse {required int64 count = 1 [default = 0];}/*提供服务的类*该类没有Builder内部类*/service CountService {rpc count(CountRequest)returns (CountResponse);}
HbaseEndpoint的实现
使用protobuf的message做为消息传递的格式，使用Rpc做为传输协议，一般会定义三个protobuf域，用于请求、响应、和业务实现。
1、endpont端要实现protobuf的service内部类，实现service定义的具体方法。使用response返回给调用者。
例子，实现count操作
@Override
public void count(RpcController controller, CountRequest request,
RpcCallback<CountResponse> done) {
Scan scan = new Scan();
scan.setFilter( new FirstKeyOnlyFilter());
scan.setStartRow(Bytes. toBytes(request.getStartKey()));
scan.setStopRow(Bytes. toBytes(request.getEndKey()));
InternalScanner scanner = null;
try {
scanner = env.getRegion().getScanner(scan);
List<Cell> results = new ArrayList<Cell>();
boolean hasMore = false;
byte[] lastRow = null;
long count = 0;
do {
hasMore = scanner.next(results);
for (Cell kv : results) {
byte[] currentRow = CellUtil.cloneRow(kv);
if (lastRow == null || !Bytes.equals(lastRow, currentRow)) {
lastRow = currentRow;
count++;
}
}
results.clear();
} while (hasMore);
response = CountProtos.CountResponse.newBuilder().setCount(count).build();
} catch (IOException ioe) {
ResponseConverter. setControllerException(controller, ioe);
} finally {
if (scanner != null) {
try {
scanner.close();
} catch (IOException ignored) {
}
}
}
done.run(response);
}
2、client端使用HTable实例的 coprocessorService 方法发送Rpc请求,得到response结果例子：public long count(HTableInterface ht, String startKey, String endKey)throws Throwable {final CountProtos.CountRequest request = CountProtos.CountRequest. newBuilder().setStartKey(startKey).setEndKey(endKey).build();Map< byte[], Long> results = ht.coprocessorService(CountProtos.CountService. class, null , null,new Batch.Call<CountProtos.CountService, Long>() {public Long call(CountProtos.CountService counter)throws IOException {ServerRpcController controller = new ServerRpcController();BlockingRpcCallback<CountProtos.CountResponse> rpcCallback = newBlockingRpcCallback<CountProtos.CountResponse>();counter.count(controller, request, rpcCallback);CountProtos.CountResponse response = rpcCallback.get();if (controller.failedOnException()) {throw controller.getFailedOn();}return (response != null && response.getCount() != 0) ?response.getCount() : 0;}});System. out.println(results.values());return results.size();}
以上是关于Hbase的Endpoint实现的主要内容，如果未能解决你的问题，请参考以下文章