请带着如下问题阅读本文。
1、什么是行锁?
2、HBase行锁的原理是什么?
3、HBase行锁是如何实现的?
4、HBase行锁是如何应用的?
一、什么是行锁?
我们知道,数据库中存在事务的概念。事务是作为单个逻辑工作单元执行的一系列操作,要么完全地执行,要么完全的不执行。而事务的四大特点即原子性、一致性、分离性和持久性。其中,原子性首当其冲,那么在HBase内部实现其原子性的重要保证是什么呢?答案就是行锁。
什么是行锁呢?顾名思义,它就是加在行上的一把锁。在它未释放该行前,最起码其他访问者是无法对该行做修改的,即要修改的话,必须得获得该行的锁才能拥有修改改行数据的权限,这就是行锁的含义。
二、HBase行锁实现原理
HBase行锁是利用Java并发包concurrent里的CountDownLatch(1)来实现的。它的主要思想就是在服务器端每个访问者单独一个数据处理线程,每个处理线程针对特定行数据修改时必须获得该行的行锁,而其他客户端线程想要修改数据的话,必须等待前面的线程释放锁后才被允许,这就利用了Java并发包中的CountDownLatch,CountDownLatch为Java中的一个同步辅助类,在完成一组正在其他线程中进行的操作之前,它允许一个或多个线程一直等待。这里,将线程数设置为1,十分巧妙的实现了独占锁的概念。
三、HBase行锁的实现
HBase的行锁主要是通过HRegion的两个内部类实现的,其中一个是RowLock,另外一个是RowLockContext。
我们首先看RowLock这个类,其定义如下:
/** * Row lock held by a given thread. * One thread may acquire multiple locks on the same row simultaneously. * The locks must be released by calling release() from the same thread. * * 给定线程持有的行锁。 * 一个线程可以同时获得同一行上的多个锁。 * 锁必须被相同线程,通过调用release()释放。 */ public static class RowLock { // 行锁上下文,持有锁定的行row、锁持有者线程thread、该行上锁的数目lockCount等内容 @VisibleForTesting final RowLockContext context; // 行锁是否被释放 private boolean released = false; // 构造函数 @VisibleForTesting RowLock(RowLockContext context) { this.context = context; } /** * Release the given lock. If there are no remaining locks held by the current thread * then unlock the row and allow other threads to acquire the lock. * 释放给定的锁。如果当前线程不再持有任何锁,那么对该行解锁并允许其他线程获得锁。 * @throws IllegalArgumentException if called by a different thread than the lock owning thread */ public void release() { if (!released) { context.releaseLock(); released = true; } } }
通过上述源码我们可以看到,行锁RowLock有两个成员变量,RowLockContext类型的行锁上下文context和布尔类型的行锁是否释放released。其中,行锁上下文context持有锁定的行row、锁持有者线程thread、该行上锁的数目lockCount等内容,并且,利用java的concurrent并发包里的CountDownLatch(1)实现了线程对对象的独占锁。
RowLockContext的源码如下:
// 行锁上下文,包括指定的行row,同步计数器latch,锁的数目lockCount和线程thread @VisibleForTesting class RowLockContext { private final HashedBytes row;// 行 private final CountDownLatch latch = new CountDownLatch(1);// private final Thread thread; private int lockCount = 0; // 构造方法 RowLockContext(HashedBytes row) { this.row = row; this.thread = Thread.currentThread(); } // 判断是否为当前线程对应的行锁上下文 boolean ownedByCurrentThread() { return thread == Thread.currentThread(); } RowLock newLock() { lockCount++; return new RowLock(this); } void releaseLock() { if (!ownedByCurrentThread()) { throw new IllegalArgumentException("Lock held by thread: " + thread + " cannot be released by different thread: " + Thread.currentThread()); } lockCount--; if (lockCount == 0) { // no remaining locks by the thread, unlock and allow other threads to access RowLockContext existingContext = lockedRows.remove(row); if (existingContext != this) { throw new RuntimeException( "Internal row lock state inconsistent, should not happen, row: " + row); } // 同步计数器减1 latch.countDown(); } } }
通过源码我们可以看到,行锁的上下文信息,主要包括行锁对应的行row以及占用该行锁的线程thread。构造RowContext时,只需传入行row即可,占用的线程则通过Thread.currentThread()获得当前线程。
新加锁时,通过调用newLock()方法即可实现,首先锁的计数器lockCount加1,然后返回由当前RowContext构造RowLock实例即可。
释放锁时,通过调用releaseLock()方法即可实现,首先通过ownedByCurrentThread()方法确保调用releaseLock()方法的当前线程是否和RowContext持有的线程一致,然后,锁的计数器lockCount减1,并且,如果lockCount为0的话,说明不再有操作占用该行,将row对应的行锁从数据结构lockedRows中删除,允许其他线程获得该行的行锁,最后,最重要的一步,latch.countDown(),就可完成行锁的释放了。
四、HBase行锁的使用
下面,我们看下HBase行锁的使用。在涉及数据变更的操作,比如Put、Delete等中,在对一行数据操作之前,都会调用getRowLockInternal()方法,获得该行数据的行锁。代码如下:
/** * A version of getRowLock(byte[], boolean) to use when a region operation has already been * started (the calling thread has already acquired the region-close-guard lock). */ protected RowLock getRowLockInternal(byte[] row, boolean waitForLock) throws IOException { // 构造HashedBytes类型表示的行,rowKey HashedBytes rowKey = new HashedBytes(row); // 创建行锁上下文实例,并制定为行rowkey和当前线程拥有 RowLockContext rowLockContext = new RowLockContext(rowKey); // loop until we acquire the row lock (unless !waitForLock) while (true) { // 将rowkey与行锁上下文的对应关系添加到Region的数据结构lockedRows中 RowLockContext existingContext = lockedRows.putIfAbsent(rowKey, rowLockContext); if (existingContext == null) { // Row is not already locked by any thread, use newly created context. // 该行已经没有被任何线程锁住,使用这个新创建的上下文 break; } else if (existingContext.ownedByCurrentThread()) { // Row is already locked by current thread, reuse existing context instead. // 该行已经被当前线程锁住,复用当前线程之前创建的行锁上下文实例 rowLockContext = existingContext; break; } else { // 该行被其他线程锁住,如果不需要等待锁,直接返回null if (!waitForLock) { return null; } TraceScope traceScope = null; try { if (Trace.isTracing()) { traceScope = Trace.startSpan("HRegion.getRowLockInternal"); } // Row is already locked by some other thread, give up or wait for it // 行已经被其他线程锁住,放弃或者等待 // 等待rowLockWaitDuration时间后,如果还未获得行锁,直接抛出异常 if (!existingContext.latch.await(this.rowLockWaitDuration, TimeUnit.MILLISECONDS)) { if(traceScope != null) { traceScope.getSpan().addTimelineAnnotation("Failed to get row lock"); } throw new IOException("Timed out waiting for lock for row: " + rowKey); } if (traceScope != null) traceScope.close(); traceScope = null; } catch (InterruptedException ie) { LOG.warn("Thread interrupted waiting for lock on row: " + rowKey); InterruptedIOException iie = new InterruptedIOException(); iie.initCause(ie); throw iie; } finally { if (traceScope != null) traceScope.close(); } } } // allocate new lock for this thread return rowLockContext.newLock(); }
具体流程整理如下:
1、利用byte[]类型的入参row构造HashedBytes类型表示的行,即rowKey;
2、利用rowKey创建行锁上下文实例,并指定为行rowKey和当前线程拥有;
3、循环:
3.1、将rowKey与行锁上下文的对应关系添加到Region的数据结构lockedRows中,可能出现以下几种情况:
3.1.1、如果lockedRows中之前不存在对应行rowKey的数据,说明该行当前没有被任何线程锁住,使用这个新创建的上下文rowLockContext,跳出循环并返回,说 明当前行可用;
3.1.2、如果该行已经被当前线程锁住,复用当前线程之前创建的行锁上下文实例,并赋值给rowLockContext,跳出循环并返回,说明当前行可用;
3.1.3、如果该行被其他线程锁住,如果入参确定不需要等待锁的获取,直接返回null,否则重复循环,直到等待rowLockWaitDuration时间后,如果还未获得行锁, 直接抛出异常。
至于都是哪些地方需要获取行锁,在以后各种数据读写流程中再做分析吧~