Java多线程

public enum State {
        /**
         * Thread state for a thread which has not yet started.
         */
        NEW,

        /**
         * Thread state for a runnable thread.  A thread in the runnable
         * state is executing in the Java virtual machine but it may
         * be waiting for other resources from the operating system
         * such as processor.
         */
        RUNNABLE,

        /**
         * Thread state for a thread blocked waiting for a monitor lock.
         * A thread in the blocked state is waiting for a monitor lock
         * to enter a synchronized block/method or
         * reenter a synchronized block/method after calling
         * {@link Object#wait() Object.wait}.
         */
        BLOCKED,

        /**
         * Thread state for a waiting thread.
         * A thread is in the waiting state due to calling one of the
         * following methods:
         * <ul>
         *   <li>{@link Object#wait() Object.wait} with no timeout</li>
         *   <li>{@link #join() Thread.join} with no timeout</li>
         *   <li>{@link LockSupport#park() LockSupport.park}</li>
         * </ul>
         *
         * <p>A thread in the waiting state is waiting for another thread to
         * perform a particular action.
         *
         * For example, a thread that has called <tt>Object.wait()</tt>
         * on an object is waiting for another thread to call
         * <tt>Object.notify()</tt> or <tt>Object.notifyAll()</tt> on
         * that object. A thread that has called <tt>Thread.join()</tt>
         * is waiting for a specified thread to terminate.
         */
        WAITING,

        /**
         * Thread state for a waiting thread with a specified waiting time.
         * A thread is in the timed waiting state due to calling one of
         * the following methods with a specified positive waiting time:
         * <ul>
         *   <li>{@link #sleep Thread.sleep}</li>
         *   <li>{@link Object#wait(long) Object.wait} with timeout</li>
         *   <li>{@link #join(long) Thread.join} with timeout</li>
         *   <li>{@link LockSupport#parkNanos LockSupport.parkNanos}</li>
         *   <li>{@link LockSupport#parkUntil LockSupport.parkUntil}</li>
         * </ul>
         */
        TIMED_WAITING,

        /**
         * Thread state for a terminated thread.
         * The thread has completed execution.
         */
        TERMINATED;
}

public class TestOne {
	public static void main(String[] args) {
		ThreadMXBean mxBean = ManagementFactory.getThreadMXBean();
		ThreadInfo[] allThreads = mxBean.dumpAllThreads(false, false);
		for (ThreadInfo threadInfo : allThreads) {
				System.out.println(threadInfo.getThreadId()+"==="+
                                   threadInfo.getThreadName());
		}
	}
}

5===Attach Listener 
4===Signal Dispatcher 分发处理发送给jvm信号的线程
3===Finalizer 调用对象finalize方法的线程，就是垃圾回收的线程
2===Reference Handler 清除reference的线程
1===main

public class SyncTest{

    public static void main(String[] args) {
        Ticket ticket = new Ticket();
        new Thread(ticket,"线程1").start();
        new Thread(ticket,"线程2").start();
        new Thread(ticket,"线程3").start();
    }
}

class Ticket implements Runnable{

    int ticket = 100;

    @Override
    public void run() {
        while (true){
            try {
                //模拟网络延迟
                Thread.sleep(100);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            if (ticket > 0) {
                // this表示Ticket对象
                sell();
            }else {
                break;
            }
        }
    }

    public void sell(){
        System.out.println(Thread.currentThread().getName() + "售票，票号为：" + ticket--);
    }

}

	public static int i = 1;
    public static final int TOTAL = 100;
    public static Object lock = new Object();

    public static void main(String[] args) {
        Thread thread1 = new Thread(() -> {
            while (i <= TOTAL) {
                synchronized (lock) {
                    if (i % 2 == 1) {
                        System.out.println("i=" + i++);
                        lock.notify();
                    } else {
                        try {
                            lock.wait();
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                        }
                    }
                }
            }
        });

        Thread thread2 = new Thread(() -> {
            while (i <= TOTAL) {
                synchronized (lock) {
                    if (i % 2 == 0) {
                        System.out.println("i=" + i++);
                        lock.notify();
                    } else {
                        try {
                            lock.wait();
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                        }
                    }
                }
            }
        });

        thread1.start();
        thread2.start();
    }

	private static AtomicInteger num = new AtomicInteger(1);
    private static CountDownLatch countDownLatch = new CountDownLatch(2);

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

        Thread thread1 = new Thread(() -> {
            while (num.intValue() < 100) {
                if (num.intValue() % 2 == 0) {
                    System.out.println(Thread.currentThread().getName() + ":" + num );
                    num.incrementAndGet();
                }
                countDownLatch.countDown();
            }

        },"偶数");

        Thread thread2 = new Thread(()->{
            while (num.intValue() < 100) {
                if (num.intValue() % 2 == 1) {
                    System.out.println(Thread.currentThread().getName() + ":" + num );
                    num.incrementAndGet();
                }
                countDownLatch.countDown();
            }
        },"奇数");

        thread1.start();
        thread2.start();
        countDownLatch.await();
    }

public ThreadPoolExecutor(int corePoolSize,      //核心线程数
                          int maximumPoolSize,   //最大线程数
                          long keepAliveTime,    //活跃时间
                          TimeUnit unit,         //时间单位
                          BlockingQueue<Runnable> workQueue, //存放任务的队列
                          ThreadFactory threadFactory, //线程工厂
                          RejectedExecutionHandler handler//超出线程范围和队列容量的任务处理程序
                         )

ThreadFactory namedThreadFactory = new ThreadFactoryBuilder()
        .setNameFormat("demo-pool-%d").build();
ExecutorService threadPool = new ThreadPoolExecutor(2, 5,
        10L, TimeUnit.MILLISECONDS,
        new LinkedBlockingQueue<>(10), namedThreadFactory, new ThreadPoolExecutor.AbortPolicy());

ExecutorService executorService = Executors.newSingleThreadExecutor();//单个线程
                                            newCachedThreadPool();  //可缓存线程池
                                            newScheduledThreadPool(2);//定时任务
                                            newFixedThreadPool(2);   //固定核心数

new RejectedExecutionHandler(){
    @Override
    public void rejectedExecution(Runnable r,ThreadPoolExecutor ExecutorService){
        System.out.println(r.toString()+"is discard");
    }
}

public class ProducerConsumer {

    private static BlockingQueue<String> queue = new ArrayBlockingQueue<>(5);

    private static class Producer extends Thread {
        @Override
        public void run() {
            try {
                queue.put("product");
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            System.out.print("produce..");
        }
    }

    private static class Consumer extends Thread {
        @Override
        public void run() {
            try {
                String product = queue.take();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            System.out.print("consume..");
        }
    }

    public static void main(String[] args) throws InterruptedException {
        for (int i = 0; i < 2; i++) {
            Producer producer = new Producer();
            producer.start();
        }
        for (int i = 0; i < 5; i++) {
            // 当2个消费完以后线程就会阻塞，直到有新的生产者
            Consumer consumer = new Consumer();
            consumer.start();
        }
        Thread.sleep(100);
        for (int i = 0; i < 3; i++) {
            Producer producer = new Producer();
            producer.start();
        }
    }
}

produce..produce..consume..consume..produce..consume..produce..consume..produce..consume..

public class SumDemo extends Thread{

    private static volatile int sum;
    private int start;
    private int end;

    public SumDemo(int start,int end) {
        this.start = start;
        this.end = end;
    }

    @Override
    public void run() {
        int tmp = 0 ;
        for (int i = start; i < end ; i++) {
            tmp+=i;
        }
        put(tmp);
    }

    /**
     * 如果不在同步方法中进行，多线程时更改sum的值会有线程安全问题
     * @param num
     */
    private static synchronized void put(int num){
        System.out.println(Thread.currentThread().getName() + "正在将结果附上" + num);
        sum += num;
    }

    public static void main(String[] args) throws InterruptedException {
        for (int i = 0; i < 10 ; i++) {
            SumDemo sumDemo = new SumDemo(i * 10 + 1, i * 10 + 10);
            sumDemo.start();
            sumDemo.join();
        }
        System.out.println("运行结果：" + SumDemo.sum);
    }
}

public class ForkJoinTest extends RecursiveTask<Integer> {

    //开始数
    private Integer start;
	//结束数
    private Integer end;
	//阈值
    private static int thresold = 10;

    public ForkJoinTest(Integer start, Integer end) {
        this.start = start;
        this.end = end;
    }

    @Override
    protected Integer compute() {
        int sum = 0;
        boolean canCompute = (end - start) <= thresold;
        if (canCompute) {
            for (int i = start; i <= end; i++) {
                sum += i;
            }
        } else {
            int mid = (start + end) >> 1;
            ForkJoinTest leftTask = new ForkJoinTest(start, mid);
            ForkJoinTest rightTask = new ForkJoinTest(mid + 1, end);
            invokeAll(leftTask, rightTask);
            sum = leftTask.join() + rightTask.join();
        }
        return sum;
    }

    public static void main(String[] args) throws ExecutionException, InterruptedException {
        ForkJoinPool pool = new ForkJoinPool();
        ForkJoinTest task = new ForkJoinTest(1,100);
        ForkJoinTask<Integer> submit = pool.submit(task);
        System.out.println(submit.get());
        System.out.println(pool.getPoolSize());
    }
}

import java.util.ArrayList;
import java.util.List;
 

public class ObjectUnit<T> {
    private Class<T> type;
    private List<T> items = new ArrayList<T>();
    private List<Boolean> checkedOut = new ArrayList<Boolean>();
    private int semaphore;
 

    public ObjectUnit(Class<T> type) {
       this.type = type;
    }
 

    public synchronized T addItem() {
       T obj;
       try {
           obj = type.newInstance();
       } catch (Exception e) {
           throw new RuntimeException(e);
       }
       items.add(obj);
       checkedOut.add(false);
       return obj;
    }
 

    public synchronized T checkOut() {
       if (semaphore < items.size()) {
           semaphore++;
           return getItem();
       } else
           return addItem();
    }
 

    public synchronized void checkIn(T x) {
       if (releaseItem(x))
           semaphore--;
    }
 

    private synchronized T getItem() {
       for (int index = 0; index < checkedOut.size(); index++)
           if (!checkedOut.get(index)) {
              checkedOut.set(index, true);
              return items.get(index);
           }
       return null;
    }
 

    private synchronized boolean releaseItem(T item) {
       int index = items.indexOf(item);
       if (index == -1)
           return false; // Not in the list
       if (checkedOut.get(index)) {
           checkedOut.set(index, false);
           return true;
       }
       return false;
    }
}

import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
 

public class Provider {
    private Map<Class<?>, ObjectUnit<?>> providers = new ConcurrentHashMap<Class<?>, ObjectUnit<?>>();
    private static Provider instance = new Provider();
 

    private Provider() {
    }
 

    public static Provider getInstance() {
       return instance;
    }
 

    @SuppressWarnings("unchecked")
    public <T> T getObj(Class<T> key) {
       ObjectUnit value = providers.get(key);
        if (value != null) {
           return (T) value.checkOut();
       } else {
           value = new ObjectUnit<T>(key);
           providers.put(key, value);
           return (T) value.addItem();
       }
    }
 

    @SuppressWarnings("unchecked")
    public <T> void renObj(T x) {
       if (providers.containsKey(x.getClass())) {
           ObjectUnit value = providers.get(x.getClass());
           value.checkIn(x);
       }
    }
}

public class SpinLock {

  private AtomicReference<Thread> sign =new AtomicReference<>();
  public void lock(){
    Thread current = Thread.currentThread();
    while(!sign .compareAndSet(null, current)){
    }
  }

  public void unlock (){
    Thread current = Thread.currentThread();
    sign .compareAndSet(current, null);
  }
}
/**
使用了CAS原子操作，lock函数将owner设置为当前线程，并且预测原来的值为空。unlock函数将owner设置为null，并且预测值为当前线程。

当有第二个线程调用lock操作时由于owner值不为空，导致循环一直被执行，直至第一个线程调用unlock函数将owner设置为null，第二个线程才能进入临界区。

由于自旋锁只是将当前线程不停地执行循环体，不进行线程状态的改变，所以响应速度更快。但当线程数不停增加时，性能下降明显，因为每个线程都需要执行，占用CPU时间。如果线程竞争不激烈，并且保持锁的时间段。适合使用自旋锁。

注：该例子为非公平锁，获得锁的先后顺序，不会按照进入lock的先后顺序进行。
*/

public ReentrantLock(boolean fair) {
     sync = fair ? new FairSync() : new NonfairSync();
}

static final class FairSync extends Sync {
    private static final long serialVersionUID = -3000897897090466540L;

    final void lock() {
        acquire(1);
    }

    /**
         * Fair version of tryAcquire.  Don't grant access unless
         * recursive call or no waiters or is first.
         */
    protected final boolean tryAcquire(int acquires) {
        final Thread current = Thread.currentThread();
        int c = getState();
        //拿到当前的同步状态, 如果是无锁状态， 则进行hasQueuedPredecessors方法逻辑
        if (c == 0) {
            if (!hasQueuedPredecessors() &&
                compareAndSetState(0, acquires)) {
                setExclusiveOwnerThread(current);
                return true;
            }
        }
        else if (current == getExclusiveOwnerThread()) {
            int nextc = c + acquires;
            if (nextc < 0)
                throw new Error("Maximum lock count exceeded");
            setState(nextc);
            return true;
        }
        return false;
    }
}

//获取锁。拿不到lock就不罢休，不然线程就一直block
void lock();

//如果当前线程未被中断，则获取锁。 
void lockInterruptibly() throws InterruptedException;

//仅在调用时锁未被另一个线程保持的情况下，才获取该锁，拿不到返回false
boolean tryLock();

//仅在调用时锁未被另一个线程保持的情况下，才获取该锁，拿不到lock，就等一段时间，超时返回false
boolean tryLock(long time, TimeUnit unit) throws InterruptedException;

//释放锁
void unlock();

//获取等待通知组件，该组件和当前的锁绑定，当前线程只有获得了锁，才能调用该组件的wait()方法，而调用后，当前线程将释放锁
Condition newCondition();

// 可重入锁
ReentrantLock.class;
// 读写锁    
ReentrantReadWriteLock.class    

Lock lock = new ReentrantLock();
Condition c = lock.newCondition()

abstract static class Sync extends AbstractQueuedSynchronizer
//非公平
static final class NonfairSync extends Sync
//公平    
static final class FairSync extends Sync
//读
public static class ReadLock implements Lock, java.io.Serializable 
//写    
public static class WriteLock implements Lock, java.io.Serializable    

	/**
     * Returns the lock used for reading.
     *
     * @return the lock used for reading
     */
    Lock readLock();

    /**
     * Returns the lock used for writing.
     *
     * @return the lock used for writing
     */
    Lock writeLock();

abstract static class Sync extends AbstractQueuedSynchronizer
//非公平
static final class NonfairSync extends Sync
//公平    
static final class FairSync extends Sync

public class Singleton {
    private volatile static Singleton uniqueInstance;
    private Singleton(){}
    //双检锁
    public static Singleton getInstance(){
        if(uniqueInstance == null){
            synchronized(Singleton.class){
                if(uniqueInstance == null){
                    uniqueInstance = new Singleton();
                }
            }
        }
        return uniqueInstance;
    }
}

// volatile保证可见性
private volatile int value;

/**
     * Creates a new AtomicInteger with the given initial value.
     *
     * @param initialValue the initial value
     */
public AtomicInteger(int initialValue) {
    value = initialValue;
}

/**
     * Atomically increments by one the current value.
     *
     * @return the previous value
     */
public final int getAndIncrement() {
    //这里利用unsafe方法
    return unsafe.getAndAddInt(this, valueOffset, 1); // i++操作
}

public final int getAndAddInt(Object var1, long var2, int var4) {
    int var5;
    do {
        var5 = this.getIntVolatile(var1, var2);
    } while(!this.compareAndSwapInt(var1, var2, var5, var5 + var4));

    return var5;
}

//native方法
public final native boolean compareAndSwapInt(Object var1, long var2, int var4, int var5);

	/**
     * Returns the value in the current thread's copy of this
     * thread-local variable.  If the variable has no value for the
     * current thread, it is first initialized to the value returned
     * by an invocation of the {@link #initialValue} method.
     *
     * @return the current thread's value of this thread-local
     */
    public T get() {
        Thread t = Thread.currentThread();
        ThreadLocalMap map = getMap(t);
        if (map != null) {
            ThreadLocalMap.Entry e = map.getEntry(this);
            if (e != null) {
                @SuppressWarnings("unchecked")
                T result = (T)e.value;
                return result;
            }
        }
        return setInitialValue();
    }

	/**
     * Sets the current thread's copy of this thread-local variable
     * to the specified value.  Most subclasses will have no need to
     * override this method, relying solely on the {@link #initialValue}
     * method to set the values of thread-locals.
     *
     * @param value the value to be stored in the current thread's copy of
     *        this thread-local.
     */
    public void set(T value) {
        Thread t = Thread.currentThread();
        ThreadLocalMap map = getMap(t);
        if (map != null)
            map.set(this, value);
        else
            createMap(t, value);
    }

	/**
     * Removes the current thread's value for this thread-local
     * variable.  If this thread-local variable is subsequently
     * {@linkplain #get read} by the current thread, its value will be
     * reinitialized by invoking its {@link #initialValue} method,
     * unless its value is {@linkplain #set set} by the current thread
     * in the interim.  This may result in multiple invocations of the
     * {@code initialValue} method in the current thread.
     *
     * @since 1.5
     */
     public void remove() {
         ThreadLocalMap m = getMap(Thread.currentThread());
         if (m != null)
             m.remove(this);
     }

//调用await()方法的线程会被挂起，它会等待直到count值为0才继续执行
public void await() throws InterruptedException { }; 

//和await()类似，只不过等待一定的时间后count值还没变为0的话就会继续执行
public boolean await(long timeout, TimeUnit unit) throws InterruptedException { };  

//将count值减1
public void countDown() { };  

public class CountDownLatchTest {

    public static void main(String[] args) throws InterruptedException {
        final int totalThread = 10;
        CountDownLatch countDownLatch = new CountDownLatch(totalThread);
        ExecutorService executorService = Executors.newCachedThreadPool();
        for (int i = 0; i < totalThread; i++) {
            executorService.execute(() -> {
                System.out.print("run..");
                countDownLatch.countDown();
            });
        }
        countDownLatch.await();
        System.out.println("end");
        executorService.shutdown();
    }
    
}

run..run..run..run..run..run..run..run..run..run..end

public CyclicBarrier(int parties, Runnable barrierAction) {
    if (parties <= 0) throw new IllegalArgumentException();
    this.parties = parties;
    this.count = parties;
    this.barrierCommand = barrierAction;
}
public CyclicBarrier(int parties) {
    this(parties, null);
}

public int await() throws InterruptedException, BrokenBarrierException {
    try {
        return dowait(false, 0L);
    } catch (TimeoutException toe) {
        throw new Error(toe); // cannot happen
    }
}

public void reset() {
    final ReentrantLock lock = this.lock;
    lock.lock();
    try {
        breakBarrier();   // break the current generation
        nextGeneration(); // start a new generation
    } finally {
        lock.unlock();
    }
}

public class CyclicBarrierTest {

    public static void main(String[] args) {
        final int totalThread = 10;
        CyclicBarrier cyclicBarrier = new CyclicBarrier(totalThread);
        ExecutorService executorService = Executors.newCachedThreadPool();
        for (int i = 0; i < totalThread; i++) {
            executorService.execute(() -> {
                System.out.print("before..");
                try {
                    cyclicBarrier.await();
                } catch (InterruptedException | BrokenBarrierException e) {
                    e.printStackTrace();
                }
                System.out.print("after..");
            });
        }
        executorService.shutdown();
    }
}

before..before..before..before..before..before..before..before..before..before..after..after..after..after..after..after..after..after..after..after..

//从此信号量获取一个许可，在提供一个许可前一直将线程阻塞，否则线程被中断。
void acquire();

//释放一个许可，将其返回给信号量。   
void release();

//返回此信号量中当前可用的许可数。
int availablePermits();

//查询是否有线程正在等待获取。
boolean hasQueuedThreads();

public class SemaphoreTest {

    public static void main(String[] args) {
        final int clientCount = 3;
        final int totalRequestCount = 10;
        Semaphore semaphore = new Semaphore(clientCount);
        ExecutorService executorService = Executors.newCachedThreadPool();
        for (int i = 0; i < totalRequestCount; i++) {
            executorService.execute(()->{
                try {
                    semaphore.acquire();
                    System.out.print(semaphore.availablePermits() + " ");
                } catch (InterruptedException e) {
                    e.printStackTrace();
                } finally {
                    semaphore.release();
                }
            });
        }
        executorService.shutdown();
    }
}

1 0 0 1 2 2 0 1 2 2