package com.brycen.concurrency03.collections.myqueue;
public class UnThreadSafeQueue<E> {
	//定義Node節(jié)點(diǎn)
	private static class Node<E> {
		private E element;//節(jié)點(diǎn)內(nèi)存儲的元素
		private Node<E> next;//下一個節(jié)點(diǎn)
		public Node(E element, Node<E> next) {
			super();
			this.element = element;
			this.next = next;
		}
		public E getElement() {
			return element;
		}
		public void setElement(E element) {
			this.element = element;
		}
		public Node<E> getNext() {
			return next;
		}
		public void setNext(Node<E> next) {
			this.next = next;
		}
		@Override
		public String toString() {
			return (element == null) ? "" : element.toString();
		}
	}
	//定義隊列的頭和尾
	private Node<E> head, last;
	//初始化隊列長度為0
	private int size = 0;
	public int size() {
		return size;
	}
	public boolean isEmpty() {
		return size == 0;
	}
	//返回第一個元素
	public E peekFirst() {
		return head.element == null ? null : head.getElement();
	}
	//返回最后一個元素
	public E peekLast() {
		return last.element == null ? null : last.getElement();
	}
	//在尾部添加元素
	public void addLast(E element) {
		Node<E> newNode = new Node<E>(element, null);
		//如果為0，則代表隊列沒有元素
		if (size == 0) {
			head = newNode;
		}else {
			//隊列有元素，則將最后一個元素的下一個值設(shè)置為新的元素
			last.setNext(newNode);
		}
		//新元素賦值給last
		last = newNode;
		//隊列長度+1
		size++;
	}
	//移除并返回第一個元素
	public E removeFirst() {
		//如果為null，直接返回null
		if (isEmpty())
			return null;
		//拿到第一個Node中的元素
		E result = head.getElement();
		//獲取第一個Node中的下一個元素并賦值給head
		head = head.getNext();
		//隊列長度-1
		size--;
		//判斷隊列是否為null，如果為null，需要將last置為null
		if (size==0)
			last = null;
		return result;
	}
	public static void main(String[] args) {
		UnThreadSafeQueue<String> queue = new UnThreadSafeQueue<String>();
		queue.addLast("Hello");
		queue.addLast("World");
		queue.addLast("Java");
		System.out.println(queue.removeFirst());
		System.out.println(queue.removeFirst());
		System.out.println(queue.removeFirst());
	}
}

package com.brycen.concurrency03.collections.myqueue;
import java.util.Random;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Executors;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;
import java.util.stream.IntStream;
public class LockFreeQueue<E> {
	//定義頭和尾的原子性節(jié)點(diǎn)
	private AtomicReference<Node<E>> head, last;
	//定義原子性size
	private AtomicInteger size = new AtomicInteger(0);
	//初始化隊列，將頭和尾都指向一個null的節(jié)點(diǎn)
	public LockFreeQueue() {
		Node<E> node = new Node<E>(null);
		head = new AtomicReference<Node<E>>(node);
		last = new AtomicReference<Node<E>>(node);
	}
	//定義節(jié)點(diǎn)類
	private static class Node<E> {
		E element;
		//需要volatile，因為防止在next賦值的時候發(fā)生重排序，并且需要對其他線程可見
		volatile Node<E> next;
		public Node(E element) {
			this.element = element;
		}
		@Override
		public String toString() {
			return element == null ? null : element.toString();
		}
	}
	//添加元素到隊尾
	public void addLast(E element) {
		//元素不允許為null
		if (element == null)
			throw new NullPointerException("The null element not allow");
		//新建一個新節(jié)點(diǎn)
		Node<E> newNode = new Node<E>(element);
		//getAndSet操作為原子性操作，先獲取last的節(jié)點(diǎn)再將新的節(jié)點(diǎn)賦值給last
		Node<E> oldNode = last.getAndSet(newNode);
		//將舊節(jié)點(diǎn)的next指向新的節(jié)點(diǎn)
		oldNode.next = newNode;
		//隊列長度+1
		size.incrementAndGet();
	}
	//移除并返回隊首元素
	public E removeFirst() {
		//因為隊首節(jié)點(diǎn)是存在的，但是他可能沒有下一個節(jié)點(diǎn)，所以需要一個valueNode來判斷
		Node<E> headNode, valueNode;
		do {
			//獲取到隊首節(jié)點(diǎn)
			headNode = head.get();
			//判斷下一個節(jié)點(diǎn)是否為null
			valueNode = headNode.next;
		//當(dāng)valueNode不為null，并且headNode不等于隊列的head節(jié)點(diǎn)時，代表該元素被別的線程拿走的，需要重新獲取。
		//當(dāng)headNode等于隊列的head時則代表頭元素沒有被其他元素拿走，并將head節(jié)點(diǎn)替換為valueNode。
		} while (valueNode != null && !head.compareAndSet(headNode, valueNode));
		E result = valueNode != null ? valueNode.element : null;
		//valueNode的元素被拿走了，所有將其置為null
		if (valueNode != null) {
			valueNode.element = null;
		}
		//隊列長度-1
		size.decrementAndGet();
		return result;
	}
	public static void main(String[] args) throws InterruptedException {
		//創(chuàng)建線程池
		ThreadPoolExecutor threadPool = (ThreadPoolExecutor) Executors.newFixedThreadPool(10);
		//實例化隊列
		LockFreeQueue<String> queue = new LockFreeQueue<String>();
		//該map用于檢查該隊列是否是線程安全的，利用其key不能重復(fù)來判斷
		ConcurrentHashMap<String, Object> map = new ConcurrentHashMap<String, Object>();
		//隨機(jī)數(shù)
		Random random = new Random(System.currentTimeMillis());
		//創(chuàng)建5個寫runnable
		IntStream.range(0, 5).boxed().map(i -> (Runnable) () -> {
			int count = 0;
			//每個runnable往隊列插入10個元素
			while (count++<10) {
				//這里值用系統(tǒng)的納秒+隨機(jī)數(shù)+count，以防止重復(fù)影響map集合對隊列線程安全的判斷
				queue.addLast(System.nanoTime()+":"+random.nextInt(10000)+":"+count);
			}
			//提交任務(wù)
		}).forEach(threadPool::submit);
		//創(chuàng)建5個讀runnable
		IntStream.range(0, 5).boxed().map(i -> (Runnable) () -> {
			int count = 10;
			//每個runnable讀10個元素
			while (count-->0) {
				//休眠
				try {
					TimeUnit.MILLISECONDS.sleep(20);
				} catch (InterruptedException e) {
					e.printStackTrace();
				}
				//移除隊列中的隊首元素
				String result = queue.removeFirst();
				//輸出
				System.out.println(result);
				//將該元素加入map中，來判斷隊列中真實存入的元素個數(shù)
				map.put(result, new Object());
			}
			//提交任務(wù)
		}).forEach(threadPool::submit);
		//關(guān)閉線程池
		threadPool.shutdown();
		//等待1小時候強(qiáng)制關(guān)閉線程池
		threadPool.awaitTermination(1, TimeUnit.HOURS);
		//打印map中的元素個數(shù)
		System.out.println(map.size());
	}
}