mask/src/main/java/com/sw/service/impl/SnatchMaskServiceImpl.java

package com.sw.service.impl;

import com.sw.activemq.producer.IMessageProducerService;
import com.sw.domain.TMask;
import com.sw.service.SnatchMaskService;
import com.sw.service.TMaskService;
import com.sw.service.TOrderService;
import lombok.extern.slf4j.Slf4j;
import org.redisson.api.RLock;
import org.redisson.api.RedissonClient;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.data.redis.connection.RedisConnection;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import javax.annotation.Resource;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;

/**
 * @Description TODO
 * @Author DingLan
 * @Date 2022/2/18 13:22
 **/
@Service
@Slf4j
public class SnatchMaskServiceImpl implements SnatchMaskService {
    @Autowired
    private TMaskService maskService;

    @Autowired
    private TOrderService orderService;

    @Resource
    private RedisTemplate redisTemplate;

    @Resource
    private IMessageProducerService mqService;

    @Autowired
    private RedissonClient redissonClient;

    @Value("${amount}")
    private Integer amount;
    Object o = new Object();
    ReentrantLock reentrantLock = new ReentrantLock();
    Condition condition = reentrantLock.newCondition();

    @Override
    @Transactional
    public String doSnatch(Integer uid, Integer maskId) {
        //1.get the amount of remaining masks
        TMask mask = maskService.getById(maskId);
        if (mask.getMaskStock() >= amount) {
            //3.create order
            int res = orderService.creatOrder(uid, maskId);
            //4.decrease the stock of mask
            Boolean flag = maskService.decrease(maskId, mask.getMaskStock() - amount);
            if (res > 0 && flag) {

                //3. add to successUserUids
                redisTemplate.opsForSet().add("successUserUids", uid);
                return "恭喜你 抢购成功";
            } else {
                return "抢购失败 稍后重试";
            }
        }
        return "库存不足 抢购失败";
    }

    /**
     * 查看redis库存是否充足
     */
    @Override
    public Boolean checkAndDecrRemaining(Integer mid) {
        RedisConnection connection = redisTemplate.getConnectionFactory().getConnection();
        Boolean flag = false;
        final byte[] key = redisTemplate.getKeySerializer().serialize("mask-setnx-mq-" + mid);
        final byte[] value = redisTemplate.getValueSerializer().serialize(1);
        try {
//
//            while (!connection.setNX(redisTemplate.getKeySerializer().serialize("mask-setnx-mq-" + mid), redisTemplate.getValueSerializer().serialize(1))) {
//                Thread.sleep(3);
//            }
            if(redisTemplate.opsForValue().setIfAbsent("mask-setnx-mq-" + mid, Thread.currentThread().getName(), 1000, TimeUnit.MILLISECONDS)) {
                Integer remaining = Integer.valueOf(redisTemplate.opsForHash().get("usableMasks", "mask-" + mid).toString());
                System.out.println("库存扣减前：" + remaining + "---" + Thread.currentThread());
                if (remaining.compareTo(amount) >= 0) {
                    Long res = redisTemplate.opsForHash().increment("usableMasks", "mask-" + mid, -amount);
                    System.out.println("库存扣减后：" + "increment res : " + res + "---" + Thread.currentThread());//扣减之后的库存数量
                    flag = true;
                } else {
                    flag = false;
                }
                connection.del(key);
                connection.close();
                return flag;
            }else {
                return flag;
            }

        } finally {
            connection.close();
        }


//        reentrantLock.lock();
//        try {
//            Integer remaining = Integer.valueOf(redisTemplate.opsForHash().get("usableMasks", "mask-" + mid).toString());
//            System.out.println("库存扣减前：" + remaining + "---" + Thread.currentThread());
//            if (remaining.compareTo(amount) >= 0) {
//                Long res = redisTemplate.opsForHash().increment("usableMasks", "mask-" + mid, -amount);
//                System.out.println("库存扣减后：" + "increment res : " + res + "---" + Thread.currentThread());//扣减之后的库存数量
//                return true;
//            } else {
//                return false;
//            }
//        } finally {
//            reentrantLock.unlock();
//        }


   /*     synchronized (o) {
//            Integer  remaining = 0;
            Integer   remaining = Integer.valueOf(redisTemplate.opsForHash().get("usableMasks", "mask-" + mid).toString());
            System.out.println("库存扣减前："+remaining+"---" + Thread.currentThread());
            if(remaining.compareTo(amount) >= 0) {
                Long res = redisTemplate.opsForHash().increment("usableMasks", "mask-" + mid, -amount);
                System.out.println("库存扣减后："+"increment res : "+res +"---" + Thread.currentThread());//扣减之后的库存数量
                return true;
            }else{
                return false;
            }
        }*/
    }

    @Override
    public String doSnatchMQ(Integer uid, Integer mid) {
        Map<String, Object> agrs = new HashMap<>();
        agrs.put("uid", uid);
        agrs.put("mid", mid);
        mqService.sendMessage(agrs);
        return "抢购成功 请稍后查看订单";
    }

    /**
     * mq消费端 调用方法 扣减sql库存 创建未支付订单
     *
     * @param uid
     * @param maskId
     */
    @Override
//    @Transactional
    public void doDecrAndCreateOrder(Integer uid, Integer maskId) {
        RLock lock = redissonClient.getLock("mask-" + maskId);
        String threadName = Thread.currentThread().getName();

        try {
            while (lock.tryLock(1000, TimeUnit.MILLISECONDS)) {
                log.info("线程：{} 获取锁", threadName);
                log.info("doSnatch方法执行结果：{}", this.doSnatch(uid, maskId));
                lock.unlock();
                log.info("{}:业务执行完成", threadName);
                break;
            }
        } catch(InterruptedException e){
                e.printStackTrace();
        }
    }
}