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 = new
BlockingRpcCallback<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();
}