分布式锁理解
1.业务场景引入
在进行代码实现之前，我们先来看一个业务场景：
系统A是一个电商系统，目前是一台机器部署，系统中有一个用户下订单的接口，但是用户下订单之前一定要去检查一下库存，确保库存足够了才会给用户下单。
由于系统有一定的并发，所以会预先将商品的库存保存在redis中，用户下单的时候会更新redis的库存。
此时系统架构如下：
但是这样一来会产生一个问题：
假如某个时刻，redis里面的某个商品库存为1，此时两个请求同时到来，其中一个请求执行到上图的第3步，更新数据库的库存为0，但是第4步还没有执行。
而另外一个请求执行到了第2步，发现库存还是1，就继续执行第3步。
这样的结果，是导致卖出了2个商品，然而其实库存只有1个。
很明显不对啊！这就是典型的库存超卖问题
此时，我们很容易想到解决方案：用锁把2、3、4步锁住，让他们执行完之后，另一个线程才能进来执行第2步。
按照上面的图，在执行第2步时，使用Java提供的synchronized或者ReentrantLock来锁住，然后在第4步执行完之后才释放锁。
这样一来，2、3、4 这3个步骤就被“锁”住了，多个线程之间只能串行化执行。
但是好景不长，整个系统的并发飙升，一台机器扛不住了。现在要增加一台机器，如下图：
增加机器之后，系统变成上图所示，我的天！
假设此时两个用户的请求同时到来，但是落在了不同的机器上，那么这两个请求是可以同时执行了，还是会出现库存超卖的问题。
为什么呢？
因为上图中的两个A系统，运行在两个不同的JVM里面，他们加的锁只对属于自己JVM里面的线程有效，对于其他JVM的线程是无效的。
因此，这里的问题是：Java提供的原生锁机制在多机部署场景下失效了
这是因为两台机器加的锁不是同一个锁(两个锁在不同的JVM里面)。
那么，我们只要保证两台机器加的锁是同一个锁，问题不就解决了吗？
此时，就该分布式锁隆重登场了，分布式锁的思路是：
任何场景下，一旦加锁，效率不可能会高，数据是安全的！
在整个分布式系统提供一个全局、唯一的获取锁的“东西”，然后每个系统在需要加锁时，都去问这个“东西”拿到一把锁，这样不同的系统拿到的就可以认为是同一把锁。
至于这个“东西”，可以是Redis、Zookeeper，也可以是数据库。
通过上面的分析，我们知道了库存超卖场景在分布式部署系统的情况下使用Java原生的锁机制无法保证线程安全，所以我们需要用到分布式锁的方案。
那么，如何实现分布式锁呢？
2.分布式锁的实现
2.1.分布式锁的简单实现代码
import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisPool;
import redis.clients.jedis.Transaction;
import redis.clients.jedis.exceptions.JedisException;
import java.util.List;
import java.util.UUID;
/**
* @Description: 分布式锁简单实现
*/
public class DistributedLock {
private final JedisPool jedisPool;
public DistributedLock(JedisPool jedisPool) {
this.jedisPool = jedisPool;
}
/**
* 加锁
* @param lockName 锁的key
* @param acquireTimeout 获取超时时间
* @param timeout 锁的超时时间
* @return 锁标识
*/
public String lockWithTimeout(String lockName, long acquireTimeout, long timeout) {
Jedis conn = null;
String retIdentifier = null;
try {
// 获取连接
conn = jedisPool.getResource();
// 随机生成一个value
String identifier = UUID.randomUUID().toString();
// 锁名，即key值
String lockKey = "lock:" + lockName;
// 超时时间，上锁后超过此时间则自动释放锁
int lockExpire = (int) (timeout / 1000);
// 获取锁的超时时间，超过这个时间则放弃获取锁
long end = System.currentTimeMillis() + acquireTimeout;
while (System.currentTimeMillis() < end) {
if (conn.setnx(lockKey, identifier) == 1) {
conn.expire(lockKey, lockExpire);
// 返回value值，用于释放锁时间确认
retIdentifier = identifier;
return retIdentifier;
}
// 返回-1代表key没有设置超时时间，为key设置一个超时时间
if (conn.ttl(lockKey) == -1) {
conn.expire(lockKey, lockExpire);
}
try {
Thread.sleep(10);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
}
} catch (JedisException e) {
e.printStackTrace();
} finally {
if (conn != null) {
conn.close();
}
}
return retIdentifier;
}
/**
* 释放锁
* @param lockName 锁的key
* @param identifier 释放锁的标识
* @return
*/
public boolean releaseLock(String lockName, String identifier) {
Jedis conn = null;
String lockKey = "lock:" + lockName;
boolean retFlag = false;
try {
conn = jedisPool.getResource();
while (true) {
// 监视lock，准备开始事务
conn.watch(lockKey);
// 通过前面返回的value值判断是不是该锁，若是该锁，则删除，释放锁
if (identifier.equals(conn.get(lockKey))) {
Transaction transaction = conn.multi();
transaction.del(lockKey);
List<Object> results = transaction.exec();
if (results == null) {
continue;
}
retFlag = true;
}
conn.unwatch();
break;
}
} catch (JedisException e) {
e.printStackTrace();
} finally {
if (conn != null) {
conn.close();
}
}
return retFlag;
}
}
2.2.测试刚才实现的分布式锁
例子中使用50个线程模拟秒杀一个商品，使用–运算符来实现商品减少，从结果有序性就可以看出是否为加锁状态。
模拟秒杀服务，在其中配置了jedis线程池，在初始化的时候传给分布式锁，供其使用。
package com.bruceliu.lock;
import redis.clients.jedis.JedisPool;
import redis.clients.jedis.JedisPoolConfig;
/**
* @Description: TODO
*/
public class SkillService {
private static JedisPool pool = null;
private DistributedLock lock = new DistributedLock(pool);
int n = 500;
static {
JedisPoolConfig config = new JedisPoolConfig();
// 设置最大连接数
config.setMaxTotal(200);
// 设置最大空闲数
config.setMaxIdle(8);
// 设置最大等待时间
config.setMaxWaitMillis(1000 * 100);
// 在borrow一个jedis实例时，是否需要验证，若为true，则所有jedis实例均是可用的
config.setTestOnBorrow(true);
pool = new JedisPool(config, "127.0.0.1", 6379, 3000);
}
public void seckill() {
// 返回锁的value值，供释放锁时候进行判断
String identifier = lock.lockWithTimeout("resource", 5000, 1000);
System.out.println(Thread.currentThread().getName() + "获得了锁");
System.out.println(--n);
lock.releaseLock("resource", identifier);
}
}
2.3.模拟线程进行秒杀服务
package com.bruceliu.lock;
/**
* @Description: TODO
*/
public class TestLock {
public static void main(String[] args) {
SkillService service = new SkillService();
for (int i = 0; i < 50; i++) {
ThreadA threadA = new ThreadA(service);
threadA.setName("ThreadNameA->"+i);
threadA.start();
}
}
}
class ThreadA extends Thread {
private SkillService skillService;
public ThreadA(SkillService skillService) {
this.skillService = skillService;
}
@Override
public void run() {
skillService.seckill();
}
}
2.4.测试结果
若注释掉使用锁的部分：
public void seckill() {
// 返回锁的value值，供释放锁时候进行判断
//String identifier = lock.lockWithTimeout("resource", 5000, 1000);
System.out.println(Thread.currentThread().getName() + "获得了锁");
System.out.println(--n);
//lock.releaseLock("resource", identifier);
}
从结果可以看出，有一些是异步进行的：
为什么需要分布式锁
1.为什么需要分布式锁
在Java中，我们学习过多线程，也会学到锁。其中，锁解决的是多线程环境下，对共享资源的操作的问题。
而在分布式中，会变成一个多进程多线程的更复杂的环境。那么，这时候，我们也需要一个锁的机制，来解决共享资源的问题。独立进程中，锁可以放到进程里。多进程环境下，锁就必须放在一个通用的第三方存储空间中。这就是分布式锁。常见的场景如 商品库存，MQ消息去重等，都可以通过分布式锁来解决。核心就是两个目的：
-
对共享资源的操作的互斥。
-
对共享资源的操作的串行化。
假设有两个进程，要对同一个资源进行操作，在分布式锁的帮助下，它们可能是下面这样的。
对于任意一个进程来说，在操作资源之前，必须先获取分布式锁。获取成功，则可以进行操作，获取失败，则需要等待或返回失败。
2.Redis分布式锁(答案)
基于Redis的分布式锁，是常见的分布式锁实现方式之一。它通过以下一些机制来解决分布式锁的相关问题。
-
通过setnx命令的排他性来达成锁的互斥。setnx命令的特性是，在插入一个key、value对时，当key不存在时，可以设置成功，返回1，当key已经存在时，会设置失败，返回0。
-
通过锁超时来达成避免死锁的问题。当在Redis中插入key成功后，即认为获取锁成功，当锁使用完成后，需要删除相应key，即释放锁。当删除失败时，就会出现死锁问题。这时候，可以通过在set时，设置超时时间来解决。如设置超时时间为10s等。
-
通过独立线程来完成锁续租。由于加入了锁超时机制，所以，有可能会发生任务还未完成，锁就被释放的问题。一些分布式锁的开源框架，会通过一个单独线程，来不断通过exprie命令重刷key的过期时间，来达成一个锁续租的目的。
-
通过value来配合解决锁重入问题。在分布式锁下，value可以是当前进程或线程的一个唯一id，当获取锁时，如果发现锁已经存在，可以根据value值来判断，是否可以获取锁，来达成一个锁重入的目的。
分布式锁
Redission
1.Redisson简介
Redis 是最流行的 NoSQL 数据库解决方案之一，而 Java 是世界上最流行（注意，我没有说“最好”）的编程语言之一。虽然两者看起来很自然地在一起“工作”，但是要知道，Redis 其实并没有对 Java 提供原生支持。
相反，作为 Java 开发人员，我们若想在程序中集成 Redis，必须使用 Redis 的第三方库。而 Redisson 就是用于在 Java 程序中操作 Redis 的库，它使得我们可以在程序中轻松地使用 Redis。Redisson 在 java.util 中常用接口的基础上，为我们提供了一系列具有分布式特性的工具类。
Redisson底层采用的是Netty 框架。支持Redis 2.8以上版本，支持Java1.6+以上版本。
Redis客户端API工具：
Jedis、SpringData Redis、Redisson(布隆过滤器、分布式锁)
2.Redisson实现分布式锁的步骤
2.1.引入依赖
引入重要的两个依赖，一个是spring-boot-starter-data-redis，一个是redisson：
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-data-redis</artifactId>
</dependency>
<dependency>
<groupId>org.redisson</groupId>
<artifactId>redisson</artifactId>
<version>3.7.5</version>
</dependency>
2.2.application.properties
# Redis服务器地址（默认session使用）
spring.redis.host=127.0.0.1
# Redis服务器连接密码（默认为空）
# spring.redis.password=
# Redis服务器连接端口
spring.redis.port=6379
2.3.定义Loker
接口编程的思想还是要保持的。我们定义一个Loker接口，用于分布式锁的一些操作：
/**
* @CreateTime: 2020-05-08 10:20
* @Description: 锁接口
*/
import java.util.concurrent.TimeUnit;
public interface Locker {
/**
* 获取锁，如果锁不可用，则当前线程处于休眠状态，直到获得锁为止。
*
* @param lockKey
*/
void lock(String lockKey);
/**
* 释放锁
*
* @param lockKey
*/
void unlock(String lockKey);
/**
* 获取锁,如果锁不可用，则当前线程处于休眠状态，直到获得锁为止。如果获取到锁后，执行结束后解锁或达到超时时间后会自动释放锁
*
* @param lockKey
* @param timeout
*/
void lock(String lockKey, int timeout);
/**
* 获取锁,如果锁不可用，则当前线程处于休眠状态，直到获得锁为止。如果获取到锁后，执行结束后解锁或达到超时时间后会自动释放锁
*
* @param lockKey
* @param unit
* @param timeout
*/
void lock(String lockKey, TimeUnit unit, int timeout);
/**
* 尝试获取锁，获取到立即返回true,未获取到立即返回false
*
* @param lockKey
* @return
*/
boolean tryLock(String lockKey);
/**
* 尝试获取锁，在等待时间内获取到锁则返回true,否则返回false,如果获取到锁，则要么执行完后程序释放锁，
* 要么在给定的超时时间leaseTime后释放锁
*
* @param lockKey
* @param waitTime
* @param leaseTime
* @param unit
* @return
*/
boolean tryLock(String lockKey, long waitTime, long leaseTime, TimeUnit unit)
throws InterruptedException;
/**
* 锁是否被任意一个线程锁持有
*
* @param lockKey
* @return
*/
boolean isLocked(String lockKey);
}
有了Locker接口，我们再添加一个基于Redisson的实现类RedissonLocker，实现Locker中的方法：
/**
* @CreateTime: 2020-05-08 10:20
* @Description: 基于Redisson的分布式锁
*/
import java.util.concurrent.TimeUnit;
import org.redisson.api.RLock;
import org.redisson.api.RedissonClient;
public class RedissonLocker implements Locker {
private RedissonClient redissonClient;
public RedissonLocker(RedissonClient redissonClient) {
super();
this.redissonClient = redissonClient;
}
@Override
public void lock(String lockKey) {
RLock lock = redissonClient.getLock(lockKey);
lock.lock();
}
@Override
public void unlock(String lockKey) {
RLock lock = redissonClient.getLock(lockKey);
lock.unlock();
}
@Override
public void lock(String lockKey, int leaseTime) {
RLock lock = redissonClient.getLock(lockKey);
lock.lock(leaseTime, TimeUnit.SECONDS);
}
@Override
public void lock(String lockKey, TimeUnit unit, int timeout) {
RLock lock = redissonClient.getLock(lockKey);
lock.lock(timeout, unit);
}
public void setRedissonClient(RedissonClient redissonClient) {
this.redissonClient = redissonClient;
}
@Override
public boolean tryLock(String lockKey) {
RLock lock = redissonClient.getLock(lockKey);
return lock.tryLock();
}
@Override
public boolean tryLock(String lockKey, long waitTime, long leaseTime,
TimeUnit unit) throws InterruptedException{
RLock lock = redissonClient.getLock(lockKey);
return lock.tryLock(waitTime, leaseTime, unit);
}
@Override
public boolean isLocked(String lockKey) {
RLock lock = redissonClient.getLock(lockKey);
return lock.isLocked();
}
}
2.4.定义工具类
有了Locker和实现类RedissonLocker，我们总不能一直去创建RedissonLocker对象或者不断的在每个要使用到分布式锁的地方都注入RedissonLocker的对象，所以我们定义一个工具类LockUtil，到时候想哪里使用就直接使用工具类的静态方法就行了：
/**
* @CreateTime: 2020-05-08 10:21
* @Description: TODO
*/
import com.bruceliu.lock.Locker;
import java.util.concurrent.TimeUnit;
/**
* redis分布式锁工具类
*
*/
public final class LockUtil {
private static Locker locker;
/**
* 设置工具类使用的locker
* @param locker
*/
public static void setLocker(Locker locker) {
LockUtil.locker = locker;
}
/**
* 获取锁
* @param lockKey
*/
public static void lock(String lockKey) {
locker.lock(lockKey);
}
/**
* 释放锁
* @param lockKey
*/
public static void unlock(String lockKey) {
locker.unlock(lockKey);
}
/**
* 获取锁，超时释放
* @param lockKey
* @param timeout
*/
public static void lock(String lockKey, int timeout) {
locker.lock(lockKey, timeout);
}
/**
* 获取锁，超时释放，指定时间单位
* @param lockKey
* @param unit
* @param timeout
*/
public static void lock(String lockKey, TimeUnit unit, int timeout) {
locker.lock(lockKey, unit, timeout);
}
/**
* 尝试获取锁，获取到立即返回true,获取失败立即返回false
* @param lockKey
* @return
*/
public static boolean tryLock(String lockKey) {
return locker.tryLock(lockKey);
}
/**
* 尝试获取锁，在给定的waitTime时间内尝试，获取到返回true,获取失败返回false,获取到后再给定的leaseTime时间超时释放
* @param lockKey
* @param waitTime
* @param leaseTime
* @param unit
* @return
* @throws InterruptedException
*/
public static boolean tryLock(String lockKey, long waitTime, long leaseTime,
TimeUnit unit) throws InterruptedException {
return locker.tryLock(lockKey, waitTime, leaseTime, unit);
}
/**
* 锁释放被任意一个线程持有
* @param lockKey
* @return
*/
public static boolean isLocked(String lockKey) {
return locker.isLocked(lockKey);
}
}
2.5.Redisson的配置类
现在我们开始配置吧，创建一个redisson的配置类RedissonConfig，内容如下：
import java.io.IOException;
import com.bruceliu.lock.RedissonLocker;
import com.bruceliu.utils.LockUtil;
import org.redisson.Redisson;
import org.redisson.api.RedissonClient;
import org.redisson.config.Config;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
/**
* @CreateTime: 2020-05-08 10:22
* @Description: TODO
*/
@Configuration
public class RedissonConfig {
@Value("${spring.redis.host}")
private String host;
@Value("${spring.redis.port}")
private String port;
//@Value("${spring.redis.password}")
//private String password;
/**
* RedissonClient,单机模式
* @return
* @throws IOException
*/
@Bean(destroyMethod = "shutdown")
public RedissonClient redisson() throws IOException {
Config config = new Config();
//config.useSingleServer().setAddress("redis://" + host + ":" + port).setPassword(password);
config.useSingleServer().setAddress("redis://" + host + ":" + port);
return Redisson.create(config);
}
@Bean
public RedissonLocker redissonLocker(RedissonClient redissonClient){
RedissonLocker locker = new RedissonLocker(redissonClient);
//设置LockUtil的锁处理对象
LockUtil.setLocker(locker);
return locker;
}
}
2.6.Redisson分布式锁业务类
import com.bruceliu.utils.LockUtil;
import java.util.concurrent.TimeUnit;
/**
* @CreateTime: 2020-05-08 10:26
* @Description: TODO
*/
public class SkillService {
int n = 500;
public void seckill() {
//加锁
LockUtil.lock("resource", TimeUnit.SECONDS,5000);
try {
System.out.println(Thread.currentThread().getName() + "获得了锁");
Thread.sleep(3000);
System.out.println(--n);
} catch (Exception e) {
//异常处理
}finally{
//释放锁
LockUtil.unlock("resource");
System.out.println(Thread.currentThread().getName() + "释放了锁");
}
}
}
2.7.Redisson分布式锁测试
import com.bruceliu.App;
import com.bruceliu.service.SkillService;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.test.context.junit4.SpringRunner;
/**
* @CreateTime: 2020-05-08 10:29
* @Description: TODO
*/
@RunWith(SpringRunner.class)
@SpringBootTest(classes = App.class)
public class TestLock {
@Test
public void testLock() throws Exception{
SkillService service = new SkillService();
for (int i = 0; i < 10; i++) {
ThreadA threadA = new ThreadA(service);
threadA.setName("ThreadNameA->"+i);
threadA.start();
}
Thread.sleep(50000);
}
}
class ThreadA extends Thread {
private SkillService skillService;
public ThreadA(SkillService skillService) {
this.skillService = skillService;
}
@Override
public void run() {
skillService.seckill();
}
}
运行结果:
注释文中加锁代码：
public void seckill() {
//加锁
//LockUtil.lock("resource", TimeUnit.SECONDS,5000);
try {
System.out.println(Thread.currentThread().getName() + "获得了锁");
Thread.sleep(3000);
System.out.println(--n);
} catch (Exception e) {
//异常处理
}finally{
//释放锁
//LockUtil.unlock("resource");
System.out.println(Thread.currentThread().getName() + "释放了锁");
}
}
运行结果：
可以看到存在并发的问题！
Redis分布式锁的细节
在分布式系统中，由于redis分布式锁相对于更简单和高效，成为了分布式锁的首先，被我们用到了很多实际业务场景当中。
但不是说用了redis分布式锁，就可以高枕无忧了，如果没有用好或者用对，也会引来一些意想不到的问题。
1.非原子操作
使用redis的分布式锁，我们首先想到的可能是setNx命令。
if (jedis.setnx(lockKey, val) == 1) {
jedis.expire(lockKey, timeout);
}
容易，三下五除二，我们就可以把代码写好。这段代码确实可以加锁成功，但你有没有发现什么问题？
加锁操作和后面的设置超时时间是分开的，并非原子操作。
假如加锁成功，但是设置超时时间失败了，该lockKey就变成永不失效。假如在高并发场景中，有大量的lockKey加锁成功了，但不会失效，有可能直接导致redis内存空间不足。
那么，有没有保证原子性的加锁命令呢？
答案是：有，请看下面。
2.忘了释放锁
上面说到使用setNx命令加锁操作和设置超时时间是分开的，并非原子操作。
而在redis中还有set命令，该命令可以指定多个参数。
String result = jedis.set(lockKey, requestId, "NX", "PX", expireTime);
if ("OK".equals(result)) {
return true;
}
return false;
其中：
- lockKey：锁的标识
- requestId：请求id
- NX：只在键不存在时，才对键进行设置操作。
- PX：设置键的过期时间为 millisecond 毫秒。
- expireTime：过期时间
set命令是原子操作，加锁和设置超时时间，一个命令就能轻松搞
定。
使用set命令加锁，表面上看起来没有问题。但如果仔细想想，加锁之后，每次都要达到了超时时间才释放锁，会不会有点不合理？加锁后，如果不及时释放锁，会有很多问题。
分布式锁更合理的用法是：
- 手动加锁
- 业务操作
- 手动释放锁
如果手动释放锁失败了，则达到超时时间，redis会自动释放锁。
大致流程图如下：
那么问题来了，如何释放锁呢？
伪代码如下：
try{
String result = jedis.set(lockKey, requestId, "NX", "PX", expireTime);
if ("OK".equals(result)) {
return true;
}
return false;
} finally {
unlock(lockKey);
}
需要捕获业务代码的异常，然后在finally中释放锁。换句话说就是：无论代码执行成功或失败了，都需要释放锁。
此时，有些朋友可能会问：假如刚好在释放锁的时候，系统被重启了，或者网络断线了，或者机房断点了，不也会导致释放锁失败？
这是一个好问题，因为这种小概率问题确实存在。
但还记得前面我们给锁设置过超时时间吗？即使出现异常情况造成释放锁失败，但到了我们设定的超时时间，锁还是会被redis自动释放。
但只在finally中释放锁，就够了吗？
3.释放了别人的锁
做人要厚道，先回答上面的问题：只在finally中释放锁，当然是不够的，因为释放锁的姿势，还是不对。
哪里不对？
答：在多线程场景中，可能会出现释放了别人的锁的情况。
有些朋友可能会反驳：假设在多线程场景中，线程A获取到了锁，但如果线程A没有释放锁，此时，线程B是获取不到锁的，何来释放了别人锁之说？
答：假如线程A和线程B，都使用lockKey加锁。线程A加锁成功了，但是由于业务功能耗时时间很长，超过了设置的超时时间。这时候，redis会自动释放lockKey锁。此时，线程B就能给lockKey加锁成功了，接下来执行它的业务操作。恰好这个时候，线程A执行完了业务功能，接下来，在finally方法中释放了锁lockKey。这不就出问题了，线程B的锁，被线程A释放了。
我想这个时候，线程B肯定哭晕在厕所里，并且嘴里还振振有词。
那么，如何解决这个问题呢？
不知道你们注意到没？在使用set命令加锁时，除了使用lockKey锁标识，还多设置了一个参数：requestId，为什么要需要记录requestId呢？
答：requestId是在释放锁的时候用的。
伪代码如下:
if (jedis.get(lockKey).equals(requestId)) {
jedis.del(lockKey);
return true;
}
return false;
在释放锁的时候，先获取到该锁的值（之前设置值就是requestId），然后判断跟之前设置的值是否相同，如果相同才允许删除锁，返回成功。如果不同，则直接返回失败。
换句话说就是：自己只能释放自己加的锁，不允许释放别人加的锁。
这里为什么要用requestId，用userId不行吗？
答：如果用userId的话，对于请求来说并不唯一，多个不同的请求，可能使用同一个userId。而requestId是全局唯一的，不存在加锁和释放锁乱掉的情况。
此外，使用lua脚本，也能解决释放了别人的锁的问题：
if (redis.call('exists', KEYS[1]) == 0) then
redis.call('hset', KEYS[1], ARGV[2], 1);
redis.call('pexpire', KEYS[1], ARGV[1]);
return nil;
end
if (redis.call('hexists', KEYS[1], ARGV[2]) == 1)
redis.call('hincrby', KEYS[1], ARGV[2], 1);
redis.call('pexpire', KEYS[1], ARGV[1]);
return nil;
end;
return redis.call('pttl', KEYS[1]);
这是redisson框架的加锁代码，写的不错，大家可以借鉴一下。
有趣，下面还有哪些好玩的东西？
4.大量失败请求
上面的加锁方法看起来好像没有问题，但如果你仔细想想，如果有1万的请求同时去竞争那把锁，可能只有一个请求是成功的，其余的9999个请求都会失败。
在秒杀场景下，会有什么问题？
答：每1万个请求，有1个成功。再1万个请求，有1个成功。如此下去，直到库存不足。这就变成均匀分布的秒杀了，跟我们想象中的不一样。
如何解决这个问题呢？
此外，还有一种场景：
比如，有两个线程同时上传文件到sftp，上传文件前先要创建目录。假设两个线程需要创建的目录名都是当天的日期，比如：20210920，如果不做任何控制，直接并发的创建目录，第二个线程必然会失败。
这时候有些朋友可能会说：这还不容易，加一个redis分布式锁就能解决问题了，此外再判断一下，如果目录已经存在就不创建，只有目录不存在才需要创建。
伪代码如下：
try {
String result = jedis.set(lockKey, requestId, "NX", "PX", expireTime);
if ("OK".equals(result)) {
if(!exists(path)) {
mkdir(path);
}
return true;
}
} finally{
unlock(lockKey,requestId);
}
return false;
一切看似美好，但经不起仔细推敲。
来自灵魂的一问：第二个请求如果加锁失败了，接下来，是返回失败，还是返回成功呢？
主要流程图如下：
显然第二个请求，肯定是不能返回失败的，如果返回失败了，这个问题还是没有被解决。如果文件还没有上传成功，直接返回成功会有更大的问题。头疼，到底该如何解决呢？
答：使用自旋锁。
try {
Long start = System.currentTimeMillis();
while(true) {
String result = jedis.set(lockKey, requestId, "NX", "PX", expireTime);
if ("OK".equals(result)) {
if(!exists(path)) {
mkdir(path);
}
return true;
}
long time = System.currentTimeMillis() - start;
if (time>=timeout) {
return false;
}
try {
Thread.sleep(50);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
} finally{
unlock(lockKey,requestId);
}
return false;
在规定的时间，比如500毫秒内，自旋不断尝试加锁（说白了，就是在死循环中，不断尝试加锁），如果成功则直接返回。如果失败，则休眠50毫秒，再发起新一轮的尝试。如果到了超时时间，还未加锁成功，则直接返回失败。
好吧，学到一招了，还有吗？
5. 锁重入问题
我们都知道redis分布式锁是互斥的。假如我们对某个key加锁了，如果该key对应的锁还没失效，再用相同key去加锁，大概率会失败。
没错，大部分场景是没问题的。
为什么说是大部分场景呢？
因为还有这样的场景：
假设在某个请求中，需要获取一颗满足条件的菜单树或者分类树。我们以菜单为例，这就需要在接口中从根节点开始，递归遍历出所有满足条件的子节点，然后组装成一颗菜单树。
需要注意的是菜单不是一成不变的，在后台系统中运营同学可以动态添加、修改和删除菜单。为了保证在并发的情况下，每次都可能获取最新的数据，这里可以加redis分布式锁。
加redis分布式锁的思路是对的。但接下来问题来了，在递归方法中递归遍历多次，每次都是加的同一把锁。递归第一层当然是可以加锁成功的，但递归第二层、第三层…第N层，不就会加锁失败了？
递归方法中加锁的伪代码如下：
private int expireTime = 1000;
public void fun(int level,String lockKey,String requestId){
try{
String result = jedis.set(lockKey, requestId, "NX", "PX", expireTime);
if ("OK".equals(result)) {
if(level<=10){
this.fun(++level,lockKey,requestId);
} else {
return;
}
}
return;
} finally {
unlock(lockKey,requestId);
}
}
如果你直接这么用，看起来好像没有问题。但最终执行程序之后发现，等待你的结果只有一个：出现异常。
因为从根节点开始，第一层递归加锁成功，还没释放锁，就直接进入第二层递归。因为锁名为lockKey，并且值为requestId的锁已经存在，所以第二层递归大概率会加锁失败，然后返回到第一层。第一层接下来正常释放锁，然后整个递归方法直接返回了。
这下子，大家知道出现什么问题了吧？
没错，递归方法其实只执行了第一层递归就返回了，其他层递归由于加锁失败，根本没法执行。
那么这个问题该如何解决呢？
答：使用可重入锁。
我们以redisson框架为例，它的内部实现了可重入锁的功能。
古时候有句话说得好：为人不识陈近南，便称英雄也枉然。
我说：分布式锁不识redisson，便称好锁也枉然。哈哈哈，只是自娱自乐一下。
由此可见，redisson在redis分布式锁中的江湖地位很高。
伪代码如下：
private int expireTime = 1000;
public void run(String lockKey) {
RLock lock = redisson.getLock(lockKey);
this.fun(lock,1);
}
public void fun(RLock lock,int level){
try{
lock.lock(5, TimeUnit.SECONDS);
if(level<=10){
this.fun(lock,++level);
} else {
return;
}
} finally {
lock.unlock();
}
}
上面的代码也许并不完美，这里只是给了一个大致的思路，如果大家有这方面需求的话，以上代码仅供参考。
接下来，聊聊redisson可重入锁的实现原理。
加锁主要是通过以下脚本实现的：
if (redis.call('exists', KEYS[1]) == 0)
then
redis.call('hset', KEYS[1], ARGV[2], 1); redis.call('pexpire', KEYS[1], ARGV[1]);
return nil;
end;
if (redis.call('hexists', KEYS[1], ARGV[2]) == 1)
then
redis.call('hincrby', KEYS[1], ARGV[2], 1);
redis.call('pexpire', KEYS[1], ARGV[1]);
return nil;
end;
return redis.call('pttl', KEYS[1]);
其中：
- KEYS[1]：锁名
- ARGV[1]：过期时间
- ARGV[2]：uuid + “:” + threadId，可认为是requestId
- 先判断如果锁名不存在，则加锁。
- 接下来，判断如果锁名和requestId值都存在，则使用hincrby命令给该锁名和requestId值计数，每次都加1。注意一下，这里就是重入锁的关键，锁重入一次值就加1。
- 如果锁名存在，但值不是requestId，则返回过期时间。
释放锁主要是通过以下脚本实现的：
if (redis.call('hexists', KEYS[1], ARGV[3]) == 0)
then
return nil
end
local counter = redis.call('hincrby', KEYS[1], ARGV[3], -1);
if (counter > 0)
then
redis.call('pexpire', KEYS[1], ARGV[2]);
return 0;
else
redis.call('del', KEYS[1]);
redis.call('publish', KEYS[2], ARGV[1]);
return 1;
end;
return nil
- 先判断如果锁名和requestId值不存在，则直接返回。
- 如果锁名和requestId值存在，则重入锁减1。
- 如果减1后，重入锁的value值还大于0，说明还有引用，则重试设置过期时间。
- 如果减1后，重入锁的value值还等于0，则可以删除锁，然后发消息通知等待线程抢锁。
再次强调一下，如果你们系统可以容忍数据暂时不一致，有些场景不加锁也行，我在这里只是举个例子，本节内容并不适用于所有场景。
6锁竞争问题
如果有大量需要写入数据的业务场景，使用普通的redis分布式锁是没有问题的。
但如果有些业务场景，写入的操作比较少，反而有大量读取的操作。这样直接使用普通的redis分布式锁，会不会有点浪费性能？
我们都知道，锁的粒度越粗，多个线程抢锁时竞争就越激烈，造成多个线程锁等待的时间也就越长，性能也就越差。
所以，提升redis分布式锁性能的第一步，就是要把锁的粒度变细。
6.1.读写锁
众所周知，加锁的目的是为了保证，在并发环境中读写数据的安全性，即不会出现数据错误或者不一致的情况。
但在绝大多数实际业务场景中，一般是读数据的频率远远大于写数据。而线程间的并发读操作是并不涉及并发安全问题，我们没有必要给读操作加互斥锁，只要保证读写、写写并发操作上锁是互斥的就行，这样可以提升系统的性能。
我们以redisson框架为例，它内部已经实现了读写锁的功能。
读锁的伪代码如下：
RReadWriteLock readWriteLock = redisson.getReadWriteLock("readWriteLock");
RLock rLock = readWriteLock.readLock();
try {
rLock.lock();
//业务操作
} catch (Exception e) {
log.error(e);
} finally {
rLock.unlock();
}
写锁的伪代码如下：
RReadWriteLock readWriteLock = redisson.getReadWriteLock("readWriteLock");
RLock rLock = readWriteLock.writeLock();
try {
rLock.lock();
//业务操作
} catch (InterruptedException e) {
log.error(e);
} finally {
rLock.unlock();
}
将读锁和写锁分开，最大的好处是提升读操作的性能，因为读和读之间是共享的，不存在互斥性。而我们的实际业务场景中，绝大多数数据操作都是读操作。所以，如果提升了读操作的性能，也就会提升整个锁的性能。
下面总结一个读写锁的特点：
- 读与读是共享的，不互斥
- 读与写互斥
- 写与写互斥
6.2.锁分段
此外，为了减小锁的粒度，比较常见的做法是将大锁：分段。
在java中ConcurrentHashMap，就是将数据分为16段，每一段都有单独的锁，并且处于不同锁段的数据互不干扰，以此来提升锁的性能。
放在实际业务场景中，我们可以这样做：
比如在秒杀扣库存的场景中，现在的库存中有2000个商品，用户可以秒杀。为了防止出现超卖的情况，通常情况下，可以对库存加锁。如果有1W的用户竞争同一把锁，显然系统吞吐量会非常低。
为了提升系统性能，我们可以将库存分段，比如：分为100段，这样每段就有20个商品可以参与秒杀。
在秒杀的过程中，先把用户id获取hash值，然后除以100取模。模为1的用户访问第1段库存，模为2的用户访问第2段库存，模为3的用户访问第3段库存，后面以此类推，到最后模为100的用户访问第100段库存。
如此一来，在多线程环境中，可以大大的减少锁的冲突。以前多个线程只能同时竞争1把锁，尤其在秒杀的场景中，竞争太激烈了，简直可以用惨绝人寰来形容，其后果是导致绝大数线程在锁等待。现在多个线程同时竞争100把锁，等待的线程变少了，从而系统吞吐量也就提升了。
需要注意的地方是：将锁分段虽说可以提升系统的性能，但它也会让系统的复杂度提升不少。因为它需要引入额外的路由算法，跨段统计等功能。我们在实际业务场景中，需要综合考虑，不是说一定要将锁分段。
7.锁超时问题
我在前面提到过，如果线程A加锁成功了，但是由于业务功能耗时时间很长，超过了设置的超时时间，这时候redis会自动释放线程A加的锁。
有些朋友可能会说：到了超时时间，锁被释放了就释放了呗，对功能又没啥影响。
答：错，错，错。对功能其实有影响。
通常我们加锁的目的是：为了防止访问临界资源时，出现数据异常的情况。比如：线程A在修改数据C的值，线程B也在修改数据C的值，如果不做控制，在并发情况下，数据C的值会出问题。
为了保证某个方法，或者段代码的互斥性，即如果线程A执行了某段代码，是不允许其他线程在某一时刻同时执行的，我们可以用synchronized关键字加锁。
但这种锁有很大的局限性，只能保证单个节点的互斥性。如果需要在多个节点中保持互斥性，就需要用redis分布式锁。
做了这么多铺垫，现在回到正题。
假设线程A加redis分布式锁的代码，包含代码1和代码2两段代码。
由于该线程要执行的业务操作非常耗时，程序在执行完代码1的时，已经到了设置的超时时间，redis自动释放了锁。而代码2还没来得及执行。
此时，代码2相当于裸奔的状态，无法保证互斥性。假如它里面访问了临界资源，并且其他线程也访问了该资源，可能就会出现数据异常的情况。（PS：我说的访问临界资源，不单单指读取，还包含写入）
那么，如何解决这个问题呢？
答：如果达到了超时时间，但业务代码还没执行完，需要给锁自动续期。
我们可以使用TimerTask类，来实现自动续期的功能：
Timer timer = new Timer();
timer.schedule(new TimerTask() {
@Override
public void run(Timeout timeout) throws Exception {
//自动续期逻辑
}
}, 10000, TimeUnit.MILLISECONDS);
获取锁之后，自动开启一个定时任务，每隔10秒钟，自动刷新一次过期时间。这种机制在redisson框架中，有个比较霸气的名字：watch dog，即传说中的看门狗。
当然自动续期功能，我们还是优先推荐使用lua脚本实现，比如：
if (redis.call('hexists', KEYS[1], ARGV[2]) == 1) then
redis.call('pexpire', KEYS[1], ARGV[1]);
return 1;
end;
return 0;
需要注意的地方是：在实现自动续期功能时，还需要设置一个总的过期时间，可以跟redisson保持一致，设置成30秒。如果业务代码到了这个总的过期时间，还没有执行完，就不再自动续期了。
自动续期的功能是获取锁之后开启一个定时任务，每隔10秒判断一下锁是否存在，如果存在，则刷新过期时间。如果续期3次，也就是30秒之后，业务方法还是没有执行完，就不再续期了。
8.主从复制的问题
上面花了这么多篇幅介绍的内容，对单个redis实例是没有问题的。
but，如果redis存在多个实例。比如：做了主从，或者使用了哨兵模式，基于redis的分布式锁的功能，就会出现问题。
具体是什么问题？
假设redis现在用的主从模式，1个master节点，3个slave节点。master节点负责写数据，slave节点负责读数据。
本来是和谐共处，相安无事的。redis加锁操作，都在master上进行，加锁成功后，再异步同步给所有的slave。
突然有一天，master节点由于某些不可逆的原因，挂掉了。
这样需要找一个slave升级为新的master节点，假如slave1被选举出来了。
如果有个锁A比较悲催，刚加锁成功master就挂了，还没来得及同步到slave1。
这样会导致新master节点中的锁A丢失了。后面，如果有新的线程，使用锁A加锁，依然可以成功，分布式锁失效了。
那么，如何解决这个问题呢？
答：redisson框架为了解决这个问题，提供了一个专门的类：RedissonRedLock，使用了Redlock算法。
RedissonRedLock解决问题的思路如下：
- 需要搭建几套相互独立的redis环境，假如我们在这里搭建了5套。
- 每套环境都有一个redisson node节点。
- 多个redisson node节点组成了RedissonRedLock。
- 环境包含：单机、主从、哨兵和集群模式，可以是一种或者多种混合。
在这里我们以主从为例，架构图如下：
RedissonRedLock加锁过程如下：
- 获取所有的redisson node节点信息，循环向所有的redisson 节点加锁，假设节点数为N，例子中N等于5。
- 如果在N个节点当中，有N/2 + 1个节点加锁成功了，那么整个RedissonRedLock加锁是成功的。
- 如果在N个节点当中，小于N/2 + 1个节点加锁成功，那么整个RedissonRedLock加锁是失败的。
- 如果中途发现各个节点加锁的总耗时，大于等于设置的最大等待时间，则直接返回失败。
从上面可以看出，使用Redlock算法，确实能解决多实例场景中，假如master节点挂了，导致分布式锁失效的问题。
也引出了一些新问题，比如：
- 需要额外搭建多套环境，申请更多的资源，需要评估一下成本和性价比。
- 如果有N个redisson node节点，需要加锁N次，最少也需要加锁N/2+1次，才知道redlock加锁是否成功。显然，增加了额外的时间成本，有点得不偿失。
由此可见，在实际业务场景，尤其是高并发业务中，RedissonRedLock其实使用的并不多。
在分布式环境中，CAP是绕不过去的。
CAP指的是在一个分布式系统中：
一致性（Consistency）
可用性（Availability）
分区容错性（Partition tolerance）
这三个要素最多只能同时实现两点，不可能三者兼顾。
如果你的实际业务场景，更需要的是保证数据一致性。那么请使用CP类型的分布式锁，比如：zookeeper，它是基于磁盘的，性能可能没那么好，但数据一般不会丢。
如果你的实际业务场景，更需要的是保证数据高可用性。那么请使用AP类型的分布式锁，比如：redis，它是基于内存的，性能比较好，但有丢失数据的风险。
其实，在我们绝大多数分布式业务场景中，使用redis分布式锁就够了，真的别太较真。因为数据不一致问题，可以通过最终一致性方案解决。但如果系统不可用了，对用户来说是暴击一万点伤害。
分布式唯一ID生成方案
系统唯一ID是我们在设计一个系统的时候常常会遇见的问题，也常常为这个问题而纠结。生成ID的方法有很多，适应不同的场景、需求以及性能要求。所以有些比较复杂的系统会有多个ID生成的策略。下面就介绍一些常见的ID生成策略。
1.数据库自增长序列或字段
最常见的方式。利用数据库，全数据库唯一。
关系型数据库都实现数据库自增ID;mysql通过auto_increment实现、oralce通过sequence实现。
在数据库集群环境下，不同数据库节点可设置不同起步值、相同步长值来实现集群下生成全局唯一、递增ID
优点：
1）简单，代码方便，性能可以接受。
2）数字ID天然排序，对分页或者需要排序的结果很有帮助。
缺点：
1）不同数据库语法和实现不同，数据库迁移的时候或多数据库版本支持的时候需要处理。
2）在单个数据库或读写分离或一主多从的情况下，只有一个主库可以生成。有单点故障的风险。
3）在性能达不到要求的情况下，比较难于扩展。
4）如果遇见多个系统需要合并或者涉及到数据迁移会相当痛苦。
5）分表分库的时候会有麻烦。
2.UUID
常见的方式。可以利用数据库也可以利用程序生成，一般来说全球唯一。
140a5382-c69a-4c14-84e8-07c95a9bae2b
优点：
1）简单，代码方便。
2）生成ID性能非常好，基本不会有性能问题。
3）全球唯一，在遇见数据迁移，系统数据合并，或者数据库变更等情况下，可以从容应对。
缺点：
1）没有排序，无法保证趋势递增。
2）UUID往往是使用字符串存储，查询的效率比较低。
3）存储空间比较大，如果是海量数据库，就需要考虑存储量的问题。
4）传输数据量大
5）不可读。不方便操作！
3.Redis生成ID
当使用数据库来生成ID性能不够要求的时候，我们可以尝试使用Redis来生成ID。这主要依赖于Redis是单线程的，所以也可以用生成全局唯一的ID。可以用Redis的原子操作 INCR和INCRBY来实现。
可以使用Redis集群来获取更高的吞吐量。假如一个集群中有5台Redis。可以初始化每台Redis的值分别是1,2,3,4,5，然后步长都是5。各个Redis生成的ID为：
A：1,6,11,16,21
B：2,7,12,17,22
C：3,8,13,18,23
D：4,9,14,19,24
E：5,10,15,20,25
这个，随便负载到哪个机确定好，未来很难做修改。但是3-5台服务器基本能够满足器上，都可以获得不同的ID。但是步长和初始值一定需要事先需要了。使用Redis集群也可以方式单点故障的问题。
另外，比较适合使用Redis来生成每天从0开始的流水号。比如订单号=日期+当日自增长号。可以每天在Redis中生成一个Key，使用INCR进行累加。
优点：
1）不依赖于数据库，灵活方便，且性能优于数据库。
2）数字ID天然排序，对分页或者需要排序的结果很有帮助。
缺点：
1）如果系统中没有Redis，还需要引入新的组件，增加系统复杂度。
2）需要编码和配置的工作量比较大。
4.Twitter的snowflake算法
snowflake是Twitter开源的分布式ID生成算法，结果是一个long型的ID。其核心思想是：使用41bit作为毫秒数，10bit作为机器的ID（5个bit是数据中心，5个bit的机器ID），12bit作为毫秒内的流水号（意味着每个节点在每毫秒可以产生 4096 个 ID），最后还有一个符号位，永远是0。具体实现的代码可以参看https://github.com/twitter/snowflake。
import java.lang.management.ManagementFactory;
import java.net.InetAddress;
import java.net.NetworkInterface;
/**
* <p>名称：IdWorker.java</p>
* <p>描述：分布式自增长ID</p>
* <pre>
* Twitter的 Snowflake JAVA实现方案
* </pre>
* 核心代码为其IdWorker这个类实现，其原理结构如下，我分别用一个0表示一位，用—分割开部分的作用：
* 1||0---0000000000 0000000000 0000000000 0000000000 0 --- 00000 ---00000 ---000000000000
* 在上面的字符串中，第一位为未使用（实际上也可作为long的符号位），接下来的41位为毫秒级时间，
* 然后5位datacenter标识位，5位机器ID（并不算标识符，实际是为线程标识），
* 然后12位该毫秒内的当前毫秒内的计数，加起来刚好64位，为一个Long型。
* 这样的好处是，整体上按照时间自增排序，并且整个分布式系统内不会产生ID碰撞（由datacenter和机器ID作区分），
* 并且效率较高，经测试，snowflake每秒能够产生26万ID左右，完全满足需要。
* <p>
* 64位ID (42(毫秒)+5(机器ID)+5(业务编码)+12(重复累加))
*
* @author Polim
*/
public class IdWorker {
// 时间起始标记点，作为基准，一般取系统的最近时间（一旦确定不能变动）
private final static long twepoch = 1288834974657L;
// 机器标识位数
private final static long workerIdBits = 5L;
// 数据中心标识位数
private final static long datacenterIdBits = 5L;
// 机器ID最大值
private final static long maxWorkerId = -1L ^ (-1L << workerIdBits);
// 数据中心ID最大值
private final static long maxDatacenterId = -1L ^ (-1L << datacenterIdBits);
// 毫秒内自增位
private final static long sequenceBits = 12L;
// 机器ID偏左移12位
private final static long workerIdShift = sequenceBits;
// 数据中心ID左移17位
private final static long datacenterIdShift = sequenceBits + workerIdBits;
// 时间毫秒左移22位
private final static long timestampLeftShift = sequenceBits + workerIdBits + datacenterIdBits;
private final static long sequenceMask = -1L ^ (-1L << sequenceBits);
/* 上次生产id时间戳 */
private static long lastTimestamp = -1L;
// 0，并发控制
private long sequence = 0L;
private final long workerId;
// 数据标识id部分
private final long datacenterId;
public IdWorker(){
this.datacenterId = getDatacenterId(maxDatacenterId);
this.workerId = getMaxWorkerId(datacenterId, maxWorkerId);
}
/**
* @param workerId
* 工作机器ID
* @param datacenterId
* 序列号
*/
public IdWorker(long workerId, long datacenterId) {
if (workerId > maxWorkerId || workerId < 0) {
throw new IllegalArgumentException(String.format("worker Id can't be greater than %d or less than 0", maxWorkerId));
}
if (datacenterId > maxDatacenterId || datacenterId < 0) {
throw new IllegalArgumentException(String.format("datacenter Id can't be greater than %d or less than 0", maxDatacenterId));
}
this.workerId = workerId;
this.datacenterId = datacenterId;
}
/**
* 获取下一个ID
*
* @return
*/
public synchronized long nextId() {
long timestamp = timeGen();
if (timestamp < lastTimestamp) {
throw new RuntimeException(String.format("Clock moved backwards. Refusing to generate id for %d milliseconds", lastTimestamp - timestamp));
}
if (lastTimestamp == timestamp) {
// 当前毫秒内，则+1
sequence = (sequence + 1) & sequenceMask;
if (sequence == 0) {
// 当前毫秒内计数满了，则等待下一秒
timestamp = tilNextMillis(lastTimestamp);
}
} else {
sequence = 0L;
}
lastTimestamp = timestamp;
// ID偏移组合生成最终的ID，并返回ID
long nextId = ((timestamp - twepoch) << timestampLeftShift)
| (datacenterId << datacenterIdShift)
| (workerId << workerIdShift) | sequence;
return nextId;
}
private long tilNextMillis(final long lastTimestamp) {
long timestamp = this.timeGen();
while (timestamp <= lastTimestamp) {
timestamp = this.timeGen();
}
return timestamp;
}
private long timeGen() {
return System.currentTimeMillis();
}
/**
* <p>
* 获取 maxWorkerId
* </p>
*/
protected static long getMaxWorkerId(long datacenterId, long maxWorkerId) {
StringBuffer mpid = new StringBuffer();
mpid.append(datacenterId);
String name = ManagementFactory.getRuntimeMXBean().getName();
if (!name.isEmpty()) {
/*
* GET jvmPid
*/
mpid.append(name.split("@")[0]);
}
/*
* MAC + PID 的 hashcode 获取16个低位
*/
return (mpid.toString().hashCode() & 0xffff) % (maxWorkerId + 1);
}
/**
* <p>
* 数据标识id部分
* </p>
*/
protected static long getDatacenterId(long maxDatacenterId) {
long id = 0L;
try {
InetAddress ip = InetAddress.getLocalHost();
NetworkInterface network = NetworkInterface.getByInetAddress(ip);
if (network == null) {
id = 1L;
} else {
byte[] mac = network.getHardwareAddress();
id = ((0x000000FF & (long) mac[mac.length - 1])
| (0x0000FF00 & (((long) mac[mac.length - 2]) << 8))) >> 6;
id = id % (maxDatacenterId + 1);
}
} catch (Exception e) {
System.out.println(" getDatacenterId: " + e.getMessage());
}
return id;
}
public static void main(String[] args) {
for (int i = 0; i <1000; i++) {
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
IdWorker worker=new IdWorker(0, i);
long num = worker.nextId();
System.out.println(num);
}
}
}
snowflake算法可以根据自身项目的需要进行一定的修改。比如估算未来的数据中心个数，每个数据中心的机器数以及统一毫秒可以能的并发数来调整在算法中所需要的bit数。
优点：
1）不依赖于数据库，灵活方便，且性能优于数据库。
2）ID按照时间在单机上是递增的。
缺点：
1）在单机上是递增的，但是由于涉及到分布式环境，每台机器上的时钟不可能完全同步，也许有时候也会出现不是全局递增的情况。
5.利用zookeeper生成唯一ID
zookeeper主要通过其znode数据版本来生成序列号，可以生成32位和64位的数据版本号，客户端可以使用这个版本号来作为唯一的序列号。
很少会使用zookeeper来生成唯一ID。主要是由于需要依赖zookeeper，并且是多步调用API，如果在竞争较大的情况下，需要考虑使用分布式锁。因此，性能在高并发的分布式环境下，也不甚理想。
https://blog.csdn.net/wuliu_forever/article/details/53389483
6.MongoDB的ObjectId
MongoDB的ObjectId和snowflake算法类似。它设计成轻量型的，不同的机器都能用全局唯一的同种方法方便地生成它。MongoDB 从一开始就设计用来作为分布式数据库，处理多个节点是一个核心要求。使其在分片环境中要容易生成得多。
其格式如下：
前4 个字节是从标准纪元开始的时间戳，单位为秒。时间戳，与随后的5 个字节组合起来，提供了秒级别的唯一性。由于时间戳在前，这意味着ObjectId 大致会按照插入的顺序排列。这对于某些方面很有用，如将其作为索引提高效率。这4 个字节也隐含了文档创建的时间。绝大多数客户端类库都会公开一个方法从ObjectId 获取这个信息。
接下来的3 字节是所在主机的唯一标识符。通常是机器主机名的散列值。这样就可以确保不同主机生成不同的ObjectId，不产生冲突。
为了确保在同一台机器上并发的多个进程产生的ObjectId 是唯一的，接下来的两字节来自产生ObjectId 的进程标识符（PID）。
前9 字节保证了同一秒钟不同机器不同进程产生的ObjectId 是唯一的。后3 字节就是一个自动增加的计数器，确保相同进程同一秒产生的ObjectId 也是不一样的。同一秒钟最多允许每个进程拥有2563（16 777 216）个不同的ObjectId。
实现的源码可以到MongoDB官方网站下载。
版权声明：本文内容由互联网用户自发贡献，该文观点仅代表作者本人。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容， 请发送邮件至 举报，一经查实，本站将立刻删除。
文章由极客之音整理，本文链接：https://www.bmabk.com/index.php/post/94147.html
