In this article, we use mysql to implement a distributed lock.
Environment: mysql8, navicat, maven, springboot2.3.11, mybatis-plus
The function of distributed lock
1. The distributed lock user is located in different machines. Only after the lock is successfully acquired can the shared resources be operated.
2. The lock has the function of reentry: that is, a user can acquire a certain lock multiple times
3. Acquisition lock has a timeout function: that is, try to acquire the lock within the specified time. If the timeout time is exceeded, if the acquisition is not successful, the acquisition will be failed.
4. It can be automatically fault-tolerant. For example: after machine A acquires lock lock1, before releasing lock1, machine A hangs up, resulting in lock lock1 not being released, and lock1 will be occupied by machine A. In this case, the distributed lock must be automatically solved. You can do this: when holding the lock, you can add a holding timeout time. If this time has not been released, other machines will have the opportunity to acquire the lock.
Preparation skills: Optimistic lock
Usually, we modify a data in the table as follows:
t1：selectGet recordsR1 t2：rightR1Make edits t3：update R1
Let's take a look at the problems in the above process:
If the two threads A and B execute to t1 at the same time, they both see the same data of R1, and then edit R1, and then execute t3. In the end, the two threads will be updated successfully, and the next thread will overwrite the result of the update of the previous thread. This is the problem of concurrent modification of data.
We can add a new version number to the table, use the version number as a condition every time the data is updated, and the version number +1 every time the update is updated. The process is optimized as follows:
t1：Open a transactionstart transaction
t2：selectGet recordsR1,Declare variablesv=
t3：rightR1Make edits
t4：Perform update operations
update R1 set version = version + 1 where user_id=#user_id# and version = #v#;
t5：t4In-houseupdateReturns the number of rows affected，We record it incountmiddle，Then according tocountTo determine whether to submit or rollback
if(count==1){
//Submit transaction commit;
}else{
//Rolleate the transaction rollback;
}
The above focus is on step t4. When multiple threads execute to t1 at the same time, the R1 they see is the same. However, when they execute to t4, the database will lock the update record to ensure that it is queued for execution in concurrency, so only the first update will return 1, and the other update results will return 0, and then it will determine whether the count is 1, and then submit or roll back the transaction. You can use the count value to know whether the data has been modified successfully.
The above method is optimistic. We can ensure the correctness of the data concurrent modification process through optimistic locking.
Use mysql to implement distributed lock
We create a distributed lock table as follows
DROP TABLE IF EXISTS t_lock; create table t_lock( lock_key varchar(32) PRIMARY KEY NOT NULL COMMENT 'Lock only logo', request_id varchar(64) NOT NULL DEFAULT '' COMMENT 'Used to identify the request object', lock_count INT NOT NULL DEFAULT 0 COMMENT 'Current lock times', timeout BIGINT NOT NULL DEFAULT 0 COMMENT 'Lock timeout time', version INT NOT NULL DEFAULT 0 COMMENT 'Version number，Every update+1' )COMMENT 'Lock information table';
The java code is as follows
mapper interface
package ;
import ;
import ;
import ;
/**
* @description: Lock mapper
* @author: stone
* @date: Created by 2021/5/30 11:12
* @version: 1.0.0
* @pakeage:
*/
@Repository
public interface LockMapper extends BaseMapper<LockModel> {
}
lock object model
package ;
import ;
import ;
import ;
import ;
/**
* @description: lock model
* @author: stone
* @date: Created by 2021/9/10 11:13
* @version: 1.0.0
* @pakeage:
*/
@Data
@TableName("t_lock")
public class LockModel {
/**
* The unique value of the lock
*/
@TableId
private String lockKey;
/**
* Request id, the same request id in the same thread
*/
private String requestId;
//Locks private Integer lockCount;
//Lock timeout private Long timeout;
//Optimistic lock version @Version
private Integer version;
}
Lock interface
package ;
/**
* @description: Lock interface
* @author: stone
* @date: Created by 2021/9/10 11:40
* @version: 1.0.0
* @pakeage:
*/
public interface ILock<T> {
/**
* Acquisition of distributed locks, support reentry
* @param lockKey lock can
* @param lockTimeout The validity time of holding the lock to prevent deadlocks
* @param getTimeout Get lock timeout,
* @return Whether the lock is successful
*/
public boolean lock(String lockKey, long lockTimeout, int getTimeout) throws Exception;
/**
* Unlock
* @param lockKey lock key
*
*/
public void unlock(String lockKey);
/**
* Reset the lock object
* @param t lock object
* @return Return to lock record
*/
public int restLock(T t);
}
The implementation code of the lock is as follows
package ;
import ;
import ;
import ;
import .slf4j.Slf4j;
import ;
import ;
import ;
import ;
import ;
/**
* @description: mysql implements distributed lock
* @author: stone
* @date: Created by 2021/9/10 11:09
* @version: 1.0.0
* @pakeage:
*/
@Component
@Slf4j
public class MysqlLock implements ILock<LockModel>{
static ThreadLocal<String> requestIds = new ThreadLocal<>();
@Autowired
private LockMapper lockMapper;
public String getRequestId() {
String requestId = ();
if ((requestId)) {
requestId = ().toString();
(requestId);
}
("ObtainedrequestId===> {}", requestId);
return requestId;
}
/**
* Acquire the lock
* @param lockKey lock can
* @param lockTimeout The validity time of holding the lock to prevent deadlocks
* @param getTimeout Get lock timeout,
* @return
*/
@Override
public boolean lock(String lockKey, long lockTimeout, int getTimeout) throws Exception {
(" lock start =======================> {}",lockKey);
//Get request id from local String requestId = ();
//Acquiring the lock result boolean lockResult = false;
//Start time long startTime = ();
while (true) {
LockModel lockModel = (lockKey);
if ((lockModel)) {
//Acquiring the request id of the lock object String reqId = ();
//If it is empty, it means that the lock is not occupied if ((reqId)) {
//Own it now //Set request id (requestId);
//Set the number of locks (1);
//Set the timeout time to prevent deadlock (() + lockTimeout);
if ((lockModel) == 1) {
lockResult = true;
break;
}
}
//If request_id is the same as request_id in the table, it means that the lock is the current thread holder, then the lock needs to be re-entered at this time else if ((reqId)) {
//Reenter lock (() + lockTimeout);
//Set to get the first time (() + 1);
if ((lockModel) == 1) {
lockResult = true;
break;
}
}
//It is not empty, nor is it equal, it means it is owned by other threads else {
//The lock is not its own and has timed out, reset the lock and continue to try again if (() < ()) {
//Timeout has not expired, continue to try again (lockModel);
}
//If the timeout does not occur, sleep for 100ms and continue to try again else {
if (startTime + getTimeout > ()) {
(100);
}
else {
//Prevent long-term blockage break;
}
}
}
}
//If it is empty, insert a lock and try to obtain the lock again else {
lockModel = new LockModel();
//Set lock key (lockKey);
(lockModel);
}
}
(" lock end =======================> {}",lockKey);
return lockResult;
}
/**
* Release the lock
* @param lockKey lock key
*/
@Override
public void unlock(String lockKey) {
LockModel lockModel = (lockKey);
//Get the request id of the current thread String reqId = ();
//Acquiring locks int count = 0;
//The current thread requestId is the same as the request_id in the library && lock_count>0, indicating that the lock can be released if ((lockModel)
&& (())
&& (count = ()) > 0) {
if (count == 1) {
//Reset the lock (lockModel);
}
//The problem of reentering the lock, reduce the number of locks by one else {
(() - 1);
//Number of updates (lockModel);
}
}
}
/**
* Reset the lock
* @param lockModel lock object
* @return Update number
*/
@Override
public int restLock(LockModel lockModel) {
(0);
("");
(0L);
return (lockModel);
}
}
The above code implements all the functions of the distributed lock column at the beginning of the article. You can carefully study the method of obtaining the lock: lock, and the method of releasing the lock: unlock.
Test cases
package ;
import ;
import ;
import .slf4j.Slf4j;
import ;
import ;
import ;
/**
* @description: Lock test
* @author: stone
* @date: Created by 2021/9/10 15:32
* @version: 1.0.0
* @pakeage:
*/
@SpringBootTest
@Slf4j
public class LockApplicationTests {
@Autowired
private ILock<LockModel> mysqlLock;
Test repetition and repetition release
@Test
public void testRepeat() throws Exception {
for (int i = 0; i < 10; i++) {
("key1", 10000L, 1000);
}
for (int i = 0; i < 10; i++) {
("key1");
}
}
// //Not released after obtaining, and is retrieved by thread1 after timeout @Test
public void testTimeout() throws Exception {
String lockKey = "key2";
(lockKey, 5000L, 1000);
Thread thread1 = new Thread(() -> {
try {
(lockKey, 5000L, 7000);
} catch (Exception e) {
();
} finally {
(lockKey);
}
}, "thread1");
();
();
}
}
The test1 method tests the effect of reentry locks.
Test2 tests that the main thread has not been released after acquiring the lock. After the hold lock timeout, it was acquired by thread1.
Leave a question for everyone
A distributed lock above also needs to consider one problem: For example, A chance to acquire the lock of key1 and set the timeout time of holding the lock to 10 seconds, but after acquiring the lock, a business operation was performed, and the business operation took more than 10 seconds. At this time, machine B can obtain the lock successfully when it acquires the lock. At this time, both machines A and B will successfully acquire the lock and are both performing business operations. How should this situation be handled? You can think about it and leave a message, let’s discuss it together.
This is the end of this article about the sample code of MySql to implement distributed locks. For more related content of MySql distributed locks, please search for my previous articles or continue browsing the related articles below. I hope everyone will support me in the future!