在程序开发过程中不得不考虑的就是并发问题。在java中对于同一个jvm而言，jdk已经提供了lock和同步等。但是在分布式情况下，往往存在多个进程对一些资源产生竞争关系，而这些进程往往在不同的机器上，这个时候jdk中提供的已经不能满足。分布式锁顾明思议就是可以满足分布式情况下的并发锁。下面我们讲解怎么利用zk实现分布式锁。
ZooKeeper是一个分布式的，开放源码的分布式应用程序协调服务，是Google的Chubby一个开源的实现，是Hadoop和Hbase的重要组件。它是一个为分布式应用提供一致性服务的软件，提供的功能包括：配置维护、域名服务、分布式同步、组服务等。
springBoot集成zk实现分布式锁目录结构
zookeeper配置类
@Configuration
@EnableConfigurationProperties(ZkProps.class)
public class ZkConfig {
private final ZkProps zkProps;
@Autowired
public ZkConfig(ZkProps zkProps) {
this.zkProps = zkProps;
}
@Bean
public Curatorframework curatorframework() {
RetryPolicy retryPolicy = new ExponentialBackoffRetry(zkProps.getTimeout(), zkProps.getRetry());
Curatorframework client = CuratorframeworkFactory.newClient(zkProps.getUrl(), retryPolicy);
client.start();
return client;
}
}
@Data
@ConfigurationProperties(prefix = "zk")
public class ZkProps {
private String url;
private int timeout = 1000;
private int retry = 3;
}
自定义分布式锁注解
@Target({ElementType.PARAMETER})
@Retention(RetentionPolicy.RUNTIME)
@documented
@Inherited
public @interface LockKeyParam {
String[] fields() default {};
}
@Target({ElementType.METHOD})
@Retention(RetentionPolicy.RUNTIME)
@documented
@Inherited
public @interface ZooLock {
String key();
long timeout() default 5 * 1000;
TimeUnit timeUnit() default TimeUnit.MILLISECONDS;
}
日志切入
@Aspect
@Component
@Slf4j
public class ZooLockAspect {
private final Curatorframework zkClient;
private static final String KEY_PREFIX = "DISTRIBUTED_LOCK_";
private static final String KEY_SEPARATOR = "/";
@Autowired
public ZooLockAspect(Curatorframework zkClient) {
this.zkClient = zkClient;
}
@Pointcut("@annotation(com.xkcoding.zookeeper.annotation.ZooLock)")
public void doLock() {
}
@Around("doLock()")
public Object around(ProceedingJoinPoint point) throws Throwable {
MethodSignature signature = (MethodSignature) point.getSignature();
Method method = signature.getMethod();
Object[] args = point.getArgs();
ZooLock zooLock = method.getAnnotation(ZooLock.class);
if (StrUtil.isBlank(zooLock.key())) {
throw new RuntimeException("分布式锁键不能为空");
}
String lockKey = buildLockKey(zooLock, method, args);
InterProcessMutex lock = new InterProcessMutex(zkClient, lockKey);
try {
// 假设上锁成功，以后拿到的都是 false
if (lock.acquire(zooLock.timeout(), zooLock.timeUnit())) {
return point.proceed();
} else {
throw new RuntimeException("请勿重复提交");
}
} finally {
lock.release();
}
}
private String buildLockKey(ZooLock lock, Method method, Object[] args) throws NoSuchFieldException, IllegalAccessException {
StringBuilder key = new StringBuilder(KEY_SEPARATOR + KEY_PREFIX + lock.key());
// 迭代全部参数的注解，根据使用LockKeyParam的注解的参数所在的下标，来获取args中对应下标的参数值拼接到前半部分key上
Annotation[][] parameterAnnotations = method.getParameterAnnotations();
for (int i = 0; i < parameterAnnotations.length; i++) {
// 循环该参数全部注解
for (Annotation annotation : parameterAnnotations[i]) {
// 注解不是 @LockKeyParam
if (!annotation.annotationType().isInstance(LockKeyParam.class)) {
continue;
}
// 获取所有fields
String[] fields = ((LockKeyParam) annotation).fields();
if (ArrayUtil.isEmpty(fields)) {
// 普通数据类型直接拼接
if (ObjectUtil.isNull(args[i])) {
throw new RuntimeException("动态参数不能为null");
}
key.append(KEY_SEPARATOR).append(args[i]);
} else {
// @LockKeyParam的fields值不为null，所以当前参数应该是对象类型
for (String field : fields) {
Class clazz = args[i].getClass();
Field declaredField = clazz.getDeclaredField(field);
declaredField.setAccessible(true);
Object value = declaredField.get(clazz);
key.append(KEY_SEPARATOR).append(value);
}
}
}
}
return key.toString();
}
}
测试类
@RunWith(SpringRunner.class)
@SpringBootTest
@Slf4j
public class SpringBootDemoZookeeperApplicationTests {
public Integer getCount() {
return count;
}
private Integer count = 10000;
private ExecutorService executorService = Executors.newFixedThreadPool(1000);
@Autowired
private Curatorframework zkClient;
@Test
public void test() throws InterruptedException {
IntStream.range(0, 10000).forEach(i -> executorService.execute(this::doBuy));
TimeUnit.MINUTES.sleep(1);
log.error("count值为{}", count);
}
@Test
public void testAopLock() throws InterruptedException {
// 测试类中使用AOP需要手动代理
SpringBootDemoZookeeperApplicationTests target = new SpringBootDemoZookeeperApplicationTests();
AspectJProxyFactory factory = new AspectJProxyFactory(target);
ZooLockAspect aspect = new ZooLockAspect(zkClient);
factory.addAspect(aspect);
SpringBootDemoZookeeperApplicationTests proxy = factory.getProxy();
IntStream.range(0, 10000).forEach(i -> executorService.execute(() -> proxy.aopBuy(i)));
TimeUnit.MINUTES.sleep(1);
log.error("count值为{}", proxy.getCount());
}
@Test
public void testManualLock() throws InterruptedException {
IntStream.range(0, 10000).forEach(i -> executorService.execute(this::manualBuy));
TimeUnit.MINUTES.sleep(1);
log.error("count值为{}", count);
}
@ZooLock(key = "buy", timeout = 1, timeUnit = TimeUnit.MINUTES)
public void aopBuy(int userId) {
log.info("{} 正在出库。。。", userId);
doBuy();
log.info("{} 扣库存成功。。。", userId);
}
public void manualBuy() {
String lockPath = "/buy";
log.info("try to buy sth.");
try {
InterProcessMutex lock = new InterProcessMutex(zkClient, lockPath);
try {
if (lock.acquire(1, TimeUnit.MINUTES)) {
doBuy();
log.info("buy successfully!");
}
} finally {
lock.release();
}
} catch (Exception e) {
log.error("zk error");
}
}
public void doBuy() {
count--;
log.info("count值为{}", count);
}
}
添加微信公众号领取springboot相关资料
