马士兵老师高并发编程同步容器

Posted 2021-01-17 RDeduction

手写固定同步容器

写一个固定容量同步容器，拥有put和get方法，以及getCount方法，能够支持2个生产者线程以及10个消费者线程的阻塞调用。

使用wait与notify

思路：使用一个集合来当做生产或者消费的中转站，然后每当生产或者消费的时刻都判断集合的容量，如果不满足条件那么就对这种操作进行阻塞也就是wait同时notify其它的所有线程。当其它线程启动之后也会遇到“不合格的线程”这时候也会阻塞，直到合格的线程进行执行。

核心代码：

public class MyContainer1<T> {
	final private LinkedList<T> lists = new LinkedList<>();
	final private int MAX = 10; //最多10个元素
	private int count = 0;
	
	
	public synchronized void put(T t) {
		while(lists.size() == MAX) { //想想为什么用while而不是用if？
			try {
				this.wait(); //effective java
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
		}
		
		lists.add(t);
		++count;
		this.notifyAll(); //通知消费者线程进行消费
	}
	
	public synchronized T get() {
		T t = null;
		while(lists.size() == 0) {
			try {
				this.wait();
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
		}
		t = lists.removeFirst();
		count --;
		this.notifyAll(); //通知生产者进行生产
		return t;
	}
	
	public static void main(String[] args) {
		MyContainer1<String> c = new MyContainer1<>();
		//启动消费者线程
		for(int i=0; i<10; i++) {
			new Thread(()->{
				for(int j=0; j<5; j++) System.out.println(c.get());
			}, "c" + i).start();
		}
		
		try {
			TimeUnit.SECONDS.sleep(2);
		} catch (InterruptedException e) {
			e.printStackTrace();
		}
		
		//启动生产者线程
		for(int i=0; i<2; i++) {
			new Thread(()->{
				for(int j=0; j<25; j++) c.put(Thread.currentThread().getName() + " " + j);
			}, "p" + i).start();
		}
	}
}

注意事项：

为什么是while？

有下面的一个场景，当消费者消费的阈值不满足条件那么它将会被wait阻塞。此时还有其的9个消费者也处于阻塞状态。当notifyAll的时候我们希望的是让生产者来生产元素，但是这时候被唤醒的消费者会继续去消费，代码会从wait()处直接向下执行。如果是一个if判断，再加上这时的集合中没有元素那么此时一定会出异常。但是如果使用的是while的话那么被唤醒的消费者就会循环检测发现不满足条件就继续阻塞。整个程序顺利进行。

 

使用signalAll唤醒对应条件的线程

signalAll与ReentrantLock共用达到只唤醒对应条件的线程。比如说当生产者生产的元素超过阈值的时候他就会调用signalAll这时所有的消费者被唤醒而所有的生产者则不受影响。这样就可以避免唤醒不必要的线程节省资源。

此处注意要给每一种线程都定义一个Condition，在上锁的时候就只用这个Condition的锁去锁定对应的线程。具体代码如下：

public class MyContainer2<T> {
	final private LinkedList<T> lists = new LinkedList<>();
	final private int MAX = 10; //最多10个元素
	private int count = 0;
	
	private Lock lock = new ReentrantLock();
	private Condition producer = lock.newCondition();
	private Condition consumer = lock.newCondition();
	
	public void put(T t) {
		try {
			lock.lock();
			while(lists.size() == MAX) { 
				producer.await();
			}
			
			lists.add(t);
			++count;
			consumer.signalAll(); //通知消费者线程进行消费
		} catch (InterruptedException e) {
			e.printStackTrace();
		} finally {
			lock.unlock();
		}
	}
	
	public T get() {
		T t = null;
		try {
			lock.lock();
			while(lists.size() == 0) {
				consumer.await();
			}
			t = lists.removeFirst();
			count --;
			producer.signalAll(); //通知生产者进行生产
		} catch (InterruptedException e) {
			e.printStackTrace();
		} finally {
			lock.unlock();
		}
		return t;
	}
	
	public static void main(String[] args) {
		MyContainer2<String> c = new MyContainer2<>();
		//启动消费者线程
		for(int i=0; i<10; i++) {
			new Thread(()->{
				for(int j=0; j<5; j++) {
					System.out.println(c.get());
//					System.out.println("c");
				}
			}, "c" + i).start();
		}
		
		try {
			TimeUnit.SECONDS.sleep(2);
		} catch (InterruptedException e) {
			e.printStackTrace();
		}
		
		//启动生产者线程
		for(int i=0; i<2; i++) {
			new Thread(()->{
				for(int j=0; j<25; j++) c.put(Thread.currentThread().getName() + " " + j);
			}, "p" + i).start();
		}
	}
}

售票员案例

 * 场景

 * 有N张火车票，每张票都有一个编号

 * 同时有10个窗口对外售票

 * 请写一个模拟程序

 *

 * 分析下面的程序可能会产生哪些问题？

 * 重复销售？超量销售？

 

Solution1：使用线程不安全的集合而且不上锁

public class TicketSeller1 {
	static List<String> tickets = new ArrayList<>();
	
	static {
		for(int i=0; i<1000; i++) tickets.add("票编号：" + i);
	}
	
	public static void main(String[] args) {
		for(int i=0; i<10; i++) {
			new Thread(()->{
				while(tickets.size() > 0) {
                   try {
						Thread.sleep(10);
					} catch (Exception e) {
						e.printStackTrace();
					}				
					System.out.println("销售了--" + tickets.remove(0));
				}
			}).start();
		}
	}
}

为了使得问题呈现的效果明显我们加上了睡眠时间。此程序会出现的问题：

总的来说会有两个点出现问题：

①程序逻辑的线程不安全，有可能有多个线程涌入while循环中这是造成并发问题的根本。这个原因会导致在size为1的时候涌入很多的线程进而执行多次删除操作下标越界。

②集合的线程不安全，remove的方法本来就不是线程安全的。为了说明问题我，我们把remove方法放大：

可以看出如果两个线程同时执行remove方法的话，由于index一样所以他们的remove的返回值就会得到同一个oldValue。也就是重复卖出。

Solution2：使用集合Vector但是代码逻辑不加锁

代码上的逻辑与solution1是一样的，但是Vector集合是线程安全的，所以它只会出现程序逻辑不安全带来的并发问题。也就是会出现有可能有多个线程涌入while循环中这是造成并发问题的根本。这个原因会导致在size为1的时候涌入很多的线程进而执行多次删除操作下标越界。但是绝对不会出现卖出同一张票的情况。我们把remove的代码放大：

这是一个同步的方法，每一个线程过来如果得不到锁得话都会陷入等待。虽然都是remove(0)但是当下一个线程来到的时候0位置已经是一个全新的元素。

 

Solution3：给代码逻辑上锁使用线程不安全的集合

不多说了无论如何都可以防止线程安全问题，因为在Solution1中已经提到过了代码的并发问题是一切问题的原因。直接上代码：

public class TicketSeller3 {
	static List<String> tickets = new LinkedList<>();
	
	
	static {
		for(int i=0; i<1000; i++) tickets.add("票 编号：" + i);
	}
	
	public static void main(String[] args) {
		
		for(int i=0; i<10; i++) {
			new Thread(()->{
				while(true) {
					synchronized(tickets) {
						if(tickets.size() <= 0) break;
						try {
							TimeUnit.MILLISECONDS.sleep(10);
						} catch (InterruptedException e) {
							e.printStackTrace();
						}
						System.out.println("销售了--" + tickets.remove(0));
					}
				}
			}).start();
		}
	}
}

Solution4使用并发集合不加锁

首先说并发的集合是线程安全的，而且效率较高因为使用了局部锁。这样的话只在取值的时候加了锁，而且如果是以下标来取值的话还可以同时取走多个地方的值这样的话效率大大提高。而且这里使用一种取值然后再判断的逻辑巧妙的避免了下表越界的错误，而前面的案例中都是先判断再取值这样就造成了线程不安全：

public class TicketSeller4 {
	static Queue<String> tickets = new ConcurrentLinkedQueue<>();
	
	
	static {
		for(int i=0; i<1000; i++) tickets.add("票 编号：" + i);
	}
	
	public static void main(String[] args) {
		
		for(int i=0; i<10; i++) {
			new Thread(()->{
				while(true) {
					String s = tickets.poll();
					if(s == null) break;
					else System.out.println("销售了--" + s);
				}
			}).start();
		}
	}
}

高并发集合简介

ConcurrectHashMap：使用了局部锁，也就是细粒度的锁来提高并发效果。

ConcurrectSkipListMap：使用了局部锁，但是结果也排序的map集合。对比TreeMap一个元素排序的map集合。

CopyOnWriteArrayList：读取时不加锁，但是写的时候回拷贝原有的数据然后对拷贝的数据进行操作最后将指针指向修改过的集合。这个集合适用于读操作远远大于写操作的情况。

BlockingQueue：阻塞队列，当队列中没有元素的时候就会对取元素产生阻塞，当队列中满元素的时候就会对添加元素产生阻塞。而且不允许添加null的值而且在取值与添加值的情况下都会加锁，换句话说它是一个线程安全的集合。以下为部分源码：

DelayQueue：执行定时任务，他的内部会装有很多的task接受的task都实现了Delay接口，因此task内部也就维护了一个conpareTo的方法，如果按照时间排序的话那么就能够实现任务的定时执行。

public class T07_DelayQueue {

	static BlockingQueue<MyTask> tasks = new DelayQueue<>();

	static Random r = new Random();
	
	static class MyTask implements Delayed {
		long runningTime;
		
		MyTask(long rt) {
			this.runningTime = rt;
		}

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

		@Override
		public long getDelay(TimeUnit unit) {
			return unit.convert(runningTime - System.currentTimeMillis(), TimeUnit.MILLISECONDS);
		}
		
		@Override
		public String toString() {
			return "" + runningTime;
		}
	}

	public static void main(String[] args) throws InterruptedException {
		long now = System.currentTimeMillis();
		MyTask t1 = new MyTask(now + 1000);
		MyTask t2 = new MyTask(now + 2000);
		MyTask t3 = new MyTask(now + 1500);
		MyTask t4 = new MyTask(now + 2500);
		MyTask t5 = new MyTask(now + 500);		
		tasks.put(t1);
		tasks.put(t2);
		tasks.put(t3);
		tasks.put(t4);
		tasks.put(t5);
		
		System.out.println(tasks);
		
		for(int i=0; i<5; i++) {
			System.out.println(tasks.take());
		}
	}
}

TransferQueue：当有消费者的话那么就直接将生产出来的元素交给消费者，但是如果没有消费者的话就会将生产的元素放到队列中。当队列中的元素消耗完的时候消费者就会阻塞。