版本：

HBase：1.1.2；Hadoop2.6.0；Eclipse：Mars.1 Release (4.5.1)；

应用场景：HBase中有一个表，里面有一条记录，如下：

现在需要根据value的值，比如当其为“true”的时候进行更新操作（也就是插入操作，表设置Versions为100），更新为false；

问题描述：

1） 当只有一个线程来操作的时候，可以直接取得值，然后根据值来判断即可进行插入；

2）如果有多个线程的话（比如10个线程），那么1）的方法就不行了，会直接插入10条记录（因为有多个版本，所以会查到11条记录）；

3）如果使用rowlock的话，可以解决这个问题，在进行put的时候，去获取rowlock，如果获取不了（其他线程已经先获得了rowlock），那么就不插入；

4）但是，rowlock在hbase之前的版本是有的，但是后面的版本就没有了（性能影响，所以hbase把client的rowlock去掉了）；

那么，怎么办？

参考：http://www.ngdata.com/hbase-row-locks/ 

使用HTable.checkAndPut() 即可解决这个问题；

测试代码：

main:

package demo;

import java.io.IOException;
import java.text.SimpleDateFormat;
import java.util.Date;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.client.Connection;
import org.apache.hadoop.hbase.client.ConnectionFactory;

public class TestMultiPut {
/**
 * 
 * @param args
 */
	public static final String TABLE = "multiput";
	public static final String FAMILY = "cf1";
	public static final String COL1 = "q1";
	public static final String ROWKEY1 = "rk1";
	public static final SimpleDateFormat sf = new SimpleDateFormat("yyyy/MM/dd HH:mm:ss SSS") ;
	public static void main(String[] args) throws IOException {
		Configuration conf = HBaseConfiguration.create();
		conf.set("hbase.master", "node2:16000");// 指定HMaster
		conf.set("hbase.rootdir", "hdfs://node1:8020/hbase");// 指定HBase在HDFS上存储路径
		conf.set("hbase.zookeeper.quorum", "node2,node3,node4");// 指定使用的Zookeeper集群
		conf.set("hbase.zookeeper.property.clientPort", "2181");// 指定使用Zookeeper集群的端口

		Connection connection = ConnectionFactory.createConnection(conf);
		for(int i=0;i<10;i++){
//			new Thread(new PutThread(connection)).start();
			new Thread(new PutThread2(connection)).start();
		}
		System.out.println("all done");
	}

	public static String getTimeStr(){
		return sf.format(new Date(System.currentTimeMillis()));
	}
}

thread:

package demo;

import static demo.TestMultiPut.COL1;
import static demo.TestMultiPut.FAMILY;
import static demo.TestMultiPut.ROWKEY1;
import static demo.TestMultiPut.TABLE;

import java.io.IOException;

import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.Connection;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Table;
import org.apache.hadoop.hbase.filter.CompareFilter.CompareOp;
import org.apache.hadoop.hbase.util.Bytes;
public class PutThread2 implements Runnable {
	Connection connection ;
	public PutThread2(Connection conn) {
		this.connection= conn;
	}
	@Override
	public void run() {
		try {
			Table table = connection.getTable(TableName.valueOf(TABLE));
			
			
			Put put = new Put(Bytes.toBytes(ROWKEY1));
			put.addColumn(FAMILY.getBytes(), COL1.getBytes(), "false".getBytes());
			
			
			boolean flag = table.checkAndPut(Bytes.toBytes(ROWKEY1), Bytes.toBytes(FAMILY), Bytes.toBytes(COL1),
					CompareOp.EQUAL, "true".getBytes(), put);
			
			table.close();
			if(flag){
				System.out.println(TestMultiPut.getTimeStr()
					+" ，thread:"+Thread.currentThread().getName()+" 插入成功--------");
			}else{
				System.out.println(TestMultiPut.getTimeStr()
						+" ，thread:"+Thread.currentThread().getName()+" 插入失败XXXXXXX");
			}
			
		} catch (IOException e) {
			System.out.println(TestMultiPut.getTimeStr()
					+" 报错：thread:"+Thread.currentThread().getName());
			e.printStackTrace();
			
		}
	}

}

运行main，得到下面的日志：

同时，查看HBase表，

也只是插入了一条记录；

总结：

1. 使用HTable.checkAndPut()，可以解决row lock的问题；

2. 为什么插入成功的，其线程时间不是最小的？

可能是因为网络延时；

分享，成长，快乐

脚踏实地，专注

转载请注明blog地址：http://blog.csdn.net/fansy1990