本篇内容主要讲解“Java根据某个key加锁怎么实现”,感兴趣的朋友不妨来看看。本文介绍的方法操作简单快捷,实用性强。下面就让小编来带大家学习“Java根据某个key加锁怎么实现”吧!
日常开发中,有时候需要根据某个 key 加锁,确保多线程情况下,对该 key 的加锁和解锁之间的代码串行执行。
大家可以借助每个 key 对应一个 ReentrantLock ,让同一个 key 的线程使用该 lock 加锁;每个 key 对应一个 Semaphore ,让同一个 key 的线程使用 Semaphore 控制同时执行的线程数。
接口定义
public interface LockByKey<T> { /** * 加锁 */ void lock(T key); /** * 解锁 */ void unlock(T key); }
每个 key 对应一个 ReentrantLock ,让同一个 key 的线程使用该 lock 加锁。
import java.util.Map; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.locks.ReentrantLock; public class DefaultLockByKeyImpl<T> implements LockByKey<T> { private final Map<T, ReentrantLock> lockMap = new ConcurrentHashMap<>(); /** * 加锁 */ @Override public void lock(T key) { // 如果key为空,直接返回 if (key == null) { throw new IllegalArgumentException("key 不能为空"); } // 获取或创建一个ReentrantLock对象 ReentrantLock lock = lockMap.computeIfAbsent(key, k -> new ReentrantLock()); // 获取锁 lock.lock(); } /** * 解锁 */ @Override public void unlock(T key) { // 如果key为空,直接返回 if (key == null) { throw new IllegalArgumentException("key 不能为空"); } // 从Map中获取锁对象 ReentrantLock lock = lockMap.get(key); // 获取不到报错 if (lock == null) { throw new IllegalArgumentException("key " + key + "尚未加锁"); } // 其他线程非法持有不允许释放 if (!lock.isHeldByCurrentThread()) { throw new IllegalStateException("当前线程尚未持有,key:" + key + "的锁,不允许释放"); } lock.unlock(); } }
注意事项:
(1)参数合法性校验
(2)解锁时需要判断该锁是否为当前线程持有
import com.google.common.collect.Lists; import org.junit.Test; import java.util.HashSet; import java.util.List; import java.util.Set; import java.util.concurrent.CountDownLatch; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.TimeUnit; public class DefaultLockByKeyImplTest { private final LockByKey<String> lockByKey = new DefaultLockByKeyImpl<>(); private final CountDownLatch countDownLatch = new CountDownLatch(7); private final ExecutorService executorService = Executors.newFixedThreadPool(10); @Test public void test() throws InterruptedException { List<String> keys = Lists.newArrayList("a", "a", "a", "b", "c", "b", "d"); Set<String> executingKeySet = new HashSet<>(); for (int i = 0; i < keys.size(); i++) { String key = keys.get(i); int finalI = i; executorService.submit(() -> { lockByKey.lock(key); if (executingKeySet.contains(key)) { throw new RuntimeException("存在正在执行的 key:" + key); } executingKeySet.add(key); try { System.out.println("index:" + finalI + "对 [" + key + "] 加锁 ->" + Thread.currentThread().getName()); TimeUnit.SECONDS.sleep(1); } catch (InterruptedException e) { throw new RuntimeException(e); } finally { System.out.println("index:" + finalI + "释放 [" + key + "] ->" + Thread.currentThread().getName()); lockByKey.unlock(key); executingKeySet.remove(key); countDownLatch.countDown(); } }); } countDownLatch.await(); } }
如果同一个 key 没释放能够再次进入,会抛出异常。
也可以通过日志来观察执行情况:
index:0对 [a] 加锁 ->pool-1-thread-1 index:6对 [d] 加锁 ->pool-1-thread-7 index:4对 [c] 加锁 ->pool-1-thread-5 index:3对 [b] 加锁 ->pool-1-thread-4 index:6释放 [d] ->pool-1-thread-7 index:4释放 [c] ->pool-1-thread-5 index:0释放 [a] ->pool-1-thread-1 index:3释放 [b] ->pool-1-thread-4 index:1对 [a] 加锁 ->pool-1-thread-2 index:5对 [b] 加锁 ->pool-1-thread-6 index:1释放 [a] ->pool-1-thread-2 index:5释放 [b] ->pool-1-thread-6 index:2对 [a] 加锁 ->pool-1-thread-3 index:2释放 [a] ->pool-1-thread-3
每个 key 对应一个 Semaphore ,让同一个 key 的线程使用 Semaphore 控制同时执行的线程数。
import lombok.SneakyThrows; import java.util.Map; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.Semaphore; public class SimultaneousEntriesLockByKey<T> implements LockByKey<T> { private final Map<T, Semaphore> semaphores = new ConcurrentHashMap<>(); /** * 最大线程 */ private int allowed_threads; public SimultaneousEntriesLockByKey(int allowed_threads) { this.allowed_threads = allowed_threads; } /** * 加锁 */ @Override public void lock(T key) { Semaphore semaphore = semaphores.compute(key, (k, v) -> v == null ? new Semaphore(allowed_threads) : v); semaphore.acquireUninterruptibly(); } /** * 解锁 */ @Override public void unlock(T key) { // 如果key为空,直接返回 if (key == null) { throw new IllegalArgumentException("key 不能为空"); } // 从Map中获取锁对象 Semaphore semaphore = semaphores.get(key); if (semaphore == null) { throw new IllegalArgumentException("key " + key + "尚未加锁"); } semaphore.release(); if (semaphore.availablePermits() >= allowed_threads) { semaphores.remove(key, semaphore); } }
import com.google.common.collect.Lists; import org.junit.Test; import java.time.LocalDateTime; import java.util.Collections; import java.util.HashMap; import java.util.List; import java.util.Map; import java.util.concurrent.CountDownLatch; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.TimeUnit; public class SimultaneousEntriesLockByKeyTest { private final int maxThreadEachKey = 2; private final LockByKey<String> lockByKey = new SimultaneousEntriesLockByKey<>(maxThreadEachKey); private final CountDownLatch countDownLatch = new CountDownLatch(7); private final ExecutorService executorService = Executors.newFixedThreadPool(10); @Test public void test() throws InterruptedException { List<String> keys = Lists.newArrayList("a", "a", "a", "b", "c", "b", "d"); Map<String, Integer> executingKeyCount = Collections.synchronizedMap(new HashMap<>()); for (int i = 0; i < keys.size(); i++) { String key = keys.get(i); int finalI = i; executorService.submit(() -> { lockByKey.lock(key); executingKeyCount.compute(key, (k, v) -> { if (v != null && v + 1 > maxThreadEachKey) { throw new RuntimeException("超过限制了"); } return v == null ? 1 : v + 1; }); try { System.out.println("time:" + LocalDateTime.now().toString() + " ,index:" + finalI + "对 [" + key + "] 加锁 ->" + Thread.currentThread().getName() + "count:" + executingKeyCount.get(key)); TimeUnit.SECONDS.sleep(1); } catch (InterruptedException e) { throw new RuntimeException(e); } finally { System.out.println("time:" + LocalDateTime.now().toString() + " ,index:" + finalI + "释放 [" + key + "] ->" + Thread.currentThread().getName() + "count:" + (executingKeyCount.get(key) - 1)); lockByKey.unlock(key); executingKeyCount.compute(key, (k, v) -> v - 1); countDownLatch.countDown(); } }); } countDownLatch.await(); } }
输出:
time:2023-03-15T20:49:57.044195 ,index:6对 [d] 加锁 ->pool-1-thread-7count:1
time:2023-03-15T20:49:57.058942 ,index:5对 [b] 加锁 ->pool-1-thread-6count:2
time:2023-03-15T20:49:57.069789 ,index:1对 [a] 加锁 ->pool-1-thread-2count:2
time:2023-03-15T20:49:57.042402 ,index:4对 [c] 加锁 ->pool-1-thread-5count:1
time:2023-03-15T20:49:57.046866 ,index:0对 [a] 加锁 ->pool-1-thread-1count:2
time:2023-03-15T20:49:57.042991 ,index:3对 [b] 加锁 ->pool-1-thread-4count:2
time:2023-03-15T20:49:58.089557 ,index:0释放 [a] ->pool-1-thread-1count:1
time:2023-03-15T20:49:58.082679 ,index:6释放 [d] ->pool-1-thread-7count:0
time:2023-03-15T20:49:58.084579 ,index:4释放 [c] ->pool-1-thread-5count:0
time:2023-03-15T20:49:58.083462 ,index:5释放 [b] ->pool-1-thread-6count:1
time:2023-03-15T20:49:58.089576 ,index:3释放 [b] ->pool-1-thread-4count:1
time:2023-03-15T20:49:58.085359 ,index:1释放 [a] ->pool-1-thread-2count:1
time:2023-03-15T20:49:58.096912 ,index:2对 [a] 加锁 ->pool-1-thread-3count:1
time:2023-03-15T20:49:59.099935 ,index:2释放 [a] ->pool-1-thread-3count:0
到此,相信大家对“Java根据某个key加锁怎么实现”有了更深的了解,不妨来实际操作一番吧!这里是亿速云网站,更多相关内容可以进入相关频道进行查询,关注我们,继续学习!
免责声明:本站发布的内容(图片、视频和文字)以原创、转载和分享为主,文章观点不代表本网站立场,如果涉及侵权请联系站长邮箱:is@yisu.com进行举报,并提供相关证据,一经查实,将立刻删除涉嫌侵权内容。