項目總結64：分別使用Redisson和Zookeeper分布式鎖模擬模擬搶紅包業務

項目總結64：分別使用Redisson和Zookeeper分布式鎖模擬模擬搶紅包業務

業務場景

　　模擬1000人在10秒內搶10000(或1000)元紅包，金額在1-100不等；

使用的框架或軟件：

　　框架或組件：Springboot(基礎框架)、Redisson(實現分布式鎖)、Zookeeper(實現分布式鎖方案)、Ngnix(負載均衡)，Redis(紅包數據存取數據庫)

　　系統或軟件：Linux服務器、Jmeter(模擬並發請求)

 

具體代碼示例和測試結果(公用方法放在文中附錄)

情況1- 單機服務——沒有任何線程安全考慮——出現數據錯誤

@GetMapping("/get/money")
public String getRedPackage(){

    Map map = new HashMap();
    Object o = redisTemplate.opsForValue().get(KEY_RED_PACKAGE_MONEY);
    int remainMoney = Integer.parseInt(String.valueOf(o));
    if(remainMoney <= 0 ){
        map.put("result","紅包已搶完");
        return ReturnModel.success(map).appendToString();
    }
    int randomMoney = (int) (Math.random() * 100);
    if(randomMoney > remainMoney){
        randomMoney = remainMoney;
    }
    int newRemainMoney = remainMoney-randomMoney;
    redisTemplate.opsForValue().set(KEY_RED_PACKAGE_MONEY,newRemainMoney);
    String result = "原有金額：" + remainMoney + " 紅包金額:" + randomMoney + "剩余金額：" + newRemainMoney;
    System.out.println(result);
    map.put("result",result);

    return ReturnModel.success(map).appendToString();
}

原有金額：1000 紅包金額:49剩余金額：951 原有金額：1000 紅包金額:62剩余金額：938
原有金額：1000 紅包金額:61剩余金額：939
原有金額：1000 紅包金額:93剩余金額：907
原有金額：1000 紅包金額:73剩余金額：927
原有金額：939 紅包金額:65剩余金額：874
原有金額：939 紅包金額:16剩余金額：923
原有金額：939 紅包金額:30剩余金額：909

 

情況2- 單台服務——使用Lock鎖——數據正常；Lock在單服務器是線程安全的

public static Lock lock = new ReentrantLock(); 
@GetMapping("/get/money/lock")
public String getRedPackageLock(){
    Map map = new HashMap();
    lock.lock(); try{
        Object o = redisTemplate.opsForValue().get(KEY_RED_PACKAGE_MONEY);
        int remainMoney = Integer.parseInt(String.valueOf(o));
        if(remainMoney <= 0 ){
            map.put("result","紅包已搶完");
            return ReturnModel.success(map).appendToString();
        }
        int randomMoney = (int) (Math.random() * 100);
        if(randomMoney > remainMoney){
            randomMoney = remainMoney;
        }
        int newRemainMoney = remainMoney-randomMoney;
        redisTemplate.opsForValue().set(KEY_RED_PACKAGE_MONEY,newRemainMoney);
        String result = "原有金額：" + remainMoney + " 紅包金額:" + randomMoney + "剩余金額：" + newRemainMoney;
        System.out.println(result);
        map.put("result",result);
        return ReturnModel.success(map).appendToString();
    }finally {
        lock.unlock();
    }
}

原有金額：1000 紅包金額:11剩余金額：989
原有金額：989 紅包金額:48剩余金額：941
原有金額：941 紅包金額:17剩余金額：924
原有金額：924 紅包金額:89剩余金額：835
原有金額：835 紅包金額:63剩余金額：772
原有金額：772 紅包金額:77剩余金額：695
原有金額：695 紅包金額:76剩余金額：619
原有金額：619 紅包金額:8剩余金額：611
原有金額：611 紅包金額:67剩余金額：544
原有金額：544 紅包金額:9剩余金額：535
原有金額：535 紅包金額:78剩余金額：457
......

 

情況3- 兩台服務器——使用Lock鎖——數據異常(代碼情況2一樣)；Lock在鍍鈦服務器下是非線程安全的

負載均衡配置

　　使用Nginx配置負載均衡，Ngnix安裝參考博客；配置參考博客；部署兩個服務分別是8001和8002端口，Nginx暴露8080端口，轉發請求到8001和8002；

 

 

 nginx配置

http {
    include       mime.types;
    default_type  application/octet-stream;
    sendfile        on;
    keepalive_timeout  65;
    ##定義負載均衡真實服務器IP:端口號 weight表示權重
    upstream myserver{
        server   XX.XX.XX.XX:8001 weight=1;
        server   XX.XX.XX.XX:8002 weight=1;
     }
    server {
        listen   8080;
        location  / {
            proxy_pass   http://myserver;
            proxy_connect_timeout 10;
        }
    }  


}

 

情況3-1- 兩台服務器——使用Redisson分布式鎖——數據正常

        <dependency>
            <groupId>io.netty</groupId>
            <artifactId>netty-all</artifactId>
            <version>4.1.31.Final</version>
        </dependency>
        <dependency>
            <groupId>org.redisson</groupId>
            <artifactId>redisson</artifactId>
            <version>3.6.5</version>
        </dependency>

 

@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;


    @Bean
    public RedissonClient getRedisson(){

        Config config = new Config();
        config.useSingleServer().setAddress("redis://" + host + ":" + port).setPassword(password);
        return Redisson.create(config);
    }

}

 

 

@Autowired private RedissonClient redissonClient; //3-搶紅包-redisson
@GetMapping("/get/money/redisson")
public String getRedPackageRedison(){
 RLock rLock = redissonClient.getLock("secKill"); rLock.lock();
    Map map = new HashMap();
    try{
        Object o = redisTemplate.opsForValue().get(KEY_RED_PACKAGE_MONEY);
        int remainMoney = Integer.parseInt(String.valueOf(o));
        if(remainMoney <= 0 ){
            map.put("result","紅包已搶完");
            return ReturnModel.success(map).appendToString();
        }
        int randomMoney = (int) (Math.random() * 100);
        if(randomMoney > remainMoney){
            randomMoney = remainMoney;
        }
        int newRemainMoney = remainMoney-randomMoney;
        redisTemplate.opsForValue().set(KEY_RED_PACKAGE_MONEY,newRemainMoney);
        String result = "原有金額：" + remainMoney + " 紅包金額:" + randomMoney + "剩余金額：" + newRemainMoney;
        System.out.println(result);
        map.put("result",result);
        return ReturnModel.success(map).appendToString();
    }finally {
      rLock.unlock();
    }
}

 

情況3-2- 兩台服務器——使用Zookeeper分布式鎖——數據正常

 

        <!-- ZooKeeper 之 Curator-->
        <!-- ZooKeeper版本號為4的話，機器安裝zookeeper的版本要求是3.5及其以上的版本-->
        <dependency>
            <groupId>org.apache.curator</groupId>
            <artifactId>curator-framework</artifactId>
            <version>4.0.1</version>
        </dependency>
        <dependency>
            <groupId>org.apache.curator</groupId>
            <artifactId>curator-recipes</artifactId>
            <version>4.0.1</version>
        </dependency>
        <!-- https://mvnrepository.com/artifact/org.apache.curator/curator-client -->
        <dependency>
            <groupId>org.apache.curator</groupId>
            <artifactId>curator-client</artifactId>
            <version>4.0.1</version>
        </dependency>
        <dependency>
            <groupId>org.apache.zookeeper</groupId>
            <artifactId>zookeeper</artifactId>
            <version>3.5.4-beta</version>
        </dependency>
        <dependency>
            <groupId>com.google.collections</groupId>
            <artifactId>google-collections</artifactId>
            <version>1.0</version>
        </dependency>

 

 

@Configuration
public class ZkConfiguration {
    /**
     * 重試次數
     */
    @Value("${curator.retryCount}")
    private int retryCount;
    /**
     * 重試間隔時間
     */
    @Value("${curator.elapsedTimeMs}")
    private int elapsedTimeMs;
    /**
     * 連接地址
     */
    @Value("${curator.connectString}")
    private String connectString;
    /**
     * Session過期時間
     */
    @Value("${curator.sessionTimeoutMs}")
    private int sessionTimeoutMs;
    /**
     * 連接超時時間
     */
    @Value("${curator.connectionTimeoutMs}")
    private int connectionTimeoutMs;

    @Bean(initMethod = "start")
    public CuratorFramework curatorFramework() {
        return CuratorFrameworkFactory.newClient(
                connectString,
                sessionTimeoutMs,
                connectionTimeoutMs,
                new RetryNTimes(retryCount,elapsedTimeMs));
    }

    /**
     * Distributed lock by zookeeper distributed lock by zookeeper.
     *
     * @return the distributed lock by zookeeper
     */
    @Bean(initMethod = "init")
    public DistributedLockByZookeeper distributedLockByZookeeper() {
        return new DistributedLockByZookeeper();
    }
}

 

 

@Slf4j
public class DistributedLockByZookeeper {
    private final static String ROOT_PATH_LOCK = "myk";

    private CountDownLatch countDownLatch = new CountDownLatch(1);

    /**
     * The Curator framework.
     */
    @Autowired
    CuratorFramework curatorFramework;

    /**
     * 獲取分布式鎖
     * 創建一個臨時節點，
     *
     * @param path the path
     */
    public void acquireDistributedLock(String path) {
        String keyPath = "/" + ROOT_PATH_LOCK + "/" + path;
        while (true) {
            try {
                curatorFramework.create()
                        .creatingParentsIfNeeded()
                        .withMode(CreateMode.EPHEMERAL)
                        .withACL(ZooDefs.Ids.OPEN_ACL_UNSAFE)
                        .forPath(keyPath);
                //log.info("success to acquire lock for path:{}", keyPath);
                break;
            } catch (Exception e) {
                //搶不到鎖，進入此處!
                //log.info("failed to acquire lock for path:{}", keyPath);
                //log.info("while try again .......");
                try {
                    if (countDownLatch.getCount() <= 0) {
                        countDownLatch = new CountDownLatch(1);
                    }
                    //避免請求獲取不到鎖，重復的while，浪費CPU資源
                    countDownLatch.await();
                } catch (InterruptedException e1) {
                    e1.printStackTrace();
                }
            }
        }
    }

    /**
     * 釋放分布式鎖
     *
     * @param path the  節點路徑
     * @return the boolean
     */
    public boolean releaseDistributedLock(String path) {
        try {
            String keyPath = "/" + ROOT_PATH_LOCK + "/" + path;
            if (curatorFramework.checkExists().forPath(keyPath) != null) {
                curatorFramework.delete().forPath(keyPath);
            }
        } catch (Exception e) {
            //log.error("failed to release lock,{}", e);
            return false;
        }
        return true;
    }

    /**
     * 創建 watcher 事件
     */
    private void addWatcher(String path) {
        String keyPath;
        if (path.equals(ROOT_PATH_LOCK)) {
            keyPath = "/" + path;
        } else {
            keyPath = "/" + ROOT_PATH_LOCK + "/" + path;
        }
        try {
            final PathChildrenCache cache = new PathChildrenCache(curatorFramework, keyPath, false);
            cache.start(PathChildrenCache.StartMode.POST_INITIALIZED_EVENT);
            cache.getListenable().addListener((client, event) -> {
                if (event.getType().equals(PathChildrenCacheEvent.Type.CHILD_REMOVED)) {
                    String oldPath = event.getData().getPath();
                    //log.info("上一個節點 " + oldPath + " 已經被斷開");
                    if (oldPath.contains(path)) {
                        //釋放計數器，讓當前的請求獲取鎖
                        countDownLatch.countDown();
                    }
                }
            });
        } catch (Exception e) {
            log.info("監聽是否鎖失敗!{}", e);
        }
    }

    /**
     * 創建父節點，並創建永久節點
     */
    public void init() {
        curatorFramework = curatorFramework.usingNamespace("lock-namespace");
        String path = "/" + ROOT_PATH_LOCK;
        try {
            if (curatorFramework.checkExists().forPath(path) == null) {
                curatorFramework.create()
                        .creatingParentsIfNeeded()
                        .withMode(CreateMode.PERSISTENT)
                        .withACL(ZooDefs.Ids.OPEN_ACL_UNSAFE)
                        .forPath(path);
            }
            addWatcher(ROOT_PATH_LOCK);
            log.info("root path 的 watcher 事件創建成功");
        } catch (Exception e) {
            log.error("connect zookeeper fail，please check the log >> {}", e.getMessage(), e);
        }
    }

}

 

 

    @Autowired
    DistributedLockByZookeeper distributedLockByZookeeper;
    private final static String PATH = "red_package";
    //4-搶紅包-zookeeper
    @GetMapping("/get/money/zookeeper")
    public String getRedPackageZookeeper(){

        Boolean flag = false;
        distributedLockByZookeeper.acquireDistributedLock(PATH);
        Map map = new HashMap();
        try {
            Object o = redisTemplate.opsForValue().get(KEY_RED_PACKAGE_MONEY);
            int remainMoney = Integer.parseInt(String.valueOf(o));
            if(remainMoney <= 0 ){
                map.put("result","紅包已搶完");
                return ReturnModel.success(map).appendToString();
            }
            int randomMoney = (int) (Math.random() * 100);
            if(randomMoney > remainMoney){
                randomMoney = remainMoney;
            }
            int newRemainMoney = remainMoney-randomMoney;
            redisTemplate.opsForValue().set(KEY_RED_PACKAGE_MONEY,newRemainMoney);
            String result = "原有金額：" + remainMoney + " 紅包金額:" + randomMoney + "剩余金額：" + newRemainMoney;
            System.out.println(result);
            map.put("result",result);
            return ReturnModel.success(map).appendToString();
        } catch(Exception e){
            e.printStackTrace();
            flag = distributedLockByZookeeper.releaseDistributedLock(PATH);
            //System.out.println("releaseDistributedLock: " + flag);
            map.put("result","getRedPackageZookeeper catch exceeption");
            return ReturnModel.success(map).appendToString();
        }finally {
            flag = distributedLockByZookeeper.releaseDistributedLock(PATH);
            //System.out.println("releaseDistributedLock: " + flag);
        }
    }

 

 

附錄

1- 其他配置和類

application.properties文件

server.port=80

#配置redis
spring.redis.host=XX.XX.XX.XX
spring.redis.port=6379
spring.redis.password=xuegaotest1234
spring.redis.database=0
#重試次數
curator.retryCount=5
#重試間隔時間
curator.elapsedTimeMs=5000
# zookeeper 地址
curator.connectString=XX.XX.XX.XX:2181
# session超時時間
curator.sessionTimeoutMs=60000
# 連接超時時間
curator.connectionTimeoutMs=5000

 

ReturnModel 類

public class ReturnModel implements Serializable{


    private int code;
    private String msg;
    private Object data;




    public static ReturnModel success(Object obj){
        return new ReturnModel(200,"success",obj);
    }

    public String appendToString(){
        return  JSON.toJSONString(this);
    }

    public ReturnModel() {
    }

    public ReturnModel(int code, String msg, Object data) {
        this.code = code;
        this.msg = msg;
        this.data = data;
    }

    public int getCode() {
        return code;
    }

    public void setCode(int code) {
        this.code = code;
    }

    public String getMsg() {
        return msg;
    }

    public void setMsg(String msg) {
        this.msg = msg;
    }

    public Object getData() {
        return data;
    }

    public void setData(Object data) {
        this.data = data;
    }
}

 SeckillController類

    public final static String KEY_RED_PACKAGE_MONEY  = "key_red_package_money";


    @Autowired
    private RedisTemplate redisTemplate;

    //1-設置紅包
    @GetMapping("/set/money/{amount}")
    public String setRedPackage(@PathVariable Integer amount){
        redisTemplate.opsForValue().set(KEY_RED_PACKAGE_MONEY,amount);
        Object o = redisTemplate.opsForValue().get(KEY_RED_PACKAGE_MONEY);
        Map map = new HashMap();
        map.put("moneyTotal",Integer.parseInt(String.valueOf(o)));
        return ReturnModel.success(map).appendToString();
    }

 

 

流程解析

　　1- Zookeeper分布鎖

1- 在ZkConfiguration類中加載CuratorFramework時，設置參數，實例化一個CuratorFramework類； 實例化過程中，執行CuratorFrameworkImpl類中的的start()，其中CuratorFrameworkImpl類是CuratorFramework的實現類；根據具體的細節可以參考博客；

@Bean(initMethod = "start")
public CuratorFramework curatorFramework() {
    return CuratorFrameworkFactory.newClient( connectString,  sessionTimeoutMs, connectionTimeoutMs,  new RetryNTimes(retryCount,elapsedTimeMs));
}

 

2- 在ZkConfiguration類中加載DistributedLockByZookeeper時；執行其中的init()方法；init()方法中主要是創建父節點和添加監聽

/**
 * 創建父節點，並創建永久節點
 */
public void init() {
    curatorFramework = curatorFramework.usingNamespace("lock-namespace");
    String path = "/" + ROOT_PATH_LOCK;
    try {
        if (curatorFramework.checkExists().forPath(path) == null) {
            curatorFramework.create()
                    .creatingParentsIfNeeded()
                    .withMode(CreateMode.PERSISTENT)
                    .withACL(ZooDefs.Ids.OPEN_ACL_UNSAFE)
                    .forPath(path);
        }
        addWatcher(ROOT_PATH_LOCK);
        log.info("root path 的 watcher 事件創建成功");
    } catch (Exception e) {
        log.error("connect zookeeper fail，please check the log >> {}", e.getMessage(), e);
    }
}

 

3- 在具體業務中調用distributedLockByZookeeper.acquireDistributedLock(PATH);獲取分布式鎖

/**
 * 獲取分布式鎖
 * 創建一個臨時節點，
 *
 * @param path the path
 */
public void acquireDistributedLock(String path) {
    String keyPath = "/" + ROOT_PATH_LOCK + "/" + path;
    while (true) {
        try {
            curatorFramework.create()
                    .creatingParentsIfNeeded()
                    .withMode(CreateMode.EPHEMERAL)
                    .withACL(ZooDefs.Ids.OPEN_ACL_UNSAFE)
                    .forPath(keyPath);
            break;
        } catch (Exception e) {
            //搶不到鎖，進入此處!
            try {
                if (countDownLatch.getCount() <= 0) {
                    countDownLatch = new CountDownLatch(1);
                }
                //避免請求獲取不到鎖，重復的while，浪費CPU資源
                countDownLatch.await();
            } catch (InterruptedException e1) {
                e1.printStackTrace();
            }
        }
    }
}

 

4- 業務結束時調用distributedLockByZookeeper.releaseDistributedLock(PATH);釋放鎖

/**
 * 釋放分布式鎖
 *
 * @param path the  節點路徑
 * @return the boolean
 */
public boolean releaseDistributedLock(String path) {
    try {
        String keyPath = "/" + ROOT_PATH_LOCK + "/" + path;
        if (curatorFramework.checkExists().forPath(keyPath) != null) {
            curatorFramework.delete().forPath(keyPath);
        }
    } catch (Exception e) {
        return false;
    }
    return true;
}

 

 原理圖如下

 

 

 

 期間碰到的問題

問題： 項目啟動時：java.lang.ClassNotFoundException: com.google.common.base.Function

原因：缺少google-collections jar包；如下

<dependency>

    <groupId>com.google.collections</groupId>

    <artifactId>google-collections</artifactId>

    <version>1.0</version>

</dependency>

 

 

 

問題：項目啟動時：org.apache.curator.CuratorConnectionLossException: KeeperErrorCode = ConnectionLoss

原因：簡單說，就是連接失敗(可能原因的有很多)；依次排查了zookeeper服務器防火牆、application.properties配置文件;最后發現IP的寫錯了，更正后就好了

 

問題：Jemter啟用多線程並發測試時：java.net.BindException: Address already in use: connect

原因和解決方案：參考博客；

 

END