一、介紹

上一屆簡單介紹了協處理器的主要功能以及相應的特點，但沒有對寫出器的具體內容作出詳細的解釋。這篇文章解釋了協處理器的觀察者模式。觀察者模式就是提供了一個觸發器，使用者通過整合相應的類（BaseRegionObverser等），重寫其中想要實現的方法，然後將協處理器載入到表中，這是表就會通過協處理器“監聽”使用者預先設定的動作，一旦該動作被執行，使用者所寫的鉤子函式就被觸發，然後就能實現相應的功能。

二、詳解

1、觀察者模式介紹

觀察者模式提供了一種類似於觸發器的功能，即在使用者執行某些已經被指定的動作時，會觸發觀察者模式，並執行使用者已經提前設定在觀察者模式中的程式碼。

觀察則模式主要分為三種類型：region級別的觀察者模式、WAL級別的觀察者模式以及master級別的觀察者模式。這三種觀察者模式所“監聽”的物件分別是region操作級別物件、wal操作級別的物件以及master操作級別的物件。

region級別：主要提供一些針對於region級別操作（put、get、delete等）的鉤子函式，使用者定義的此類觀察這在這些操作執行時被觸發。

wal級別：主要是提供一些針對於wal級別操作的鉤子函式。

master級別：主要是提供了一些針對於master級別操作（createtable、disabletable、droptable等）的鉤子函式

2、region級別觀察者

實現region級別的觀察者時需要繼承一個基本類，該類中已經包含了所有的region級別的鉤子函式，使用者只需要進行重寫即可：BaseRegiobObverser

在提供的所有的鉤子函式，都是以preDo() / postDo() 成對存在，例如 prePut() / postPut() 函式就是成對存在。preDo() 系列的函式表明了在執行Do所執行的動作之前執行鉤子函式。postDo() 系列函式標明在執行了Do之後執行鉤子函式。

例如下面的例子，我們要實現一個複製例子的資料，即將插入test表中的資料也全都插入到testCopy表中：

（1）進行Obverser兌現的編寫

觀察者例項：

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

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HColumnDescriptor;
import org.apache.hadoop.hbase.HTableDescriptor;
import org.apache.hadoop.hbase.KeyValue;
import org.apache.hadoop.hbase.client.Durability;
import org.apache.hadoop.hbase.client.HBaseAdmin;
import org.apache.hadoop.hbase.client.HTable;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.coprocessor.BaseRegionObserver;
import org.apache.hadoop.hbase.coprocessor.ObserverContext;
import org.apache.hadoop.hbase.coprocessor.RegionCoprocessorEnvironment;
import org.apache.hadoop.hbase.regionserver.wal.WALEdit;
import org.apache.hadoop.hbase.util.Bytes;

/**
 * @author wozipa
 * @Date 2016-3-20 20:57
 * @see 建立兩張一模一樣的表
 */
public class RegionObverserExample extends BaseRegionObserver{
	
	private static String tableName="testCopy";
	private static String family="copy";
	
	/**
	 * @author wozipa
	 * @Date 2016-3-20 21：00
	 * @see 進行表格的配置
	 * @return
	 */
	public Configuration init()
	{
		Configuration conf=new Configuration();
		conf.set("hbase.master", "hadoop1:16000");
		conf.set("hbase.zookeeper.quorum", "hadoop1,hadoop2,hadoop3,hadoop4,hadoop5,hadoop6");
		conf.set("hbase.zookeeper.property.clientPort", "2181");
		return conf;
	}
	

	/**
	 * @author wozipa
	 * @Date 2016-3-20 21:02
	 * @see 在插入資料之前將資料插入到指定的表格中
	 */
	@Override
	public void prePut(ObserverContext<RegionCoprocessorEnvironment> e, Put put, WALEdit edit, Durability durability)
			throws IOException {
		// TODO Auto-generated method stub
		Configuration conf=init();
		try {
			HBaseAdmin admin=new HBaseAdmin(conf);
			if(!admin.tableExists(tableName))
			{
				HTableDescriptor tableDescriptor=new HTableDescriptor(tableName);
				HColumnDescriptor columnDescriptor=new HColumnDescriptor(family);
				tableDescriptor.addFamily(columnDescriptor);
				admin.createTable(tableDescriptor);
				admin.close();
			}
			//建立資料插入該保重，保持行鍵值和列名、value值的相同
			Map<byte[],List<KeyValue>> map=put.getFamilyMap();
			List<KeyValue> list=map.get(put.getRow());
			Put put2=new Put(put.getRow());
			for(KeyValue kv:list)
			{
				put2.addColumn(Bytes.toBytes(family),kv.getQualifier(),kv.getValue());
			}
			//建立表連線
			HTable table=new HTable(conf, tableName);
			table.put(put2);
			//
			table.close();
		} catch (Exception e1) {
			// TODO: handle exception
			e1.printStackTrace();
		}
	}
}
 

在該物件中，通過實現prePut() 函式，即在test表的region執行put方法之前，將put中的資料插入到testCopy中。

（2）將專案打包成jar包後上傳到HDFS上

然後將該類所在的專案打包上傳到HDFS上：

（3）編寫例子類

import java.io.IOException;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hbase.Coprocessor;
import org.apache.hadoop.hbase.HColumnDescriptor;
import org.apache.hadoop.hbase.HTableDescriptor;
import org.apache.hadoop.hbase.client.HBaseAdmin;
import org.apache.hadoop.hbase.client.HTable;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.util.Bytes;

import com.yammer.metrics.stats.EWMA;

/**
 * @author wozipa
 * @Date 2016-3-20 21:16
 * @see 用來測試協處理器的各個功能的函式集合
 *
 */
public class Main {
	
	private static String tableName="test";
	private static String family="test";
	
	public static void main(String[] args)
	{
		Main main=new Main();
		main.putCoprocessExample();
	}
	
	public Configuration init()
	{
		Configuration conf=new Configuration();
		conf.set("hbase.master", "hadoop1:16000");
		conf.set("hbase.zookeeper.quorum", "hadoop1,hadoop2,hadoop3,hadoop4,hadoop5,hadoop6");
		conf.set("hbase.zookeeper.property.clientPort", "2181");
		return conf;
	}
	
	/**
	 * @author wozipa
	 * @Date 2016-3-20 21：17
	 * @see 進行資料
	 */
	public void putCoprocessExample()
	{
		Configuration conf=init();
		HBaseAdmin admin=null;
		try {
			admin=new HBaseAdmin(conf);
			if(!admin.tableExists(tableName))
			{
				HTableDescriptor tableDescriptor=new HTableDescriptor(tableName);
				HColumnDescriptor columnDescriptor=new HColumnDescriptor(family);
				tableDescriptor.addFamily(columnDescriptor);
				Path path=new Path("hdfs://192.98.12.234:9000/Coprocessor/Copro.jar");
				tableDescriptor.setValue("COPROCESSOR$1",path.toString()+"|"+"com.geoway.coprocessor.RegionObverserExample|"+Coprocessor.PRIORITY_USER);
				admin.createTable(tableDescriptor);
			}
			//想資料庫中插入資料
			Put put=new Put(Bytes.toBytes("row-2"));
			put.addColumn(Bytes.toBytes(family), Bytes.toBytes("qualifier1"), Bytes.toBytes("value2"));
			HTable table=new HTable(conf, tableName);
			table.put(put);
			table.close();
		} catch (IOException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
	}

}

最後執行結構會發現兩個表同時被建立：

3、master級別的觀察者模式

master級別的觀察者模式與region級別的模式相同，只不過其“監聽”的物件變成了master級別的操作。也就是相當於SQL語句中的DDL語句。主要包括對錶的一些操作物件和region級別的操作鉤子函式。

HBase中也提供了一個 BaseMasterObverser物件，該物件中也封裝了所有的DDL操作的鉤子函式。使用者只需要繼承該物件，並重寫相應的方法，就可實現相應的功能。其實現流程與region級別的流程一樣。

三、總結

觀察者模式提供了使用者在進行region級別操作之前與之後的自定義動作，因此可以實現很多功能，例如可以聽過觀察者實現二級索引等功能。