自旋分佈式鎖實現

時間 2019-11-16

原文原文鏈接

最近都在說AQS，因爲手頭有地方要實現一個自旋分佈式鎖，就不得不說一下ReentrantLock的AQS了，有關鎖的通常特性能夠參考線程，JVM鎖整理 java

AQS的全稱爲AbstractQueuedSynchronizer，抽象隊列同步器web

在ReentrantLock類中，咱們來看一下加鎖是怎麼來實現的。redis

private final Sync sync;

public void lock() {
    sync.lock();
}

這個sync就是一個AQS的子類，而且是一個抽象類websocket

abstract static class Sync extends AbstractQueuedSynchronizer

它的lock()方法是一個抽象方法多線程

abstract void lock();

具體實現sync的是兩個子類，公平鎖類socket

static final class FairSync extends Sync

和非公平鎖類分佈式

static final class NonfairSync extends Sync

這裏咱們主要以非公平鎖來講明，由於咱們日常用的大部分都是非公平鎖，在非公平鎖中，lock()方法的實現以下ide

final void lock() {
    //AQS的內部方法，無鎖競爭AQS中state的狀態，state的初始值爲0，得到鎖的將0變爲1
    if (compareAndSetState(0, 1))
        //競爭到state爲1的將當前線程設爲AQS的獨家主線程
        setExclusiveOwnerThread(Thread.currentThread());
    else
        acquire(1);
}

在AbstractQueuedSynchronizer類中測試

private static final long stateOffset;

在靜態代碼塊中，咱們能夠看到這個stateOffset取的就是state,而且這個state是多線程可見的volatile。ui

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來實現的，咱們能夠用java來模擬該方法

@Slf4j
@Getter
public 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

如今咱們能夠來寫一個支持自旋的分佈式鎖了。

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));
    }
}

常規分佈式鎖能夠參考採用redis token,分佈式鎖的接口冪等性實現

如今咱們來進行一個簡單的測試，先不使用分佈式鎖

咱們在redis中手動設置一個鍵count,0

127.0.0.1:6379> set count 0
OK

咱們的目的是累加這個count，但不能讓其超過10

@Service
public 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();
    }
}

咱們啓動兩個進程，兩個進程啓動完成後，咱們再來看一下該鍵的值。

127.0.0.1:6379> get count
"15"

這個時候咱們能夠看到，在沒有鎖的狀況下，數量超過了10.

如今用分佈式鎖來進行測試。

將count鍵從新設爲0

127.0.0.1:6379> set count 0
OK

@Slf4j
@Service
public 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() + "進入鎖");
                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();
    }
}