一、读写锁

读写锁的特点：多个线程可以并发读

线程间读读不互斥
	
读写互斥
	
写写互斥
	
读：加读锁 readlock
	
写：加写锁 writelock
	
只有读：不加锁
	
只有写：加sychronized、ReentrantLock
	
既有读又有写：读写锁
在vector中加读、写锁：

import java.util.AbstractList;
import java.util.Arrays;
import java.util.List;
import java.util.RandomAccess;
import java.util.concurrent.locks.ReentrantReadWriteLock;

public class MyVector<E>
        extends AbstractList<E>
        implements List<E>, RandomAccess, Cloneable, java.io.Serializable {
    private ReentrantReadWriteLock lock = new ReentrantReadWriteLock();
    protected Object[] elementData;

    protected int elementCount;

    protected int capacityIncrement;

    @Override
    public E get(int index) {
        lock.readLock().lock();   //加上了读锁
        try {
            if (index >= elementCount)
                throw new ArrayIndexOutOfBoundsException(index);
            //get 操作
            return elementData(index);
        } finally {
            lock.readLock().unlock();
        }
    }

    E elementData(int index) {
        return (E) elementData[index];
    }

    public void add(int index, E element) {
        insertElementAt(element, index);
    }

    public void insertElementAt(E obj, int index) {
        lock.writeLock().lock();   //加上了写锁
        try {
            if (index > elementCount) {
                throw new ArrayIndexOutOfBoundsException(index
                        + " > " + elementCount);
            }
            ensureCapacityHelper(elementCount + 1);
            System.arraycopy(elementData, index, elementData, index + 1, elementCount - index);
            elementData[index] = obj;
            elementCount++;
        } finally {
            lock.writeLock().unlock();
        }
    }

    private void ensureCapacityHelper(int minCapacity) {
        // overflow-conscious code
        if (minCapacity - elementData.length > 0)
            grow(minCapacity);
    }

    private static final int MAX_ARRAY_SIZE = Integer.MAX_VALUE - 8;

    private static int hugeCapacity(int minCapacity) {
        if (minCapacity < 0) // overflow
            throw new OutOfMemoryError();
        return (minCapacity > MAX_ARRAY_SIZE) ?
                Integer.MAX_VALUE :
                MAX_ARRAY_SIZE;
    }

    private void grow(int minCapacity) {
        // overflow-conscious code
        int oldCapacity = elementData.length;//原数组长度
        int newCapacity = oldCapacity + ((capacityIncrement > 0) ?
                capacityIncrement : oldCapacity);
        if (newCapacity - minCapacity < 0)
            newCapacity = minCapacity;
        if (newCapacity - MAX_ARRAY_SIZE > 0)
            newCapacity = hugeCapacity(minCapacity);
        elementData = Arrays.copyOf(elementData, newCapacity);
    }

    @Override
    public int size() {
        return 0;
    }

    public static void main(String[] args) {
        //   new MyVector   只做读操作    get()  睡眠1S
        // new  Vector    只做读操作

        //多个线程     线程数量到达多少时    MyVector 性能高于  Vector
        //线程的切换
    }
}


二、悲观锁与乐观锁

1、悲观锁

总是假设最坏的情况，每次去拿数据的时候都认为别人会修改，所以每次在拿数据的时候都会上锁，这样别人想拿这个数据就会阻塞直到它拿到锁。传统的关系型数据库里就用到了很多这种锁机制，比如行锁，表锁等，读锁，写锁等，都是在做操作之前先上锁。再比如Java里面的同步原语synchronized关键字的实现也是悲观锁。

虽然有多个线程，但是在某一时间段内，并不会互相造成影响。

悲观锁机制存在以下问题：

在多线程竞争下，加锁、释放锁会导致比较多的上下文切换和调度延时，引起性能问题
	
一个线程持有锁会导致其他所有需要此锁的线程挂起
	
如果一个优先级高的线程等待一个优先级低的线程释放锁会导致优先级倒置，引起性能风险
2、乐观锁

乐观锁就是一种思想。相对悲观锁而言，乐观锁假设认为数据一般情况下不会产生并发冲突，所以在数据进行提更新的时候，才会正式对数据是否产生并发冲突进行检测，如果发行并发冲突了，则让返回用户错误的信息，让用户决定如何去做。乐观锁的概念中其实已经阐述了它的具体实现细节：主要两个步骤：冲突检测和数据更新。

其实现方式有一种比较典型的就是Compare and Swap（CAS）

CAS：



CAS是乐观锁技术，当多个线程尝试使用CAS同时更新同一个变量时，只有其一个线程能更新变量的 值，而其它线程都失败，失败的线程并不会被挂起，而是被告知这次竞争中失败，并可以再次尝试。CAS 操作中包含三个操作数 —— 需要读写的内存位置（V）、进行比较的预期原值（A）1  2   和拟写入的新值(B)。3如果内存位置V的值与预期原值A相匹配，那么处理器会自动将该位置值更新为新值B。否则处理器不做任何操作。无论哪种情况，它都会在 CAS 指令之前返回该位置的值。（在 CAS 的一些特殊情况下将仅返回 CAS 是否成功，而不提取当前值。）CAS 有效地说明了“ 我认为位置 V 应该包含值 A；如果包含该值，则将 B 放到这个位置；否则，不要更改该位置，只告诉我这个位置现在的值即可。 ”这其实和乐观锁的冲突检查+数据更新的原理是一样的。

这里再强调一下，乐观锁是一种思想。CAS是这种思想的一种实现方式。

JAVA对CAS的支持：

在JDK1.5 中新增 java.util.concurrent (J.U.C)就是建立在CAS之上的。相对于对于 synchronized 这种阻塞算法，CAS是非阻塞算法的一种常见实现。所以J.U.C在性能上有了很大的提升。以 java.util.concurrent 中的 AtomicInteger 为例，看一下在不使用锁的情况下是如何保证线程安全的。主要理解 getAndIncrement 方法，该方法的作用相当于 ++i 操作。



Atomic实现十个线程++一万次：

import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.ReentrantLock;

public class Main {
    private static CountDownLatch latch = new CountDownLatch(10);//count = 10
    //    private static int index = 0;
    private static AtomicInteger aIn = new AtomicInteger();
    private static ReentrantLock lock = new ReentrantLock();
    // private static final Object monitor = new Object();

    public static void increase() {
        aIn.getAndIncrement();

//        lock.lock();  //无限阻塞  容易产生死锁 （预习）
//        try {
//            index++;//非原子性操作
//        } finally {
//            lock.unlock();
//        }
    }

    public static void main(String[] args) {

        Object o = new Object();
        for (int i = 0; i < 10; i++) {
            new Thread() {
                @Override
                public void run() {
                    for (int i = 0; i < 10000; i++) {
                        increase();
                    }
                    latch.countDown();  //-1
                }
            }.start();
        }
        try {
            latch.await();  //等count减为0
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        System.out.println(aIn.get());//  主线程打印
    }
}

class Program
    {
        private static ReaderWriterLockSlim _LockSlim = new ReaderWriterLockSlim();
        private static Hashtable list = Hashtable.Synchronized(new Hashtable());

        static void Main(string[] args)
        {
            new Thread(new ParameterizedThreadStart(delegate(object obj) {
                Write(0, 10000);
            })).Start();

            Thread.Sleep(1000);

            new Thread(new ParameterizedThreadStart(delegate(object obj)
            {
                Read(1, 1000);
            })).Start();

            new Thread(new ParameterizedThreadStart(delegate(object obj)
            {
                Read(2, 1000);
            })).Start();

            new Thread(new ParameterizedThreadStart(delegate(object obj)
            {
                Read(3, 1000);
            })).Start();

            new Thread(new ParameterizedThreadStart(delegate(object obj)
            {
                Read(4, 1000);
            })).Start();

            Console.ReadKey();
        }

        /// <summary>
        /// 对于i相同的线程进行阻塞保证不会并行多个
        /// </summary>
        /// <param name="i"></param>
        /// <param name="j"></param>
        public static void Read(int i,int j)
        {
            Console.WriteLine(DateTime.Now.ToString() + "\t线程" + i.ToString() + "即将进入读取状态");
            _LockSlim.EnterReadLock();
            Console.WriteLine(DateTime.Now.ToString() + "\t线程" + i.ToString() + "进入读取状态");
            
            Thread.Sleep(j);

            _LockSlim.ExitReadLock();
            Console.WriteLine(DateTime.Now.ToString() + "\t线程" + i.ToString() + "退出读取状态");
        }

        public static void Write(int i, int j)
        {
            Console.WriteLine(DateTime.Now.ToString() + "\t线程" + i.ToString() + "即将进入写入状态");
            _LockSlim.EnterWriteLock();
            Console.WriteLine(DateTime.Now.ToString() + "\t线程" + i.ToString() + "进入写入状态");

            Thread.Sleep(j);

            _LockSlim.ExitWriteLock();
            Console.WriteLine(DateTime.Now.ToString() + "\t线程" + i.ToString() + "退出写入状态");
        }


    }

输出：

2017/2/20 17:51:55      线程0即将进入写入状态
2017/2/20 17:51:55      线程0进入写入状态
2017/2/20 17:51:56      线程1即将进入读取状态
2017/2/20 17:51:56      线程2即将进入读取状态
2017/2/20 17:51:56      线程3即将进入读取状态
2017/2/20 17:51:56      线程4即将进入读取状态
2017/2/20 17:52:05      线程1进入读取状态
2017/2/20 17:52:05      线程4进入读取状态
2017/2/20 17:52:05      线程0退出写入状态
2017/2/20 17:52:05      线程2进入读取状态
2017/2/20 17:52:05      线程3进入读取状态
2017/2/20 17:52:06      线程2退出读取状态
2017/2/20 17:52:06      线程1退出读取状态
2017/2/20 17:52:06      线程4退出读取状态
2017/2/20 17:52:06      线程3退出读取状态

 
转载于:https://www.cnblogs.com/nanfei/p/6420681.html

public class ReadWriteLockTest {
    ReentrantReadWriteLock readWriteLock = new ReentrantReadWriteLock();
    Lock readLock = readWriteLock.readLock();
    Lock writeLock = readWriteLock.writeLock();
    List<User> userList = new ArrayList<>();

    public static void main(String[] args) {
        ReadWriteLockTest readWriteLockTest = new ReadWriteLockTest();
        for(int i=0; i<100; i++){
            new Thread(new Runnable() {
                @Override
                public void run() {
                    readWriteLockTest.read();
                }
            }).start();
            new Thread(new Runnable() {
                @Override
                public void run() {
                    readWriteLockTest.write();
                }
            }).start();
        }
    }

    public void read(){
        readLock.lock();
        try {
            System.out.println(JSON.toJSONString(userList));
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            readLock.unlock();
        }
    }

    public void write(){
        writeLock.lock();
        Thread thread = Thread.currentThread();
        try {
            User user = new User()
                    .setId(thread.getPriority())
                    .setName(thread.getName())
                    .setBirthday(new Date());
            userList.add(user);
           // System.out.println(JSON.toJSONString(userList));
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            writeLock.unlock();
        }
    }
}

 

    
需求：现需要将多个数据源的数据导入到目标数据库，这是一个经典的生产消费应用的例子。
直接上代码，看下实现：

            // 初始化列队缓冲区 队列大小为100
            IDataCollection<List<T>> queue = new QueueCollection<List<T>>(100);

            //开启X个后台任务，读取RabbitMQ队列信息, 把列队信息插入缓冲区队列
            var count = 1;
            for (int i = 0; i < count; i++)
            {
                Task.Factory.StartNew(() => new Producer<List<T>>(queue).Start(new RabbitSource<List<T>>().Get));
            }

            //开启X个后台任务，主动获取数据库数据，作为数据生产者，插入到缓冲区队列，
            for (int i = 0; i < count; i++)
            {
                Task.Factory.StartNew(() => new Producer<List<T>>(queue).Start(new DatabaseSource<List<T>>().Get));
            }

            //开启X个后台任务，主动获取读取缓冲区列队，作为数据消息者，把数据插入到ES库，
            for (int i = 0; i < count; i++)
            {
                Task.Factory.StartNew(() => new Customer<List<T>>(queue).Start(new Elastic().Insert));
            }

队列我们采用线程安全的ConcurrentQueue队列：

/// <summary>
    /// 缓冲区队列
    /// ConcurrentQueue线程安全，不用考虑锁的问题
    /// </summary>
    public class QueueCollection<T> :IDataCollection<T>
    {
        //队列最大值
        private int _maxSize;

        /// <summary>
        /// 线程安全的队列
        /// </summary>
        private ConcurrentQueue<T> _queue;

        public QueueCollection(int maxSize)
        {
            this._maxSize = maxSize;
            _queue = new ConcurrentQueue<T>();
        }
        
        public bool isPopWaiting()
        {
            return !_queue.Any();
        }

        public bool isPushWaiting()
        {
            return this._maxSize == _queue.Count;
        }
        
        public T Pop()
        {
            T _obj = default(T);
            if (!_queue.IsEmpty)
                _queue.TryDequeue(out _obj);

            return _obj;
        }

        public void Push(T t)
        {
            if (this._maxSize > _queue.Count)
            {
                _queue.Enqueue(t);
            }
        }
    }

如果我们不使用这个队列，只要满足IDataCollection接口，也可以进行替换：

public interface IDataCollection<T>
    {
        /// <summary>
        /// 插入数据 
        /// </summary>
        /// <param name="t"></param>
        void Push(T t);

        /// <summary>
        /// 取出数据
        /// </summary>
        /// <returns></returns>
        T Pop();

        /// <summary>
        /// 是否插入数据等待
        /// </summary>
        /// <returns></returns>
        bool isPushWaiting();

        /// <summary>
        /// 是否取出数据等待
        /// </summary>
        /// <returns></returns>
        bool isPopWaiting();
        
    }

生产者：

 public class Producer<T> : ITransientDependency
    {
        private int sleep;

        private IDataCollection<T> bufferQueue;

        public Producer(IDataCollection<T> queue)
        {
            sleep = 3000;
            bufferQueue = queue;
        }

        public void Start(Action<Action<T>> methodCall)
        {
            //入队
            methodCall((bills) => 
            {
                this.Enqueue(bills);
            });
        }

        private void Enqueue(T t)
        {
            var isWaiting = true;

            while (isWaiting)
            {
                if (!bufferQueue.isPushWaiting())
                {
                    this.bufferQueue.Push(t);
                    isWaiting = false;
                }
                else
                {
                    //生产者等待时间
                    Thread.Sleep(sleep);
                }
            }
        }
    }

消费者：

/// <summary>
    /// 消费者
    /// </summary>
    public class Customer<T>
    {
        //产品缓存队列
        private IDataCollection<T> _queue;
        
        //消费者等待时间
        private int Spead = 5000;//消费者等待时间

        public Customer(IDataCollection<T> queue)
        {
            this._queue = queue;
        }

        public void Start(Action<T> method)
        {
            while (true)
            {
                if (!_queue.isPopWaiting())
                {
                    T box = this._queue.Pop();

                    method(box);
                }
                else
                {
                    Thread.Sleep(Spead);
                }
            }
        }
    }

方法委托，也写了个基类，其实意义并不大，只是为了规范, 防止方法命名随意起。

    public interface IDataSource<T>
    {
        void Get(Action<T> func);
    }

最后，在DataSource的get方法中，调用 func即可。
 

转载于:https://www.cnblogs.com/hankuikui/p/8042817.html