码迷,mamicode.com
首页 > 其他好文 > 详细

谈论高并发(七)几个自旋锁的实现(二)

时间:2015-12-12 09:40:37      阅读:188      评论:0      收藏:0      [点我收藏+]

标签:

谈论高并发(六)几个自旋锁的实现(一) 这篇中实现了两种主要的自旋锁:TASLock和TTASLock,它们的问题是会进行频繁的CAS操作。引发大量的缓存一致性流量,导致锁的性能不好。


对TTASLock的一种改进是BackoffLock,它会在锁高争用的情况下对线程进行回退,降低竞争,降低缓存一致性流量。可是BackoffLock有三个基本的问题:

1. 还是有大量的缓存一致性流量,由于全部线程在同一个共享变量上旋转,每一次成功的获取锁都会产生缓存一致性流量

2. 由于回退的存在,不能及时获取锁释放的信息。存在一个时间差,导致获取锁的时间变长

3. 不能保证无饥饿,有的线程可能一直无法获取锁


这篇会实现2种基于队列的锁,来解决上面提到的三个问题。

基本的思路是将线程组织成一个队列。有4个长处:

1. 每一个线程仅仅须要检查它的前驱线程的状态,将自旋的变量从一个分散到多个,降低缓存一致性流量

2. 能够即使获取锁释放的通知

3. 队列提供了先来先服务的公平性

4. 无饥饿,队列中的每一个线程都能保证被运行到


队列锁分为两类。一类是基于有界队列,一类是基于无界队列。


先看一下基于有界队列的队列锁。 ArrayLock有3个特点:

1. 基于一个volatile数组来组织线程

2. 通过一个原子变量tail来表示对尾线程

3. 通过一个ThreadLocal变量给每一个线程一个索引號。表示它位于队列的哪个位置。


package com.test.lock;

import java.util.concurrent.atomic.AtomicInteger;

/**
 * 有界队列锁。使用一个volatile数组来组织线程
 * 缺点是得预先知道线程的规模n。全部线程获取同一个锁的次数不能超过n
 * 如果L把锁,那么锁的空间复杂度为O(Ln)
 * **/
public class ArrayLock implements Lock{
	// 使用volatile数组来存放锁标志, flags[i] = true表示能够获得锁
	private volatile boolean[] flags;
	
	// 指向新增加的节点的后一个位置
	private AtomicInteger tail;
	
	// 总容量
	private final int capacity;
	
	private ThreadLocal<Integer> mySlotIndex = new ThreadLocal<Integer>(){
		 protected Integer initialValue() {
		        return 0;
		 }
	};
	
	public ArrayLock(int capacity){
		this.capacity = capacity;
		flags = new boolean[capacity];
		tail = new AtomicInteger(0);
		// 默认第一个位置可获得锁
		flags[0] = true;
	}
	
	@Override
	public void lock() {
		int slot = tail.getAndIncrement() % capacity;
		mySlotIndex.set(slot);
		// flags[slot] == true 表示获得了锁。 volatile变量保证锁释放及时通知
		 while(!flags[slot]){
			
		 }
	 }

	 @Override
	 public void unlock() {
	       int slot = mySlotIndex.get();
	       flags[slot] = false;
	       flags[(slot + 1) % capacity] = true;
	}
<pre name="code" class="java">
        public String toString(){
           return "ArrayLock";
        }

 }



我们能够看到有界队列锁的缺点是:

1. 它必须知道线程的规模数。对于同一把锁假设线程获取的次数超过了n会出现线程状态被覆盖的问题

2. 空间复杂度是O(Ln)

3. 对于共享的volatile数组来保存线程获取锁的状态。仍然可能存在缓存一致性。我们知道CPU读取一次内存时,会读满数据总线的位长,比方64位总线。一次读取64位长度的数据。

那么对于boolean类型的数组。boolean长度是1个字节,那么一次读取能读到8个boolean变量。而快速缓存的一个缓存块的长度也是64位,也就是说一个缓存块上能够保存8个boolean变量,所以假设一次CAS操作改动了一个变量导致一个缓存块无效,它实际上可能导致8个变量失效。

解决的方法是把变量以8个长度为单位分散。比方flag[0] = thread1  flag[8] = thread2。这种问题是消耗的空间更大。


无界队列锁能够克服有界队列锁的几个问题。

1. 它使用链表来取代数组,实现无界队列

2. 使用两个ThreadLocal变量表示指针。一个指向自己的节点,一个指向前一个节点

3. 使用一个原子引用变量指向队尾

4. 空间复杂度减少,假设有L把锁。n个线程。每一个线程仅仅获取一把锁,那么空间复杂度为O(L + n)

5. 对同一个锁,一个线程能够多次获取而不添加空间复杂度

6. 当线程结束后。GC会自己主动回收内存


package com.test.lock;

import java.util.concurrent.atomic.AtomicReference;

/**
 * 无界队列锁,使用一个链表来组织线程
 * 如果L把锁。n个线程,那么锁的空间复杂度为O(L+n)
 * **/
public class CLHLock implements Lock{
	// 原子变量指向队尾
	private AtomicReference<QNode> tail;
	// 两个指针,一个指向自己的Node,一个指向前一个Node
	ThreadLocal<QNode> myNode;
	ThreadLocal<QNode> myPreNode;
	
	public CLHLock(){
		tail = new AtomicReference<QNode>(new QNode());
		myNode = new ThreadLocal<QNode>(){
			protected QNode initialValue(){
				return new QNode();
			}
		};
		myPreNode = new ThreadLocal<QNode>(){
			protected QNode initialValue(){
				return null;
			}
		};
	}
	
	@Override
	public void lock() {
		QNode node = myNode.get();
		node.lock = true;
		// CAS原子操作,保证原子性
		QNode preNode = tail.getAndSet(node);
		myPreNode.set(preNode);
		// volatile变量。能保证锁释放及时通知
		// 仅仅对前一个节点的状态自旋,降低缓存一致性流量
		while(preNode.lock){
			
		}
	}

	@Override
	public void unlock() {
		QNode node = myNode.get();
		node.lock = false;
		// 把myNode指向preNode。目的是保证同一个线程下次还能使用这个锁,由于myNode原来指向的节点有它的后一个节点的preNode引用
		// 防止这个线程下次lock时myNode.get获得原来的节点
		myNode.set(myPreNode.get());
	}
	
	public static class QNode {
		volatile boolean lock;
	}

        public String toString(){
           return "CLHLock";
        }
 }

以下我们从正确性和平均获取锁的时间上来測试这两种锁。

我们设计一个測试用例来验证正确性: 使用50个线程对一个volatile变量++操作,因为volatile变量++操作不是原子的,在不加锁的情况下。可能同一时候有多个线程同一时候对voaltile变量++, 终于的结果是无法预測的。

然后使用这两种锁,先获取锁再volatile变量++。因为volatile变量会防止重排序。并能保证可见性,我们能够确定假设锁是正确获取的,也就是说同一时刻仅仅有一个线程对volatile变量++,那么结果肯定是顺序的1到50。

先看不加锁的情况

package com.test.lock;

public class Main {
	//private static Lock lock = new ArrayLock(150);
	
	private static Lock lock = new CLHLock();
	
	//private static TimeCost timeCost = new TimeCost(new TTASLock());
	
	private static volatile int value = 0;
	public static void method(){
		//lock.lock();
		System.out.println("Value: " + ++value);
		//lock.unlock();
	}
	
	public static void main(String[] args) {
		for(int i = 0; i < 50; i ++){
		   Thread t = new Thread(new Runnable(){
	
		   @Override
	              public void run() {
			 method();
		   }
				
	           });
		  t.start();
	        }
	}

}

执行结果: 我们能够看到确实是发生的线程同一时候对volatile变量++的操作,结果是无法预料的

Value: 1
Value: 1
Value: 2
Value: 3
Value: 4
Value: 5
Value: 6
Value: 7
Value: 8
Value: 9
Value: 10
Value: 11
Value: 13
Value: 12
Value: 14
Value: 15
Value: 16
Value: 17
Value: 18
Value: 19
Value: 20
Value: 21
Value: 22
Value: 23
Value: 24
Value: 25
Value: 26
Value: 27
Value: 28
Value: 29
Value: 30
Value: 31
Value: 32
Value: 33
Value: 34
Value: 35
Value: 36
Value: 37
Value: 38
Value: 37
Value: 39
Value: 40
Value: 41
Value: 42
Value: 43
Value: 44
Value: 45
Value: 46
Value: 47
Value: 48
Value: 50

使用有界队列锁:

package com.test.lock;

public class Main {
	private static Lock lock = new ArrayLock(100);
	
	//private static Lock lock = new CLHLock();
	
	//private static TimeCost timeCost = new TimeCost(new TTASLock());
	
	private static volatile int value = 0;
	public static void method(){
		lock.lock();
		System.out.println("Value: " + ++value);
		lock.unlock();
	}
	
	public static void main(String[] args) {
		for(int i = 0; i < 50; i ++){
			Thread t = new Thread(new Runnable(){
	
				@Override
				public void run() {
					method();
				}
				
			});
			t.start();
		}
	}

}

执行结果是1到50的顺序自增。说明锁保证了同一时刻仅仅有一个线程在对volatile变量++,是正确的

Value: 1
Value: 2
Value: 3
Value: 4
Value: 5
Value: 6
Value: 7
Value: 8
Value: 9
Value: 10
Value: 11
Value: 12
Value: 13
Value: 14
Value: 15
Value: 16
Value: 17
Value: 18
Value: 19
Value: 20
Value: 21
Value: 22
Value: 23
Value: 24
Value: 25
Value: 26
Value: 27
Value: 28
Value: 29
Value: 30
Value: 31
Value: 32
Value: 33
Value: 34
Value: 35
Value: 36
Value: 37
Value: 38
Value: 39
Value: 40
Value: 41
Value: 42
Value: 43
Value: 44
Value: 45
Value: 46
Value: 47
Value: 48
Value: 49
Value: 50

使用无界队列锁的情况也是正确的。因为篇幅原因这里就不帖代码了。


再看平均获取锁的时间。

package com.test.lock;

public class Main {
	private static Lock lock = new TimeCost(new CLHLock());
	
	//private static Lock lock = new CLHLock();
	
	//private static TimeCost timeCost = new TimeCost(new TTASLock());
	
	private static volatile int value = 0;
	public static void method(){
		lock.lock();
		//System.out.println("Value: " + ++value);
		lock.unlock();
	}
	
	public static void main(String[] args) {
		for(int i = 0; i < 100; i ++){
			Thread t = new Thread(new Runnable(){
	
				@Override
				public void run() {
					method();
				}
				
			});
			t.start();
		}
	}

}

在100个线程并发的情况下。

ArrayLock获取锁的平均时间是: 719550 ns

CLHLock平均时间获得锁:  488577 ns


查看,在多个共享变量队列自旋锁的情况下,减少的一致性流量。比TASLock和TTASLock 提高该程序的性能。和CLHLock比ArrayLock更好的扩展性和性能,这是一个非常不错的自旋锁执行。


谈论高并发(七)几个自旋锁的实现(二)

标签:

原文地址:http://www.cnblogs.com/bhlsheji/p/5040737.html

(0)
(0)
   
举报
评论 一句话评论(0
登录后才能评论!
© 2014 mamicode.com 版权所有  联系我们:gaon5@hotmail.com
迷上了代码!