您好,登錄后才能下訂單哦!
如何實(shí)現(xiàn)一個(gè)自旋分布式鎖,很多新手對(duì)此不是很清楚,為了幫助大家解決這個(gè)難題,下面小編將為大家詳細(xì)講解,有這方面需求的人可以來學(xué)習(xí)下,希望你能有所收獲。
AQS的全稱為AbstractQueuedSynchronizer,抽象隊(duì)列同步器
在ReentrantLock類中,我們來看一下加鎖是怎么來實(shí)現(xiàn)的。
private final Sync sync;
public void lock() {sync.lock();}
這個(gè)sync就是一個(gè)AQS的子類,并且是一個(gè)抽象類
abstract static class Sync extends AbstractQueuedSynchronizer
它的lock()方法是一個(gè)抽象方法
abstract void lock();
具體實(shí)現(xiàn)sync的是兩個(gè)子類,公平鎖類
static final class FairSync extends Sync
和非公平鎖類
static final class NonfairSync extends Sync
這里我們主要以非公平鎖來說明,因?yàn)槲覀兤匠S玫拇蟛糠侄际欠枪芥i,在非公平鎖中,lock()方法的實(shí)現(xiàn)如下
final void lock() { //AQS的內(nèi)部方法,無鎖競爭AQS中state的狀態(tài),state的初始值為0,獲得鎖的將0變?yōu)?if (compareAndSetState(0, 1)) //競爭到state為1的將當(dāng)前線程設(shè)為AQS的獨(dú)家主線程setExclusiveOwnerThread(Thread.currentThread()); else acquire(1);}
在AbstractQueuedSynchronizer類中
private static final long stateOffset;
在靜態(tài)代碼塊中,我們可以看到這個(gè)stateOffset取的就是state,并且這個(gè)state是多線程可見的volatile。
stateOffset = unsafe.objectFieldOffset (AbstractQueuedSynchronizer.class.getDeclaredField("state"));
private volatile int state;
protected final boolean compareAndSetState(int expect, int update) {// See below for intrinsics setup to support this return unsafe.compareAndSwapInt(this, stateOffset, expect, update);}
private transient Thread exclusiveOwnerThread;
protected final void setExclusiveOwnerThread(Thread thread) {exclusiveOwnerThread = thread;}
這里unsafe.compareAndSwapInt()是用C來實(shí)現(xiàn)的,我們可以用java來模擬該方法
@Slf4j@Getterpublic class GetState {private AtomicReference<Integer> state = new AtomicReference<>(0); private boolean lockState() {while (true) {if (state.compareAndSet(0,1)) {return true; } } }private void unlockState() {state.set(0); }@AllArgsConstructor private static class Task implements Runnable {private GetState getState; @Override public void run() {if (getState.lockState()) {log.info(Thread.currentThread().getName() + "獲取鎖"); } } }public static void main(String[] args) throws InterruptedException { ExecutorService service = Executors.newFixedThreadPool(16); GetState state = new GetState(); for (int i = 0;i < 10;i++) { service.execute(new Task(state)); }while (state.getState().get() == 1) { Thread.sleep(1000); state.unlockState(); } service.shutdown(); } }
打印日志(每秒打印一條)
15:35:42.953 [pool-1-thread-1] INFO com.guanjian.websocket.tomic.GetState - pool-1-thread-1獲取鎖
15:35:43.953 [pool-1-thread-9] INFO com.guanjian.websocket.tomic.GetState - pool-1-thread-9獲取鎖
15:35:44.957 [pool-1-thread-5] INFO com.guanjian.websocket.tomic.GetState - pool-1-thread-5獲取鎖
15:35:45.962 [pool-1-thread-2] INFO com.guanjian.websocket.tomic.GetState - pool-1-thread-2獲取鎖
15:35:46.962 [pool-1-thread-7] INFO com.guanjian.websocket.tomic.GetState - pool-1-thread-7獲取鎖
15:35:47.962 [pool-1-thread-3] INFO com.guanjian.websocket.tomic.GetState - pool-1-thread-3獲取鎖
15:35:48.967 [pool-1-thread-8] INFO com.guanjian.websocket.tomic.GetState - pool-1-thread-8獲取鎖
15:35:49.969 [pool-1-thread-6] INFO com.guanjian.websocket.tomic.GetState - pool-1-thread-6獲取鎖
15:35:50.970 [pool-1-thread-4] INFO com.guanjian.websocket.tomic.GetState - pool-1-thread-4獲取鎖
15:35:51.971 [pool-1-thread-10] INFO com.guanjian.websocket.tomic.GetState - pool-1-thread-10獲取鎖
Process finished with exit code 0
現(xiàn)在我們可以來寫一個(gè)支持自旋的分布式鎖了。
public class SpinDistributedLock {private volatile AtomicReference<Boolean> state = new AtomicReference<>(false); public boolean lock(RedisService redisService,String key,String value,int expire) {while (true) {if (state.compareAndSet(false, RedisTool.tryGetDistributedLock(redisService,key,value,expire))) {if (state.get()) {return true; } } } }public void unlock(RedisService redisService,String key,String value) {state.set(!RedisTool.releaseDistributedLock(redisService,key,value)); } }
常規(guī)分布式鎖可以參考采用redis token,分布式鎖的接口冪等性實(shí)現(xiàn)
現(xiàn)在我們來進(jìn)行一個(gè)簡單的測試,先不使用分布式鎖
我們?cè)趓edis中手動(dòng)設(shè)置一個(gè)鍵count,0
127.0.0.1:6379> set count 0
OK
我們的目的是累加這個(gè)count,但不能讓其超過10
@Servicepublic class NoDistributedTest {@Autowired private RedisService redisService; private class Task implements Runnable {@Override public void run() {if (Integer.valueOf(redisService.get("count")) < 10) {redisService.incr("count"); } } }@PostConstruct public void test() { ExecutorService service = Executors.newFixedThreadPool(16); for (int i = 0;i < 100000;i++) { service.execute(new Task()); } service.shutdown(); } }
我們啟動(dòng)兩個(gè)進(jìn)程,兩個(gè)進(jìn)程啟動(dòng)完成后,我們?cè)賮砜匆幌略撴I的值。
127.0.0.1:6379> get count
"15"
這個(gè)時(shí)候我們可以看到,在沒有鎖的情況下,數(shù)量超過了10.
現(xiàn)在用分布式鎖來進(jìn)行測試。
將count鍵重新設(shè)為0
127.0.0.1:6379> set count 0
OK
@Slf4j@Servicepublic class DistributedTest {private SpinDistributedLock lock = new SpinDistributedLock(); @Autowired private RedisService redisService; private class Task implements Runnable {@Override public void run() {try {lock.lock(redisService,"countlock","countlock",3); log.info(Thread.currentThread().getName() + "進(jìn)入鎖"); if (Integer.valueOf(redisService.get("count")) < 10) {redisService.incr("count"); } } finally {lock.unlock(redisService,"countlock","countlock"); log.info(Thread.currentThread().getName() + "釋放鎖"); } } }@PostConstruct public void test() { ExecutorService service = Executors.newFixedThreadPool(16); for (int i = 0;i < 100000;i++) { service.execute(new Task()); } service.shutdown(); } }
同樣啟動(dòng)兩個(gè)進(jìn)程或者更多進(jìn)程,啟動(dòng)完成后,我們來看一下count鍵的值
127.0.0.1:6379> get count
"10"
看完上述內(nèi)容是否對(duì)您有幫助呢?如果還想對(duì)相關(guān)知識(shí)有進(jìn)一步的了解或閱讀更多相關(guān)文章,請(qǐng)關(guān)注億速云行業(yè)資訊頻道,感謝您對(duì)億速云的支持。
免責(zé)聲明:本站發(fā)布的內(nèi)容(圖片、視頻和文字)以原創(chuàng)、轉(zhuǎn)載和分享為主,文章觀點(diǎn)不代表本網(wǎng)站立場,如果涉及侵權(quán)請(qǐng)聯(lián)系站長郵箱:is@yisu.com進(jìn)行舉報(bào),并提供相關(guān)證據(jù),一經(jīng)查實(shí),將立刻刪除涉嫌侵權(quán)內(nèi)容。