《Java 7 并发编程指南》学习概要 (4) 并发集合

1、ConcurrentLinkedDeque  非阻塞并发双端链表

getFirst()和getLast()：这些方法将分别返回列表的第一个和最后一个元素。它们不会从列表删除返回的元素。如果列表为空，这些方法将抛出NoSuchElementExcpetion异常。

peek()、peekFirst()和peekLast()：这些方法将分别返回列表的第一个和最后一个元素。它们不会从列表删除返回的元素。如果列表为空，这些方法将返回null值。

remove()、removeFirst()、 removeLast()：这些方法将分别返回列表的第一个和最后一个元素。它们将从列表删除返回的元素。如果列表为空，这些方法将抛出NoSuchElementExcpetion异常。

2、LinkedBlockingDeque  阻塞双端链表

takeFirst() 和takeLast()：这些方法分别返回列表的第一个和最后一个元素。它们从列表删除返回的元素。如果列表为空，这些方法将阻塞线程，直到列表有元素。

getFirst() 和getLast()：这些方法分别返回列表的第一个和最后一个元素。它们不会从列表删除返回的元素。如果列表为空，这些方法将抛出NoSuchElementExcpetion异常。

peek()、peekFirst(),和peekLast()：这些方法分别返回列表的第一个和最后一个元素。它们不会从列表删除返回的元素。如果列表为空，这些方法将返回null值。

poll()、pollFirst()和 pollLast()：这些方法分别返回列表的第一个和最后一个元素。它们从列表删除返回的元素。如果列表为空，这些方法将返回null值。

add()、 addFirst()、addLast()：这些方法分别在第一个位置和最后一个位置上添加元素。如果列表已满（你已使用固定大小创建它），这些方法将抛出IllegalStateException异常。

3、PriorityBlockingQueue  阻塞优先级队列

take()：这个方法返回并删除队列中的第一个元素。如果队列是空的，这个方法将阻塞线程直到队列有元素。

put(E e)：E是用来参数化PriorityBlockingQueue类的类。这个方法将作为参数传入的元素插入到队列中。

peek()：这个方法返回列队的第一个元素，但不删除它。

4、DelayedQueue  

DelayedQueue  可以存储带有激活日期的元素。方法返回或抽取队列的元素将忽略未到期的数据元素。它们对这些方法来说是看不见的。

• clear()：这个方法删除队列中的所有元素。
• offer(E e)：E是代表用来参数化DelayQueue类的类。这个方法插入作为参数传入的元素到队列中。
• peek()：这个方法检索，但不删除队列的第一个元素。
• take()：这具方法检索并删除队列的第一个元素。如果队列中没有任何激活的元素，执行这个方法的线程将被阻塞，直到队列有一些激活的元素。

public class DelayedQueueTest {

	public static void main(String[] args) throws InterruptedException {

		DelayQueue<Event> queue = new DelayQueue<>();
		Thread threads[] = new Thread[5];

		for (int i = 0; i < threads.length; i++) {
			Task task = new Task(i + 1, queue);
			threads[i] = new Thread(task);
		}

		for (int i = 0; i < threads.length; i++) {
			threads[i].start();
		}

		for (int i = 0; i < threads.length; i++) {
			threads[i].join();
		}
		do {
			int counter = 0;
			Event event;
			do {
				event = queue.poll();
				if (event != null)
					counter++;
			} while (event != null);
			System.out.printf("At %s you have read %d events
", new Date(),
					counter);
			TimeUnit.MILLISECONDS.sleep(500);
		} while (queue.size() > 0);
	}

}

class Event implements Delayed {
	private Date startDate;

	public Event(Date startDate) {
		this.startDate = startDate;
	}

	@Override
	public int compareTo(Delayed o) {
		long result = this.getDelay(TimeUnit.NANOSECONDS)
				- o.getDelay(TimeUnit.NANOSECONDS);
		if (result < 0) {
			return -1;
		} else if (result > 0) {
			return 1;
		}
		return 0;
	}

	public long getDelay(TimeUnit unit) {
		Date now = new Date();
		long diff = startDate.getTime() - now.getTime();
		return unit.convert(diff, TimeUnit.MILLISECONDS);
	}
}

class Task implements Runnable {
	private int id;
	private DelayQueue<Event> queue;

	public Task(int id, DelayQueue<Event> queue) {
		this.id = id;
		this.queue = queue;
	}

	@Override
	public void run() {
		Date now = new Date();
		Date delay = new Date();
		delay.setTime(now.getTime() + (id * 1000));
		System.out.printf("Thread %s: %s
", id, delay);

		for (int i = 0; i < 100; i++) {
			Event event = new Event(delay);
			queue.add(event);
		}
	}

}

5、ConcurrentSkipListMap

 ConcurrentSkipListMap提供了一种线程安全的并发访问的排序映射表。内部是SkipList（跳表）结构实现，在理论上能够在O(log(n))时间内完成查找、插入、删除操作。

 SkipList是一种红黑树的替代方案，由于SkipList与红黑树相比无论从理论和实现都简单许多，所以得到了很好的推广。SkipList是基于一种统计学原理实现的，有可能出现最坏情况，即查找和更新操作都是O(n)时间复杂度，但从统计学角度分析这种概率极小。

ConcurrentSkipListMap 实现了非阻塞列表且拥有ConcurrentNavigableMap的行为。在内部实现中，它使用Skip List来存储数据。Skip List是基于并行列表的数据结构，它允许我们获取类似二叉树的效率。使用它，你可以得到一个排序的数据结构，这比排序数列使用更短的访问时间来插入、搜索和删除元素。
当你往map中插入数据时，它使用key来排序它们，所以，所有元素将是有序的。除了返回具体的元素，这个类也提供了获取map的子map的方法。

public class ConcurrentSkipListMapTest {

	public static void main(String[] args) {

		ConcurrentSkipListMap<String, Contact> map = new ConcurrentSkipListMap<>();
		Thread threads[] = new Thread[25];
		int counter = 0;
		for (char i = 'A'; i < 'Z'; i++) {
			MyTask task = new MyTask(map, String.valueOf(i));
			threads[counter] = new Thread(task);
			threads[counter].start();
			counter++;
		}
		for (int i = 0; i < 25; i++) {
			try {
				threads[i].join();
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
		}
		System.out.printf("Main: Size of the map: %d
", map.size());
		Map.Entry<String, Contact> element = map.firstEntry();
		Contact contact = element.getValue();
		System.out.printf("Main: First Entry: %s: %s
", contact.getName(),
				contact.getPhone());

		System.out.printf("Main: Submap from A1996 to B1002: 
");
		ConcurrentNavigableMap<String, Contact> submap = map.subMap("A1996",
				"B1002");
		do {
			element = submap.pollFirstEntry();
			if (element != null) {
				contact = element.getValue();
				System.out.printf("%s: %s
", contact.getName(),
						contact.getPhone());
			}
		} while (element != null);
	}

}

class Contact {
	private String name;
	private String phone;

	public Contact(String name, String phone) {
		this.name = name;
		this.phone = phone;
	}

	public String getName() {
		return name;
	}

	public String getPhone() {
		return phone;
	}
}

class MyTask implements Runnable {
	private ConcurrentSkipListMap<String, Contact> map;
	private String id;

	public MyTask(ConcurrentSkipListMap<String, Contact> map, String id) {
		this.id = id;
		this.map = map;
	}

	@Override
	public void run() {
		for (int i = 0; i < 1000; i++) {
			Contact contact = new Contact(id, String.valueOf(i + 1000));
			map.put(id + contact.getPhone(), contact);
		}
	}

}

6、ThreadLocalRandom

使用线程局部变量。每个线程希望以不同的生成器生成随机数，但它们是来自相同类的管理，这对程序员是透明的。

7、AtomicLong 、AtomicBoolean、AtomicInteger、AtomicReference

当一个线程正在操作一个原子变量时，即使其他线程也想要操作这个变量，类的实现中含有一个检查那步骤操作是否完成的机制。 基本上，操作获取变量的值，改变本地变量值，然后尝试以新值代替旧值。如果旧值还是一样，那么就改变它。如果不一样，方法再次开始操作。这个操作称为 Compare and Set（校对注：简称CAS，比较并交换的意思）。原子变量不使用任何锁或者其他同步机制来保护它们的值的访问。他们的全部操作都是基于CAS操作。它保证几个线程可以同时操作一个原子对象也不会出现数据不一致的错误，并且它的性能比使用受同步机制保护的正常变量要好。

public class AtomicLongTest {

	public static void main(String[] args) {
		Account account = new Account();
		account.setBalance(1000);

		Company company = new Company(account);
		Thread companyThread = new Thread(company);

		Bank bank = new Bank(account);
		Thread bankThread = new Thread(bank);

		System.out.printf("Account : Initial Balance: %d
",
				account.getBalance());

		companyThread.start();
		bankThread.start();

		try {
			companyThread.join();
			bankThread.join();
			System.out.printf("Account : Final Balance: %d
",
					account.getBalance());
		} catch (InterruptedException e) {
			e.printStackTrace();
		}

	}

}

class Account {

	private AtomicLong balance;

	public Account() {
		balance = new AtomicLong();
	}

	public long getBalance() {
		return balance.get();
	}

	public void setBalance(long balance) {
		this.balance.set(balance);
	}

	public void addAmount(long amount) {
		this.balance.getAndAdd(amount);
	}

	public void subtractAmount(long amount) {
		this.balance.getAndAdd(-amount);
	}
}

class Company implements Runnable {

	private Account account;

	public Company(Account account) {
		this.account = account;
	}

	@Override
	public void run() {
		for (int i = 0; i < 10; i++) {
			account.addAmount(1000);
		}
	}
}

class Bank implements Runnable {

	private Account account;

	public Bank(Account account) {
		this.account = account;
	}

	@Override
	public void run() {
		for (int i = 0; i < 10; i++) {
			account.subtractAmount(1000);
		}
	}

}

8、AtomicIntegerArray、AtomicLongArray 

public class Incrementer implements Runnable {

	// 2.声明一个私有 AtomicIntegerArray 属性，名为 vector，用来储存一个整数 array。
	private AtomicIntegerArray vector;

	// 3.实现类的构造函数，初始化它的属性值。
	public Incrementer(AtomicIntegerArray vector) {
		this.vector = vector;
	}

	// 4.实现 run() 方法。使用 getAndIncrement() 方操作array里的所有元素。
	@Override
	public void run() {
		for (int i = 0; i < vector.length(); i++) {
			vector.getAndIncrement(i);
		}
	}

	public static void main(String[] args) {

		// 10.声明一个常量，名为 THREADS，分配它的值为 100。创建一个有1，000个元素的 AtomicIntegerArray 对象。
		final int THREADS = 100;
		AtomicIntegerArray vector = new AtomicIntegerArray(1000);

		// 11. 创建一个 Incrementer 任务来操作之前创建的原子 array。
		Incrementer incrementer = new Incrementer(vector);

		// 12.创建一个 Decrementer 任务来操作之前创建的原子 array。
		Decrementer decrementer = new Decrementer(vector);

		// 13.创建2个array 分别存储 100 个Thread 对象。
		Thread threadIncrementer[] = new Thread[THREADS];
		Thread threadDecrementer[] = new Thread[THREADS];

		// 14.创建并运行 100 个线程来执行 Incrementer 任务和另外 100 个线程来执行 Decrementer
		// 任务。把线程储存入之前创建的arrays内。
		for (int i = 0; i < THREADS; i++) {
			threadIncrementer[i] = new Thread(incrementer);
			threadDecrementer[i] = new Thread(decrementer);

			threadIncrementer[i].start();
			threadDecrementer[i].start();
		}
		// 15.使用 join() 方法来等待线程的完结。
		for (int i = 0; i < 100; i++) {
			try {
				threadIncrementer[i].join();
				threadDecrementer[i].join();
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
		}
		// 16.把原子array里非0的元素写入操控台。使用 get() 方法来获取原子 array 元素。
		for (int i = 0; i < vector.length(); i++) {
			if (vector.get(i) != 0) {
				System.out.println("Vector[" + i + "] : " + vector.get(i));
			}
		}

		// 17.在操控台写个信息表明例子结束。
		System.out.println("Main: End of the example");
	}
}

class Decrementer implements Runnable {

	// 6.声明一个私有 AtomicIntegerArray 属性，名为 vector，用来储存一个整数 array。
	private AtomicIntegerArray vector;

	// 7.实现类的构造函数，初始化它的属性值。
	public Decrementer(AtomicIntegerArray vector) {
		this.vector = vector;
	}

	// 8.实现 run() 方法。使用 getAndDecrement() 方法操作array里的所有元素。
	@Override
	public void run() {
		for (int i = 0; i < vector.length(); i++) {
			vector.getAndDecrement(i);
		}
	}
}

9、ConcurrentHashMap

10、CopyOnWriteArrayList