你手写过一把锁吗？你对轮询缓存怎么看？

当多个线程同时去操作一块内存的数据时如果不做一些限制，极其可能出现数据一致性问题。这时候，我们用一把锁锁住这块数据，持有钥匙者可以进入，不持有者等待钥匙用完再分配。所以在我看来啊，锁的本质就是一个标志位，代表当前线程是否有权限去操作目标内存，但是你的这把锁要穿透当前线程视野，穿透当前实例内存，穿透当前模块层级，到达整个系统可见共享的层次，且处理上要及时释放，再三过滤一切会出现死锁的情况。

所以常见的分布式锁，在可见性由redis缓存实现的解决方案里，通过大家都到redis这块实例上去拿钥匙，恰好进行同一代码块时 通常会将方法名以及时间戳带上某些id等按照一定规则作为key，value不要太大（大key可是会出现问题的，笔者生产环境就遇到过大key造成的数据流异常缓慢直接熔断请求）。为避免死锁也会保证在finally里强制释放锁。

实现lock接口的可重入锁与其使用demo

public class ReentrantTimeoutLock implements Lock {private static class Sync extends AbstractQueuedSynchronizer {private static final int FREE = 0;private static final int LOCKED = 1;private Thread owner = null;private int recursionCount = 0;@Overrideprotected boolean tryAcquire(int arg) {Thread currentThread = Thread.currentThread();int state = getState();if (state == FREE) {if (compareAndSetState(FREE, LOCKED)) {owner = currentThread;recursionCount = 1;return true;}} else if (currentThread == owner) {recursionCount++;return true;}return false;}@Overrideprotected boolean tryRelease(int arg) {if (Thread.currentThread() != owner) {throw new IllegalMonitorStateException("Lock not owned by current thread");}recursionCount--;if (recursionCount == 0) {owner = null;setState(FREE);}return true;}@Overrideprotected boolean isHeldExclusively() {return owner == Thread.currentThread();}Condition newCondition() {return new ConditionObject();}}private final Sync sync = new Sync();private final long timeout;public ReentrantTimeoutLock(long timeout) {this.timeout = timeout;}@Overridepublic void lock() {sync.acquire(1);}@Overridepublic void lockInterruptibly() throws InterruptedException {sync.acquireInterruptibly(1);}@Overridepublic boolean tryLock() {return sync.tryAcquire(1);}@Overridepublic boolean tryLock(long timeout,TimeUnit timeUnit) throws InterruptedException {return sync.tryAcquireNanos(1,timeUnit.toNanos(timeout));}@Overridepublic void unlock() {sync.release(1);}@Overridepublic Condition newCondition() {return sync.newCondition();}
}

使用这把锁我们可以简单做一个接口上的ip限流与请求限流

@Component
public class AccessLimit {private final Lock lock = new ReentrantTimeoutLock(500); // 创建可重入锁对象private final HashMap<String, Long> ipAccesses = new HashMap<>(); // 存储IP地址访问时间private final HashMap<String, Long> apiAccesses = new HashMap<>(); // 存储接口访问时间/*** Limit access.** @param ipAddress      the ip address* @param limitPerSecond the limit per second* @param apiName        the api name*/public void limitAccess(String ipAddress, int limitPerSecond, String apiName) {try {lock.lock(); // 获取锁long currentTime = System.currentTimeMillis();Long lastIPAccess = ipAccesses.get(ipAddress);if (lastIPAccess != null && currentTime - lastIPAccess < 1000 / limitPerSecond) {throw new RuntimeException("IP refuse");}Long lastApiAccess = apiAccesses.get(apiName);if (lastApiAccess != null && currentTime - lastApiAccess < 1000 / limitPerSecond) {throw new RuntimeException("API refuse");}ipAccesses.put(ipAddress, currentTime);apiAccesses.put(apiName, currentTime);} finally {lock.unlock(); // 释放锁}}/*** Release access.** @param ipAddress the ip address* @param apiName   the api name*/public void releaseAccess(String ipAddress, String apiName) {try {lock.lock(); // 获取锁ipAccesses.remove(ipAddress);apiAccesses.remove(apiName);} finally {lock.unlock(); // 释放锁}}
}

真实ip的获取在请求头的x-forwarded-for属性里

String ip = request.getHeader("x-forwarded-for");

对于如何获取真实ip的方法读者自行查找笔者以前的文章。

将限流组件使用在登录接口上如下形式

    @GetMapping("/login")@ResponseBody@ApiOperation(value = "登录", notes = "请求被限制3秒内单一ip无法连续访问，接口3秒内无法连续访问，无需携带token")@ApiImplicitParams({@ApiImplicitParam(name = "managerPhone", value = "管理员电话", required = true, dataTypeClass = String.class),@ApiImplicitParam(name = "password", value = "密码", required = true, dataTypeClass = String.class),@ApiImplicitParam(name = "request", value = "请求对象", required = true, dataTypeClass = HttpServletRequest.class)})public GeneralResponse<Manager> managerLogin(String managerPhone, String password, HttpServletRequest request) {accessLimit.limitAccess(HttpFactory.getIpAddress(request), 3, "managerLogin");ResponseTwoArgInterface<String, String, GeneralResponse<Manager>> responseThreeArgInterface = (value1, value2) -> {Manager manager = managerService.managerLogin(value1, value2);if (Objects.nonNull(manager)) {jedisService.createToken(value1 + "&" + value2, MagicNumber.DEFAULT_OPTION_COUNT, MagicNumber.DEFAULT_TIME_OUT);return GeneralResponse.ServerSuccess(manager, "管理员登录成功");} else {return GeneralResponse.ServerError("管理员登录失败：请检查数据库在线状态或查看日志排查");}};return responseThreeArgInterface.returnResponse(managerPhone, password);}

经过测试可行。

使用redis设计一把分布式锁并设计轮询任务

@Component
public class RedisLock {private static final JedisPool jedisPool = JedisFactory.getJedisPool();private static final Jedis jedis = jedisPool.getResource();public Boolean getLock(String key, int expireTime) {try {jedis.auth(RedisConfig.AuthPassword);/*获取锁，如果上锁成功返回1*/Long lockIsSuccess = jedis.setnx(key, System.currentTimeMillis() + "");if (lockIsSuccess != null && lockIsSuccess == 1L) {/*锁计时，每一把锁都应设置一个计时释放的时间*/jedis.expire(key, expireTime);return true;} else {String lockGoneTime = jedis.get(key);if (lockGoneTime == null) {lockIsSuccess = jedis.setnx(key, System.currentTimeMillis() + "");if (lockIsSuccess != null && lockIsSuccess == 1L) {jedis.expire(key, expireTime);}return lockIsSuccess != null && lockIsSuccess == 1L;} else {long currentTimeMillis = System.currentTimeMillis();if (currentTimeMillis - Long.parseLong(lockGoneTime) < expireTime * 1000L) {return false;} else {String lockNowTime = jedis.getSet(key, currentTimeMillis + "");if (lockNowTime == null || lockNowTime.equals(lockGoneTime)) {jedis.expire(key, expireTime);}return lockNowTime == null || lockNowTime.equals(lockGoneTime);}}}}finally {jedis.close();}}public void unLock(String key){try {jedis.auth(RedisConfig.AuthPassword);jedis.expire(key, 0);}finally {jedis.close();}}
}

这把锁也是可重入锁，这里的key读者可以自行设计，需要注意的是redis连接池设置，以及客户端连接后要及时释放资源；

将这把锁运用到轮询任务组件中的代码如下：

@Component
@DependsOn(value = {"ThreadPool"})
@Slf4j
public class RoundCheckTask {@Resourceprivate ThreadPoolTaskExecutor threadPool;@Resourceprivate RedisLock redisLock;private static final JedisPool jedisPool = JedisFactory.getJedisPool();private static final Jedis jedis = jedisPool.getResource();public void roundCheckCache(String key, String taskId, int roundTime) {jedis.auth(RedisConfig.AuthPassword);/*尝试获取一把轮询任务的锁,key由时间戳组成保证并发抢锁*/Boolean lock = redisLock.getLock(key, roundTime);if (lock) {try {long start = System.currentTimeMillis();threadPool.execute(() -> {while (true) {Long ttl = jedis.ttl(taskId);if (ttl == null || ttl == 0L) {runTaskThread(taskId, roundTime);break;}if (System.currentTimeMillis() - start > roundTime && ttl > roundTime) {/*循环进入的时间大于轮询时长直接退出轮询*/break;}try {Thread.sleep(500);} catch (Exception e) {throw new RuntimeException("thread sleep exception");}}}, MagicNumber.DEFAULT_TIME_OUT);} catch (Exception e) {throw new RuntimeException("start round check thread is fail");} finally {redisLock.unLock(key);jedis.close();}} else {jedis.close();}}public void runTaskThread(String taskId, int runTime) {jedis.auth(RedisConfig.AuthPassword);Boolean lock = redisLock.getLock(taskId, runTime);if (lock) {try {CountDownLatch countDownLatch = new CountDownLatch(1);threadPool.execute(() -> {/*执行业务逻辑*/System.out.println(taskId);countDownLatch.countDown();});countDownLatch.await();} catch (Exception e) {throw new RuntimeException("task service running error");} finally {redisLock.unLock(taskId);jedis.close();}} else {jedis.close();}}}

这里轮询的逻辑也就不仔细讲解了，核心的地方就在

                        Long ttl = jedis.ttl(taskId);if (ttl == null || ttl == 0L) {runTaskThread(taskId, roundTime);break;}if (System.currentTimeMillis() - start > roundTime && ttl > roundTime) {/*循环进入的时间大于轮询时长直接退出轮询*/break;}try {Thread.sleep(500);} catch (Exception e) {throw new RuntimeException("thread sleep exception");}

此处使用计数器来等待业务逻辑的执行。

对于redis锁的实现方案，redisson是不错的选型，有兴趣的读者可以了解了解redisson的redlock

最后笔者祝各位劳动节快乐~