带回收功能的lockfree queue的java实现

谁践踏了优雅 2022-08-19 08:29 80阅读 0赞

/*
     * LockFreeQueueRecycle.java
     *
     
     *
     * From "Multiprocessor Synchronization and Concurrent Data Structures",
     * modify by juneshen ,base on the work of Maurice Herlihy and Nir Shavit.
     * 
     */
    
    package june.shen;
    
    
    import java.util.concurrent.atomic.AtomicStampedReference;
    
    public class LockFreeQueueRecycle<T> {
      private AtomicStampedReference<Node> head;
      private AtomicStampedReference<Node> tail;
      ThreadLocal<Node> freeList = new ThreadLocal<Node>() {
        protected Node initialValue() { return null; };
      };
      /**
       * Create a new object of this class.
       */
      public LockFreeQueueRecycle() {
        Node sentinel = new Node();
        head = new AtomicStampedReference<Node>(sentinel, 0);
        tail = new AtomicStampedReference<Node>(sentinel, 0);
      }
      private Node allocate(T value) {
        int[] stamp = new int[1];
        Node node = freeList.get();
        if (node == null) { // nothing to recycle
          node = new Node();
        } else {            // recycle existing node
          freeList.set(node.next.get(stamp));
        }
        // initialize
        node.value = value;
        return node;
      }
      private void free(Node node) {
        Node free = freeList.get();
        node.next = new AtomicStampedReference<Node>(free, 0);
        freeList.set(node);
      }
      /**
       * Enqueue an item.
       * @param value Item to enqueue.
       */
      public void enq(T value) {
        // try to allocate new node from local pool
        Node node = allocate(value);
        int[] lastStamp = new int[1];
        int[] nextStamp = new int[1];
        int[] stamp = new int[1];
        while (true) {		 // keep trying
          Node last = tail.get(lastStamp);    // read tail
          Node next = last.next.get(nextStamp); // read next
          // are they consistent?
          if (last == tail.get(stamp) && stamp[0] == lastStamp[0]) {
            if (next == null) {	// was tail the last node?
              // try to link node to end of list
              if (last.next.compareAndSet(next, node,
                  nextStamp[0], nextStamp[0]+1)) {
                // enq done, try to advance tail
                tail.compareAndSet(last, node,
                    lastStamp[0], lastStamp[0]+1);
                return;
              }
            } else {		// tail was not the last node
              // try to swing tail to next node
              tail.compareAndSet(last, next,
                  lastStamp[0], lastStamp[0]+1);
            }
          }
        }
      }
      /**
       * Dequeue an item.
       * @throws queue.EmptyException The queue is empty.
       * @return Item at the head of the queue.
       */
      public T deq() throws EmptyException {
        int[] lastStamp = new int[1];
        int[] firstStamp = new int[1];
        int[] nextStamp = new int[1];
        int[] stamp = new int[1];
        while (true) {
          Node first = head.get(firstStamp);
          Node last = tail.get(lastStamp);
          Node next = first.next.get(nextStamp);
          // are they consistent?
          if (first == head.get(stamp) && stamp[0] == firstStamp[0]) {
            if (first == last) {	// is queue empty or tail falling behind?
              if (next == null) {	// is queue empty?
                throw new EmptyException();
              }
              // tail is behind, try to advance
              tail.compareAndSet(last, next,
                  lastStamp[0], lastStamp[0]+1);
            } else {
              T value = next.value; // read value before dequeuing
              if (head.compareAndSet(first, next, firstStamp[0], firstStamp[0]+1)) {
                free(first);
                return value;
              }
            }
          }
        }
      }
      /**
       * Items are kept in a list of nodes.
       */
      public class Node {
        /**
         * Item kept by this node.
         */
        public T value;
        /**
         * Next node in the queue.
         */
        public AtomicStampedReference<Node> next;
        
        /**
         * Create a new node.
         */
        public Node() {
          this.next  = new AtomicStampedReference<Node>(null, 0);
        }
      }
    }

/*
     * EmptyException.java
     *
     * Created on December 28, 2005, 12:02 AM
     *
     * Copyright 2005 Sun Microsystems, Inc.  All Rights Reserved.
     *
     * Sun Microsystems, Inc. has intellectual property rights relating to technology embodied in the product that is described in this document.  In particular, and without limitation, these intellectual property rights may include one or more of the U.S. patents listed at http://www.sun.com/patents and one or more additional patents or pending patent applications in the U.S. and in other countries.
     * U.S. Government Rights - Commercial software.  Government users are subject to the Sun Microsystems, Inc. standard license agreement and applicable provisions of the FAR and its supplements.  Use is subject to license terms.
     * Sun, Sun Microsystems, the Sun logo and Java are trademarks or registered trademarks of Sun Microsystems, Inc. in the U.S. and other countries.
     */
    
    package june.shen;
    
    /**
     * @author Maurice Herlihy
     */
    public class EmptyException extends java.lang.Exception {
      
      /** Creates a new instance of EmptyException */
      public EmptyException() {
        super();
      }
      
    }

package june.shen;
    
    /**
     @author juneshen
     lockfree queue testsuit
     **/
    
    public class LockFreeQueueTest {
    
    	public  static LockFreeQueueRecycle<Integer> instance=new LockFreeQueueRecycle<Integer>();
    	private static int THREADS=10;
    	private static  Thread[] pushthread = new Thread[THREADS];
    	private static  Thread[] popthread = new Thread[THREADS];
    	
    	public static void main(String[] args) {
    		
    		
    		
    		 long startTime;
    
    			startTime = System.currentTimeMillis();
    			
    		    for (int i = 0; i < THREADS; i++) {
    		    	pushthread[i] = new PushThread();
    		    }
    		    for (int i = 0; i < THREADS; i ++) {
    		    	pushthread[i].start();
    		    }
    		    for (int i = 0; i < THREADS; i ++) {
    		    	try {
    					pushthread[i].join();
    				} catch (InterruptedException e) {
    					// TODO Auto-generated catch block
    					e.printStackTrace();
    				}
    		    }
    		    
    		    Thread pushend= new Thread(new PushendThread());
    		    pushend.start();
    		    
    		    for (int i = 0; i < THREADS; i++) {
    		    	popthread[i] = new PopThread();
    		    }
    		    for (int i = 0; i < THREADS; i ++) {
    		    	popthread[i].start();
    		    }
    		    for (int i = 0; i < THREADS; i ++) {
    		    	try {
    					popthread[i].join();
    				} catch (InterruptedException e) {
    					// TODO Auto-generated catch block
    					e.printStackTrace();
    				}
    		    }
    		    
    		    long endTime = System.currentTimeMillis();
    			double elapsedTime = ((double) (endTime - startTime)) / 1000.0;
    			
    			double throughput = ((double) THREADS*2 / elapsedTime);
    		 
    		    System.out.printf(" 吞吐量: %f, 运行时间: %f \n",throughput,elapsedTime);
    	}
    
    	
    
    }
    
    class PopThread extends Thread {
    	
    	    public void run() { 
    	    	try {
    	    		LockFreeQueueTest.instance.deq();
    			} catch (EmptyException e) {
    				// TODO Auto-generated catch block
    				e.printStackTrace();
    			}
    	    }
    	  }
    
    class PushThread extends Thread {
    	
    	    public void run() { 
    	    	LockFreeQueueTest.instance.enq(1);
    	    }
    }
    
    class PushendThread extends Thread {
    	
    	    public void run() { 
    	      for(int i=0;i<1000;i++)
    	    	LockFreeQueueTest.instance.enq(i);
    	    }
    }