本篇文章為大家展示了hbase0.98.9中如何實(shí)現(xiàn)endpoints，內(nèi)容簡(jiǎn)明扼要并且容易理解，絕對(duì)能使你眼前一亮，通過這篇文章的詳細(xì)介紹希望你能有所收獲。

成都創(chuàng)新互聯(lián)專注于企業(yè)成都全網(wǎng)營(yíng)銷推廣、網(wǎng)站重做改版、烏魯木齊網(wǎng)站定制設(shè)計(jì)、自適應(yīng)品牌網(wǎng)站建設(shè)、H5建站、商城網(wǎng)站開發(fā)、集團(tuán)公司官網(wǎng)建設(shè)、成都外貿(mào)網(wǎng)站建設(shè)、高端網(wǎng)站制作、響應(yīng)式網(wǎng)頁設(shè)計(jì)等建站業(yè)務(wù)，價(jià)格優(yōu)惠性價(jià)比高，為烏魯木齊等各大城市提供網(wǎng)站開發(fā)制作服務(wù)。

定制一個(gè)endpoint的過程。

下面是實(shí)現(xiàn)過程：

1、定義接口描述文件（該功能有protobuf提供出來）

option java_package = "coprocessor.endpoints.generated";
option java_outer_classname = "RowCounterEndpointProtos";
option java_generic_services = true;
option java_generate_equals_and_hash = true;
option optimize_for = SPEED;

message CountRequest {
}

message CountResponse {
  required int64 count = 1 [default = 0];
}

service RowCountService {
  rpc getRowCount(CountRequest)
    returns (CountResponse);
  rpc getKeyValueCount(CountRequest)
    returns (CountResponse);
}

這個(gè)文件我直接拿的hbase提供的example中的例子。其中的語法應(yīng)該有過類似經(jīng)驗(yàn)的一看就清楚了，實(shí)在不清楚就請(qǐng)查查protobuf的幫助手冊(cè)吧。

2、根據(jù)接口描述文件生成java接口類（該功能有protobuf提供出來）

有了接口描述文件，還需要生成java語言的接口類。這個(gè)需要借助protobuf提供的工具protoc。

$protoc --java_out=./ Examples.proto

簡(jiǎn)單解釋下，protoc這個(gè)命令在你裝了protobuf后就有了。Examples.proto這個(gè)是文件名，也就是剛才編寫的那個(gè)接口描述文件。“--java_out”這個(gè)用來指定生成后的java類放的地方。

所以，這地方如果你沒有裝protobuf，你需要裝一個(gè)，window和linux版都有，多說一句，如果你去裝hadoop64位的編譯環(huán)境的話，應(yīng)該是要裝protobuf。

3、實(shí)現(xiàn)接口

package coprocessor;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;

import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.Coprocessor;
import org.apache.hadoop.hbase.CoprocessorEnvironment;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.coprocessor.CoprocessorException;
import org.apache.hadoop.hbase.coprocessor.CoprocessorService;
import org.apache.hadoop.hbase.coprocessor.RegionCoprocessorEnvironment;
import org.apache.hadoop.hbase.filter.FirstKeyOnlyFilter;
import org.apache.hadoop.hbase.protobuf.ResponseConverter;
import org.apache.hadoop.hbase.regionserver.InternalScanner;
import org.apache.hadoop.hbase.util.Bytes;

import com.google.protobuf.RpcCallback;
import com.google.protobuf.RpcController;
import com.google.protobuf.Service;

import coprocessor.endpoints.generated.RowCounterEndpointProtos.CountRequest;
import coprocessor.endpoints.generated.RowCounterEndpointProtos.CountResponse;
import coprocessor.endpoints.generated.RowCounterEndpointProtos.RowCountService;

public class RowCounterEndpointExample extends RowCountService implements
		Coprocessor, CoprocessorService {
	private RegionCoprocessorEnvironment env;

	public RowCounterEndpointExample() {
	}

	@Override
	public Service getService() {
		return this;
	}

	@Override
	public void getRowCount(RpcController controller, CountRequest request,
			RpcCallback<CountResponse> done) {
		Scan scan = new Scan();
		scan.setFilter(new FirstKeyOnlyFilter());
		CountResponse response = null;
		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 = 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);
	}

	@Override
	public void getKeyValueCount(RpcController controller,
			CountRequest request, RpcCallback<CountResponse> done) {
		CountResponse response = null;
		InternalScanner scanner = null;
		try {
			scanner = env.getRegion().getScanner(new Scan());
			List<Cell> results = new ArrayList<Cell>();
			boolean hasMore = false;
			long count = 0;
			do {
				hasMore = scanner.next(results);
				for (Cell kv : results) {
					count++;
				}
				results.clear();
			} while (hasMore);

			response = 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);
	}

	@Override
	public void start(CoprocessorEnvironment env) throws IOException {
		if (env instanceof RegionCoprocessorEnvironment) {
			this.env = (RegionCoprocessorEnvironment) env;
		} else {
			throw new CoprocessorException("Must be loaded on a table region!");
		}
	}

	@Override
	public void stop(CoprocessorEnvironment env) throws IOException {
		// TODO Auto-generated method stub

	}

}

4、注冊(cè)接口（Hbase功能，通過配置文件或者表模式方式注冊(cè)）

這部分，可以看hbase權(quán)威指南了，我就看這部分做的。

5、測(cè)試調(diào)用

package coprocessor;

import java.io.IOException;
import java.util.Map;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.client.HTable;
import org.apache.hadoop.hbase.client.coprocessor.Batch;
import org.apache.hadoop.hbase.ipc.BlockingRpcCallback;
import org.apache.hadoop.hbase.ipc.ServerRpcController;
import org.apache.hadoop.hbase.util.Bytes;

import com.google.protobuf.ServiceException;

import coprocessor.endpoints.generated.RowCounterEndpointProtos.CountRequest;
import coprocessor.endpoints.generated.RowCounterEndpointProtos.CountResponse;
import coprocessor.endpoints.generated.RowCounterEndpointProtos.RowCountService;
import util.HBaseHelper;

public class RowCounterEndpointClientExample {
	public static void main(String[] args) throws ServiceException, Throwable {
		Configuration conf = HBaseConfiguration.create();
		HBaseHelper helper = HBaseHelper.getHelper(conf);
		//helper.dropTable("testtable");
		//helper.createTable("testtable", "colfam1", "colfam2");
		System.out.println("Adding rows to table...");
		helper.fillTable("testtable", 1, 10, 10, "colfam1", "colfam2");

		HTable table = new HTable(conf, "testtable");

		final CountRequest request = CountRequest.getDefaultInstance();
		
		final Batch.Call<RowCountService, Long> call =new Batch.Call<RowCountService, Long>() {
			public Long call(RowCountService counter)
					throws IOException {
				ServerRpcController controller = new ServerRpcController();
				BlockingRpcCallback<CountResponse> rpcCallback = new BlockingRpcCallback<CountResponse>();
				counter.getRowCount(controller, request, rpcCallback);
				CountResponse response = rpcCallback.get();
				if (controller.failedOnException()) {
					throw controller.getFailedOn();
				}
				return (response != null && response.hasCount()) ? response
						.getCount() : 0;
			}
		};
		
		Map<byte[], Long> results = table.coprocessorService(
				RowCountService.class, null, null, call);
		
		for(byte[] b : results.keySet()){
			System.err.println(Bytes.toString(b) + ":" + results.get(b));
		} 
	}
}

上述內(nèi)容就是hbase0.98.9中如何實(shí)現(xiàn)endpoints，你們學(xué)到知識(shí)或技能了嗎？如果還想學(xué)到更多技能或者豐富自己的知識(shí)儲(chǔ)備，歡迎關(guān)注創(chuàng)新互聯(lián)行業(yè)資訊頻道。

網(wǎng)站欄目：hbase0.98.9中如何實(shí)現(xiàn)endpoints
網(wǎng)站網(wǎng)址：http://chinadenli.net/article40/gepheo.html

成都網(wǎng)站建設(shè)公司_創(chuàng)新互聯(lián)，為您提供電子商務(wù)、品牌網(wǎng)站設(shè)計(jì)、網(wǎng)站營(yíng)銷、網(wǎng)站建設(shè)、建站公司、品牌網(wǎng)站建設(shè)

廣告

聲明：本網(wǎng)站發(fā)布的內(nèi)容（圖片、視頻和文字）以用戶投稿、用戶轉(zhuǎn)載內(nèi)容為主，如果涉及侵權(quán)請(qǐng)盡快告知，我們將會(huì)在第一時(shí)間刪除。文章觀點(diǎn)不代表本網(wǎng)站立場(chǎng)，如需處理請(qǐng)聯(lián)系客服。電話：028-86922220；郵箱：631063699@qq.com。內(nèi)容未經(jīng)允許不得轉(zhuǎn)載，或轉(zhuǎn)載時(shí)需注明來源： 創(chuàng)新互聯(lián)