解决接口怎么避免多个人同时访问取到相同的数据

我会带着你远行 2023-07-13 13:46 6阅读 0赞

一.简单粗暴，直接使用 @Transactional 或者 synchronized

二.使用阻塞队列 如BlockingQueue

1 BlockingQueue概述：

BlocingQueue,顾名思义：阻塞队列.BlockingQueue是在java.util.concurrent下的，因此不难理解，BlockingQueue是为了解决多线程中数据高效安全传输而提出的。 
阻塞队列所谓的“阻塞”，指的是某些情况下线程会挂起（即阻塞），一旦条件满足，被挂起的线程又会自动给唤醒。使用BlockingQueue，不需要关心什么时候需要阻塞线程，什么时候需哟啊唤醒线程，这些内容BlockingQueue都已经做好了。

2 BlockingQueue中的方法：

方法摘要
 boolean add(E e) 
 将指定元素插入此队列中（如果立即可行且不会违反容量限制），成功时返回 true，如果当前没有可用的空间，则抛出 IllegalStateException。
 boolean contains(Object o) 
 如果此队列包含指定元素，则返回 true。
 int drainTo(Collection<? super E> c) 
 移除此队列中所有可用的元素，并将它们添加到给定 collection 中。
 int drainTo(Collection<? super E> c, int maxElements) 
 最多从此队列中移除给定数量的可用元素，并将这些元素添加到给定 collection 中。
 boolean offer(E e) 
 将指定元素插入此队列中（如果立即可行且不会违反容量限制），成功时返回 true，如果当前没有可用的空间，则返回 false。
 boolean offer(E e, long timeout, TimeUnit unit) 
 将指定元素插入此队列中，在到达指定的等待时间前等待可用的空间（如果有必要）。
 E poll(long timeout, TimeUnit unit) 
 获取并移除此队列的头部，在指定的等待时间前等待可用的元素（如果有必要）。
 void put(E e) 
 将指定元素插入此队列中，将等待可用的空间（如果有必要）。
 int remainingCapacity() 
 返回在无阻塞的理想情况下（不存在内存或资源约束）此队列能接受的附加元素数量；如果没有内部限制，则返回 Integer.MAX_VALUE。
 boolean remove(Object o) 
 从此队列中移除指定元素的单个实例（如果存在）。
 E take() 
 获取并移除此队列的头部，在元素变得可用之前一直等待（如果有必要）。

3 ArrayBlockingQueue

基于数组的阻塞队列，必须制定队列大小。ArrayBlocingQueue中只有一个ReetrantLock对象，这意味着生产者和消费者无法并行。另外创建ArrayBlockingQueue时，可以指定ReetrantLock是否为公平锁，默认采用非公平锁。

/** Main lock guarding all access */
        final ReentrantLock lock;
    
        /** Condition for waiting takes */
        private final Condition notEmpty;
    
        /** Condition for waiting puts */
        private final Condition notFull;

LinkedBlockingQueue

/** Lock held by take, poll, etc */
        private final ReentrantLock takeLock = new ReentrantLock();
    
        /** Wait queue for waiting takes */
        private final Condition notEmpty = takeLock.newCondition();
    
        /** Lock held by put, offer, etc */
        private final ReentrantLock putLock = new ReentrantLock();
    
        /** Wait queue for waiting puts */
        private final Condition notFull = putLock.newCondition();

基于链表的阻塞队列，和ArrayBlockingQueue差不多。不过LinkedBlockingQueue如果不指定队列容量大小，会默认一个类似无限大小的容量，之所以说是类似是因为这个无限大小是Integer.MAX\_VALUE，这么说就好理解ArrayBlockingQueue为什么必须要制定大小了，如果ArrayBlockingQueue不指定大小的话就用Integer.MAX\_VALUE，那将造成大量的空间浪费，但是基于链表实现就不一样的，一个一个节点连起来而已。另外，LinkedBlockingQueue生产者和消费者都有自己的锁（见下面的代码），这意味着生产者和消费者可以"同时"运行。

5 SynchronousQueue

一种没有缓冲的等待队列。  
什么叫做没有缓存区：  
ArrayBlockingQueue中有：

/** The queued items  */
    private final E[] items;

数组用以存储队列  
LinkedBlockingQueue：

/**
 * Linked list node class
 */
 static class Node<E> {
 /** The item, volatile to ensure barrier separating write and read */
 volatile E item;
 Node<E> next;
 Node(E x) { item = x; }
 }

将队列以链表形式连接。  
生产者/消费者操作数据实际上都是通过这两个"中介"来操作数据的，但是SynchronousQueue则是生产者直接把数据给消费者（消费者直接从生产者这里拿数据），好像又回到了没有生产者/消费者模型的老办法了。换句话说，每一个插入操作必须等待一个线程对应的移除操作。SynchronousQueue又有两种模式：

1、公平模式

采用公平锁，并配合一个FIFO队列（Queue）来管理多余的生产者和消费者

2、非公平模式

采用非公平锁，并配合一个LIFO栈（Stack）来管理多余的生产者和消费者，这也是SynchronousQueue默认的模式

利用BlockingQueue实现生产者消费者模型：

public static void main(String[] args)
 {
 final BlockingQueue<String> bq = new ArrayBlockingQueue<String>(10);
 Runnable producerRunnable = new Runnable()
 {
 int i = 0;
 public void run()
 {
 while (true)
 {
 try
 {
 System.out.println("我生产了一个" + i++);
 bq.put(i + "");
 Thread.sleep(1000);
 } 
 catch (InterruptedException e)
 {
 e.printStackTrace();
 }
 }
 }
 };
 Runnable customerRunnable = new Runnable()
 {
 public void run()
 {
 while (true)
 {
 try
 {
 System.out.println("我消费了一个" + bq.take());
 Thread.sleep(3000);
 } 
 catch (InterruptedException e)
 {
 e.printStackTrace();
 }
 }
 }
 };
 Thread producerThread = new Thread(producerRunnable);
 Thread customerThread = new Thread(customerRunnable);
 producerThread.start();
 customerThread.start();
 }
 运行结果：
 我生产了一个0
 2 我消费了一个1
 3 我生产了一个1
 4 我生产了一个2
 5 我消费了一个2
 6 我生产了一个3
 7 我生产了一个4
 8 我生产了一个5
 9 我消费了一个3
 10 我生产了一个6
 11 我生产了一个7
 12 我生产了一个8
 13 我消费了一个4
 14 我生产了一个9
 15 我生产了一个10
 16 我生产了一个11
 17 我消费了一个5
 18 我生产了一个12
 19 我生产了一个13
 20 我生产了一个14
 21 我消费了一个6
 22 我生产了一个15
 23 我生产了一个16
 24 我消费了一个7
 25 我生产了一个17
 26 我消费了一个8
 27 我生产了一个18

分两部分来看输出结果：

1、第1行~第23行。这块BlockingQueue未满，所以生产者随便生产，消费者随便消费，基本上都是生产3个消费1个，消费者消费速度慢

2、第24行~第27行，从前面我们可以看出，生产到16，消费到6，说明到了ArrayBlockingQueue的极限10了，这时候没办法，生产者生产一个ArrayBlockingQueue就满了，所以不能继续生产了，只有等到消费者消费完才可以继续生产。所以之后的打印内容一定是一个生产者、一个消费者

三.使用第三方中间件如redis

模拟高并发操作redis数据  
我们假设用多线程去操作缓存中的price，正常的假设有两个线程来操作，每个线程的逻辑是对price加1，理论值是2，但是实际是两个线程同时取到了0，后面写入的时候都是覆盖，所以price在缓存中的值还是1.

Jedis client = new Jedis("127.0.0.1", 6379);
         ExecutorService threadPool = Executors.newFixedThreadPool(10);
         String key="priceTest";
         client.set(key, "0");
         for(int i=1;i<10;i++){
             threadPool.submit(new Runnable() {
 
                 public void run() {
                     String key="priceTest";
                     Jedis client = new Jedis("127.0.0.1", 6379);
                     //判断key是否存在
                     if(client.exists(key)){
                         int b = Integer.parseInt(client.get(key));
                         System.out.println("======="+b);
                         String a = String.valueOf(b+1);
 
                         client.set(key, a);
                     }else{
 
                     };
 
                 }
             });
 
         }
         System.out.println(client.get(key));

setnx命令  
关于setnx命令，redis操作手册可以看到该命令的效果：  
setnx k v;当k不存在的时候将k的值设置为v,如果k存在，则不做任何操作。如果设置成功返回1，设置失败返回0。如下图，用setnx命令对同一个key设置两次，第一次设置的时候key不存在。

通过setnx我们可以在操作某一个key之前给其加锁（加入给price加锁，那么就设置一个lock.price，其他想操作price的线程都要先判断price是否有锁，如果有则等释放），为了防止某一个操作加完锁而没有释放，所以需要给锁加一个过期时间，自动释放

ExecutorService threadPool = Executors.newFixedThreadPool(10);
         for(int i=1;i<10;i++){
             threadPool.submit(new Runnable() {
                 public void run() {
                     String key="asdasdasdasdasd";
                     Jedis client = new Jedis("127.0.0.1", 6379);
                     //尝试获取锁
                     long lockStatue = client.setnx("lock"+key, "value");
                     System.out.println(lockStatue);
                     //如果获取不到锁
                     while(lockStatue==0){
                         //休眠300ms，再次尝试获取
                         try {
                             Thread.currentThread();
                             Thread.sleep(300);
                             //尝试重新获取锁
                             lockStatue = client.setnx("lock"+key, "value");
                         } catch (InterruptedException e) {
                             e.printStackTrace();
                         }
                     }
                     //如果获取到锁,设置一个三秒的过期时间，防止死锁
                     client.expire("lock"+key, 3000);
                     //判断key是否存在
                     if(client.exists(key)){
                         int b = Integer.parseInt(client.get(key));
                         System.out.println("======="+b);
                         String a = String.valueOf(b+1);
                         client.set(key, a);
                     }else{
                         client.set(key, "0");
                     };
                     System.out.println("开始释放锁");
                     //全部操作成功之后，释放锁
                     client.del("lock"+key);
                 }
             });
         }

如果害怕在setnx之后设置超时时间设置的时候连不上redis，可以将value设置一个时间，然后通过判断时间时候过期来进一步避免死锁。这是就需要getset命令来结合使用。  
  
原文链接：https://blog.csdn.net/weixin\_41098980/article/details/80107038