利用synchronized和wait，notify实现一个简单的锁。
开始，先定义一个接口Lock
public interface Lock {
/**
* 是否激活
* @return
*/
boolean isActive();
/**
* 释放锁
*/
void release();
}
公共锁管理器主要用于抽象公共，创建锁，清理锁的方法
package demos.spring.core.utils.backOff.lock;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import java.lang.ref.WeakReference;
import java.util.Collection;
import java.util.HashSet;
import java.util.Iterator;
import java.util.Set;
import java.util.concurrent.atomic.AtomicLong;
/**
* creat date:2021/3/14 13:15
* author:xxydliuyss
* note:
*/
public class LockManager {
private static final int INITIAL_WAIT_TO_COLLECT = 10000;
/**
* 最大等待时间为90分钟，
*/
private static final int MAX_WAIT_TO_COLLECT = 90 * 60 * 1000;
/**
* 序列
*/
private static final AtomicLong seq = new AtomicLong();
/**
* 锁对象，弱引用，当system.gc(),未被占用的对象，会被清理掉
*/
private static class WeakLockReference {
/***
* 锁对象别名
*/
String alias;
/**
* 锁对象所有者的名字
*/
String acquiredName;
/**
* 锁对象的占有者，为线程ID
*/
long acquiredId;
/**
* 当前线程的线程堆栈，当发生异常时
*/
Throwable stack;
/**
* 锁对象的弱引用
*/
WeakReference<Lock> reference;
}
private final Log logger = LogFactory.getLog(LockManager.class);
/**
* 控制LockManager是否保存Lock的堆栈信息，主要用于debug
*/
private final boolean trackLocks;
/**
* 垃圾回收释放一激活的锁的清理间隔
*/
private int waitToCollect;
/**
* 存放激活锁的集合
*/
private final Set<WeakLockReference> activeLocks = new HashSet<WeakLockReference>();
public LockManager() {
this(false);
}
public LockManager(boolean trackLocks) {
this(trackLocks, INITIAL_WAIT_TO_COLLECT);
}
/**
*
* @param trackLocks
* 控制是否保存堆栈信息
* @param collectionFrequency
* 清理间隔（ms）
*/
public LockManager(boolean trackLocks, int collectionFrequency) {
this.trackLocks = trackLocks;
this.waitToCollect = collectionFrequency;
}
/**
* 如果有集合中有任何没有释放的锁，都表示当前lockManager一直处于激活态
*
*/
public boolean isActiveLock() {
synchronized (activeLocks) {
return !activeLocks.isEmpty();
}
}
/**
* 阻塞当前线程直到激活的锁的数量为0为止
* @throws InterruptedException
*/
public void waitForActiveLocks()
throws InterruptedException
{
long now = -1;
while (true) {
boolean nochange;
Set<WeakLockReference> before;
synchronized (activeLocks) {
if (activeLocks.isEmpty())
return;
before = new HashSet<WeakLockReference>(activeLocks);
if (now < 0) {
now = System.currentTimeMillis();
}
//等待waitToCollect（ms）
activeLocks.wait(waitToCollect);
if (activeLocks.isEmpty())
return;
nochange = before.equals(activeLocks);
}
// guard against so-called spurious wakeup
//防范虚拟的唤醒
if (nochange && System.currentTimeMillis() - now >= waitToCollect / 2) {
releaseAbandoned();
now = -1;
}
}
}
/**
* 创建新的激活的锁，每调用一次此方法锁的数量会增加一次，当调用lock.realease方法，锁的个数会减一
*/
public synchronized Lock createLock(String alias) {
final WeakLockReference weak = new WeakLockReference();
weak.alias = alias;
weak.acquiredName = Thread.currentThread().getName();
weak.acquiredId = Thread.currentThread().getId();
if (trackLocks) {
weak.stack = new Throwable(
alias + " lock " + seq.incrementAndGet() + " acquired in " + weak.acquiredName);
}
Lock lock = new Lock() {
public synchronized boolean isActive() {
synchronized (activeLocks) {
return activeLocks.contains(weak);
}
}
public synchronized void release() {
synchronized (activeLocks) {
if (activeLocks.remove(weak)) {
activeLocks.notifyAll();
}
}
}
@Override
public String toString() {
if (weak.stack == null) {
return weak.alias + " lock acquired in " + weak.acquiredName;
}
else {
return weak.stack.getMessage();
}
}
};
weak.reference = new WeakReference<Lock>(lock);
synchronized (activeLocks) {
activeLocks.add(weak);
}
return lock;
}
private void releaseAbandoned() {
System.gc();
Thread.yield();
synchronized (activeLocks) {
if (!activeLocks.isEmpty()) {
boolean stalled = true;
Iterator<WeakLockReference> iter = activeLocks.iterator();
while (iter.hasNext()) {
WeakLockReference lock = iter.next();
if (lock.reference.get() == null) {
iter.remove();
activeLocks.notifyAll();
stalled = false;
logAbandonedLock(lock);
}
}
if (stalled) {
//如果么有激活的锁待释放的时候，清理间隔时间会翻倍
// No active locks were found to be abandoned
// wait longer next time before running gc
if (waitToCollect < MAX_WAIT_TO_COLLECT) {
waitToCollect = waitToCollect * 2;
}
logStalledLock(activeLocks);
}
}
}
}
private void logAbandonedLock(WeakLockReference lock) {
if (lock.stack == null && logger.isWarnEnabled()) {
String msg = lock.alias
+ " lock abandoned; lock was acquired in" + lock.acquiredName;
logger.warn(msg);
}
else if (logger.isWarnEnabled()) {
String msg = lock.alias + " lock abandoned; lock was acquired in " + lock.acquiredName;
logger.warn(msg, lock.stack);
}
}
private void logStalledLock(Collection<WeakLockReference> activeLocks) {
Thread current = Thread.currentThread();
if (activeLocks.size() == 1) {
WeakLockReference lock = activeLocks.iterator().next();
if (logger.isWarnEnabled()) {
String msg = "Thread " + current.getName() + " is waiting on an active " + lock.alias
+ " lock acquired in " + lock.acquiredName;
if (lock.acquiredId == current.getId()) {
if (lock.stack == null) {
logger.warn(msg, new Throwable());
}
else {
logger.warn(msg, new Throwable(lock.stack));
}
}
else {
if (lock.stack == null) {
logger.info(msg);
}
else {
logger.info(msg, new Throwable(lock.stack));
}
}
}
}
else {
String alias = null;
boolean warn = false;
for (WeakLockReference lock : activeLocks) {
warn |= lock.acquiredId == current.getId();
if (alias == null) {
alias = lock.alias;
}
else if (!alias.contains(lock.alias)) {
alias = alias + ", " + lock.alias;
}
}
String msg = "Thread " + current.getName() + " is waiting on " + activeLocks.size() + " active "
+ alias + " locks";
if (warn) {
logger.warn(msg);
}
else {
logger.info(msg);
}
}
}
}
独占锁
public class ExclusiveLockManager {
private final LockManager lock;
/**
* Creates an ExclusiveLockManager.
*/
public ExclusiveLockManager() {
this(false);
}
/**
* Creates an ExclusiveLockManager.
*
* @param trackLocks If create stack traces should be logged
*/
public ExclusiveLockManager(boolean trackLocks) {
this.lock = new LockManager(trackLocks);
}
/**
* Gets the exclusive lock, if available. This method will return <tt>null</tt> if the exclusive lock is not
* immediately available.
*/
public Lock tryExclusiveLock() {
if (lock.isActiveLock()) {
return null;
}
synchronized (this) {
if (lock.isActiveLock()) {
return null;
}
return createLock();
}
}
/**
* Gets the exclusive lock. This method blocks when the exclusive lock is currently in use until it is released.
*/
public synchronized Lock getExclusiveLock() throws InterruptedException {
while (lock.isActiveLock()) {
// Someone else currently has the lock
lock.waitForActiveLocks();
}
return createLock();
}
private Lock createLock() {
return lock.createLock("Exclusive");
}
}
使用demo
package demos.spring.core.utils.backOff;
import demos.spring.core.utils.backOff.lock.ExclusiveLockManager;
import demos.spring.core.utils.backOff.lock.Lock;
import demos.spring.core.utils.backOff.lock.LockManager;
import org.springframework.util.backoff.BackOff;
import org.springframework.util.backoff.FixedBackOff;
import javax.swing.plaf.TableHeaderUI;
import java.util.concurrent.TimeUnit;
/**
* creat date:2021/3/13 15:10
* author:xxydliuyss
* note:
*/
public class BackOffDemo {
public static void main(String[] args) {
ThreadDemo A = new ThreadDemo("A",40);
ThreadDemo B = new ThreadDemo("B",60);
A.start();
B.start();
}
public static ExclusiveLockManager lockManager = new ExclusiveLockManager(false);
//利用BackOff机制实现一个获取内存锁的一个场景。
public static class ThreadDemo extends Thread{
private String threadName;
private Integer sleep;
public ThreadDemo(String threadName,int sleep) {
this.threadName = threadName;
this.sleep = sleep;
}
public void run(){
Lock abcd=null;
try {
abcd = lockManager.getExclusiveLock();
if(abcd==null){
System.out.println("thread" + threadName + "未获取到锁");
return;
}
if(abcd.isActive()){
System.out.println("TheadName:" + threadName +"doSomeWorker" );
TimeUnit.SECONDS.sleep(sleep);
System.out.println("TheadName:" + threadName +"doSomeWorke,expend time " + sleep );
}
} catch (InterruptedException e) {
e.printStackTrace();
} finally {
if(abcd!=null) {
abcd.release();
}
}
}
}
}
读写锁
定义一个读写锁的接口。
public interface ReadWriteLockManager {
/**
* Gets a read lock, if available. This method will return null if the read lock is not immediately
* available.
*/
public Lock tryReadLock();
/**
* Gets a read lock. This method blocks until the read lock is available.
*
* @throws InterruptedException In case the thread requesting the lock was {@link Thread#interrupt() interrupted}.
*/
public Lock getReadLock() throws InterruptedException;
/**
* Gets an exclusive write lock, if available. This method will return null if the write lock is not
* immediately available.
*/
public Lock tryWriteLock();
/**
* Gets an exclusive write lock. This method blocks until the write lock is available.
*
* @throws InterruptedException In case the thread requesting the lock was {@link Thread#interrupt() interrupted}.
*/
public Lock getWriteLock() throws InterruptedException;
}
读写锁，有读锁，和写锁两把锁，读写互斥，写写互斥。允许多次读的特性。
public abstract class AbstractReadWriteLockManager implements ReadWriteLockManager {
/**
* Flag indicating whether a writer is active.
*/
private final LockManager activeWriter;
/**
* Counter that keeps track of the numer of active read locks.
*/
private final LockManager activeReaders;
/**
* Creates a MultiReadSingleWriteLockManager.
*/
public AbstractReadWriteLockManager() {
this(false);
}
/**
* Creates a new MultiReadSingleWriteLockManager, optionally with lock tracking enabled.
*
* @param trackLocks Controls whether the lock manager will keep track of active locks. Enabling lock tracking will
* add some overhead, but can be very useful for debugging.
*/
public AbstractReadWriteLockManager(boolean trackLocks) {
boolean trace = trackLocks || Properties.lockTrackingEnabled();
activeWriter = new LockManager(trace);
activeReaders = new LockManager(trace);
}
/**
* If a writer is active
*/
protected boolean isWriterActive() {
return activeWriter.isActiveLock();
}
/**
* If one or more readers are active
*/
protected boolean isReaderActive() {
return activeReaders.isActiveLock();
}
/**
* Blocks current thread until after the writer lock is released (if active).
*
* @throws InterruptedException
*/
protected void waitForActiveWriter() throws InterruptedException {
activeWriter.waitForActiveLocks();
}
/**
* Blocks current thread until there are no reader locks active.
*
* @throws InterruptedException
*/
protected void waitForActiveReaders() throws InterruptedException {
activeReaders.waitForActiveLocks();
}
/**
* Creates a new Lock for reading and increments counter for active readers. The lock is tracked if lock tracking is
* enabled. This method is not thread safe itself, the calling method is expected to handle synchronization issues.
*
* @return a read lock.
*/
protected Lock createReadLock() {
return activeReaders.createLock("Read");
}
/**
* Creates a new Lock for writing. The lock is tracked if lock tracking is enabled. This method is not thread safe
* itself for performance reasons, the calling method is expected to handle synchronization issues.
*
* @return a write lock.
*/
protected Lock createWriteLock() {
return activeWriter.createLock("Write");
}
}
public class ReadPrefReadWriteLockManager extends AbstractReadWriteLockManager {
/**
* Creates a MultiReadSingleWriteLockManager.
*/
public ReadPrefReadWriteLockManager() {
super();
}
/**
* Creates a new MultiReadSingleWriteLockManager, optionally with lock tracking enabled.
*
* @param trackLocks Controls whether the lock manager will keep track of active locks. Enabling lock tracking will
* add some overhead, but can be very useful for debugging.
*/
public ReadPrefReadWriteLockManager(boolean trackLocks) {
super(trackLocks);
}
/*
* --------- Methods ---------
*/
/**
* Gets a read lock, if available. This method will return <tt>null</tt> if the read lock is not immediately
* available.
*/
@Override
public Lock tryReadLock() {
if (isWriterActive()) {
return null;
}
synchronized (this) {
if (isWriterActive()) {
return null;
}
return createReadLock();
}
}
/**
* Gets a read lock. This method blocks when a write lock is in use or has been requested until the write lock is
* released.
*/
@Override
public synchronized Lock getReadLock() throws InterruptedException {
// Wait for the writer to finish
while (isWriterActive()) {
waitForActiveWriter();
}
return createReadLock();
}
/**
* Gets an exclusive write lock, if available. This method will return <tt>null</tt> if the write lock is not
* immediately available.
*/
@Override
public Lock tryWriteLock() {
if (isWriterActive() || isReaderActive()) {
return null;
}
synchronized (this) {
if (isWriterActive() || isReaderActive()) {
return null;
}
return createWriteLock();
}
}
/**
* Gets an exclusive write lock. This method blocks when the write lock is in use or has already been requested
* until the write lock is released. This method also block when read locks are active until all of them are
* released.
*/
@Override
public Lock getWriteLock() throws InterruptedException {
while (true) {
Lock lock = tryWriteLock();
if (lock != null) {
return lock;
}
waitForActiveWriter();
waitForActiveReaders();
}
}
}
另外优化的实现
public class WritePrefReadWriteLockManager extends AbstractReadWriteLockManager {
/*
* ----------- Variables -----------
*/
/**
* Flag indicating whether a write lock has been requested.
*/
private volatile boolean writeRequested = false;
/**
* Creates a MultiReadSingleWriteLockManager.
*/
public WritePrefReadWriteLockManager() {
super();
}
/**
* Creates a new MultiReadSingleWriteLockManager, optionally with lock tracking enabled.
*
* @param trackLocks Controls whether the lock manager will keep track of active locks. Enabling lock tracking will
* add some overhead, but can be very useful for debugging.
*/
public WritePrefReadWriteLockManager(boolean trackLocks) {
super(trackLocks);
}
/**
* Gets a read lock, if available. This method will return null if the read lock is not immediately
* available.
*/
@Override
public Lock tryReadLock() {
if (writeRequested || isWriterActive()) {
return null;
}
synchronized (this) {
if (isWriterActive()) {
return null;
}
return createReadLock();
}
}
/**
* Gets a read lock. This method blocks when a write lock is in use or has been requested until the write lock is
* released.
*/
@Override
public Lock getReadLock() throws InterruptedException {
while (true) {
Lock lock = tryReadLock();
if (lock != null) {
return lock;
}
waitForActiveWriter();
}
}
/**
* Gets an exclusive write lock, if available. This method will return <tt>null</tt> if the write lock is not
* immediately available.
*/
@Override
public Lock tryWriteLock() {
if (isWriterActive() || isReaderActive()) {
return null;
}
synchronized (this) {
if (isWriterActive() || isReaderActive()) {
return null;
}
return createWriteLock();
}
}
/**
* Gets an exclusive write lock. This method blocks when the write lock is in use or has already been requested
* until the write lock is released. This method also block when read locks are active until all of them are
* released.
*/
@Override
public synchronized Lock getWriteLock() throws InterruptedException {
writeRequested = true;
try {
// Wait for the write lock to be released
while (isWriterActive()) {
waitForActiveWriter();
}
// Wait for the read locks to be released
while (isReaderActive()) {
waitForActiveReaders();
}
return createWriteLock();
} finally {
writeRequested = false;
}
}
}
文件锁
public interface LockManager {
/**
* Technical description of where the lock is located, such as a URL.
*/
String getLocation();
/**
* Determines if the SAIL is locked.
*
* @return <code>true</code> if the SAIL is already locked.
*/
boolean isLocked();
/**
* Creates a lock in a SAIL if it does not yet exist.
*
* @return a newly acquired lock or null if the SAIL is already locked.
*/
Lock tryLock();
/**
* Creates a lock in a SAIL if it does not yet exist.
*
* @return a newly acquired lock.
* @throws SailLockedException if the directory is already locked.
*/
Lock lockOrFail() throws SailLockedException;
/**
* Revokes a lock owned by another process.
*
* @return <code>true</code> if a lock was successfully revoked.
*/
boolean revokeLock();
}
实现
public class DirectoryLockManager implements LockManager {
private static final String LOCK_DIR_NAME = "lock";
private static final String LOCK_FILE_NAME = "locked";
private static final String INFO_FILE_NAME = "process";
private final Logger logger = LoggerFactory.getLogger(DirectoryLockManager.class);
private final File dir;
public DirectoryLockManager(File dir) {
this.dir = dir;
}
@Override
public String getLocation() {
return dir.toString();
}
private File getLockDir() {
return new File(dir, LOCK_DIR_NAME);
}
/**
* Determines if the directory is locked.
*
* @return <code>true</code> if the directory is already locked.
*/
@Override
public boolean isLocked() {
return getLockDir().exists();
}
/**
* Creates a lock in a directory if it does not yet exist.
*
* @return a newly acquired lock or null if the directory is already locked.
*/
@Override
public Lock tryLock() {
File lockDir = getLockDir();
if (lockDir.exists()) {
removeInvalidLock(lockDir);
}
if (!lockDir.mkdir()) {
return null;
}
Lock lock = null;
try {
File infoFile = new File(lockDir, INFO_FILE_NAME);
File lockedFile = new File(lockDir, LOCK_FILE_NAME);
RandomAccessFile raf = new RandomAccessFile(lockedFile, "rw");
try {
FileLock fileLock = raf.getChannel().lock();
lock = createLock(raf, fileLock);
sign(infoFile);
} catch (IOException e) {
if (lock != null) {
// Also closes raf
lock.release();
} else {
raf.close();
}
throw e;
}
} catch (IOException e) {
logger.error(e.toString(), e);
}
return lock;
}
/**
* Creates a lock in a directory if it does not yet exist.
*
* @return a newly acquired lock.
* @throws SailLockedException if the directory is already locked.
*/
@Override
public Lock lockOrFail() throws SailLockedException {
Lock lock = tryLock();
if (lock != null) {
return lock;
}
String requestedBy = getProcessName();
String lockedBy = getLockedBy();
if (lockedBy != null) {
throw new SailLockedException(lockedBy, requestedBy, this);
}
lock = tryLock();
if (lock != null) {
return lock;
}
throw new SailLockedException(requestedBy);
}
/**
* Revokes a lock owned by another process.
*
* @return <code>true</code> if a lock was successfully revoked.
*/
@Override
public boolean revokeLock() {
File lockDir = getLockDir();
File lockedFile = new File(lockDir, LOCK_FILE_NAME);
File infoFile = new File(lockDir, INFO_FILE_NAME);
lockedFile.delete();
infoFile.delete();
return lockDir.delete();
}
private void removeInvalidLock(File lockDir) {
try {
boolean revokeLock = false;
File lockedFile = new File(lockDir, LOCK_FILE_NAME);
try (RandomAccessFile raf = new RandomAccessFile(lockedFile, "rw")) {
FileLock fileLock = raf.getChannel().tryLock();
if (fileLock != null) {
logger.warn("Removing invalid lock {}", getLockedBy());
fileLock.release();
revokeLock = true;
}
} catch (OverlappingFileLockException exc) {
// lock is still valid
}
if (revokeLock) {
revokeLock();
}
} catch (IOException e) {
logger.warn(e.toString(), e);
}
}
private String getLockedBy() {
try {
File lockDir = getLockDir();
File infoFile = new File(lockDir, INFO_FILE_NAME);
try (BufferedReader reader = new BufferedReader(new FileReader(infoFile))) {
return reader.readLine();
}
} catch (IOException e) {
logger.warn(e.toString(), e);
return null;
}
}
private Lock createLock(final RandomAccessFile raf, final FileLock fileLock) {
return new Lock() {
private Thread hook;
{
try {
Thread hook = new Thread(this::delete);
Runtime.getRuntime().addShutdownHook(hook);
this.hook = hook;
} catch (AccessControlException e) {
// okay, just remember to close it yourself
}
}
@Override
public boolean isActive() {
return fileLock.isValid() || hook != null;
}
@Override
public void release() {
try {
if (hook != null) {
Runtime.getRuntime().removeShutdownHook(hook);
hook = null;
}
} catch (IllegalStateException e) {
// already shutting down
} catch (AccessControlException e) {
logger.warn(e.toString(), e);
}
delete();
}
synchronized void delete() {
try {
if (raf.getChannel().isOpen()) {
fileLock.release();
raf.close();
}
} catch (IOException e) {
logger.warn(e.toString(), e);
}
revokeLock();
}
};
}
private void sign(File infoFile) throws IOException {
try (FileWriter out = new FileWriter(infoFile)) {
out.write(getProcessName());
out.flush();
}
}
private String getProcessName() {
return ManagementFactory.getRuntimeMXBean().getName();
}
}
注意以上代码来源于互联网，仅觉得写的还可以，我作为搬运工而已...