問題背景

背景是調用一個外部接口要做集群限流、想到不依賴過多中間件的方法就是用db的唯一性、比如集群限制并發(fā)限制每秒最多調用10次、之前用的方法是db悲觀鎖、可以參考mysql + spring transaction

悲觀鎖適用于非常嚴格的場景、同時只能有一個調用并且每秒累積不超10次
相對來講樂觀鎖比較簡單、并發(fā)度更好、比如zk transactional和非transactional請求并發(fā)被執(zhí)行、為了盡量并行、zk不是一個請求followe一個請求的執(zhí)行、對transactional應用就是樂觀鎖
通常樂觀鎖實現如下：

image

變種樂觀鎖

這個問題不是版本一致而是當前時間內調用次數、為了避免不同服務器上的時間誤差、統(tǒng)一使用db的時間

image

create database test
use test
CREATE TABLE `optimism_lock_test` (
  `id` bigint(16) NOT NULL AUTO_INCREMENT COMMENT '主鍵id',
  `total_count` tinyint(8) NOT NULL DEFAULT '0' COMMENT '調用次數',
  `create_time` timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP COMMENT '創(chuàng)建時間',
  `update_time` timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT '更新時間',
  PRIMARY KEY (`id`),
  UNIQUE KEY `uniq_idx_create_time` (`create_time`)
) ENGINE=InnoDB AUTO_INCREMENT=1042 DEFAULT CHARSET=utf8mb4 COMMENT='樂觀鎖測試表';

已常用spring + mybatis為例
同步服務、服務入口、當插入失敗后顯示進入exception、釋放當前的transaction、避免插入失敗insertion intention lock還在記錄上造成后續(xù)transaction死鎖
因為每秒最多10次、所以單個transaction保守重試最多9次、避免給db造成無畏的壓力、超過次數就等到下一秒重復這個過程

@Service
public class SyncWaitServiceTest {

    private static final Logger logger = LoggerFactory.getLogger(SyncWaitServiceTest.class);

    @Resource
    private OpLockDaoService oplockDaoService;
    public void syncWait() throws InterruptedException {
        while(true){
            try {
                oplockDaoService.tryToInsert();
                break;
            }catch (DuplicateKeyException e) {
                //INSERT sets an exclusive lock on the inserted row. This lock is an index-record lock,
                // not a next-key lock (that is, there is no gap lock) and does not prevent other sessions
                // from inserting into the gap before the inserted row.Prior to inserting the row,
                // a type of gap lock called an insertion intention gap lock is set. This lock signals the
                // intent to insert in such a way that multiple transactions inserting into the same index gap
                // need not wait for each other if they are not inserting at the same position within the gap.
                // If a duplicate-key error occurs, a shared lock on the duplicate index record is set.
                // This use of a shared lock can result in deadlock should there be multiple sessions
                // trying to insert the same row if another session already has an exclusive lock.
                //用exception 結束transaction、放棄對應s鎖、插入失敗后新的transaction不一定能獲得別的transaction新插入的值
                logger.info("duplicate key");
                int i = 0;
                while(i++ < 10) {
                    int affectCount = oplockDaoService.update();
                    if(affectCount == 1) {
                        return;
                    }
                }
                waitUntilNextSecond();
            }
        }
    }
    private void waitUntilNextSecond() throws InterruptedException {
        LocalDateTime cur = LocalDateTime.now();
        LocalDateTime floor = cur.truncatedTo(ChronoUnit.SECONDS);
        logger.info("[op_wait] sleep for a while until next second");
        TimeUnit.MILLISECONDS.sleep(ChronoUnit.MILLIS.between(cur, floor));
    }
}

DaoService層原封轉到Dao層代碼就不貼了、我們看下mybatis xml中的sql
統(tǒng)一用db的時間、插入失敗后、馬上重試可能已經進入下一秒、新的記錄還沒有插入造成更新一直失敗、這個case比較corner、最終獲得到對應記錄鎖的trancation還是會插入、會影響些效率

<?xml version="1.0" encoding="UTF-8" ?>
<!DOCTYPE mapper PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN" "http://mybatis.org/dtd/mybatis-3-mapper.dtd" >
<mapper namespace="test.OpLockRecordDao">
    <insert id="create">
        insert into optimism_lock_test  (total_count)
        values(1)
    </insert>

    <update id="update">
        update optimism_lock_test
        set
            total_count = total_count + 1,
            update_time = now()
            where create_time = now() and total_count  <![CDATA[ < ]]> 10
    </update>
</mapper>

Test

測試代碼如下、啟動20個線程并發(fā)執(zhí)行50次任務

    @Test
    public void testOP() throws InterruptedException {
        ExecutorService executorService = Executors.newFixedThreadPool(20);

        for (int i = 0; i < 50 ; ++i) {
           executorService.submit(() -> {
               try {
                   syncWaitServiceTest.syncWait();
               } catch (InterruptedException e) {
                   e.printStackTrace();
               }
           });
        }
        executorService.awaitTermination(100, TimeUnit.DAYS);
    }

測試結果如下：

+-------+-------------+---------------------+---------------------+
| id    | total_count | create_time         | update_time         |
+-------+-------------+---------------------+---------------------+
| 10739 |           1 | 2017-11-02 23:18:43 | 2017-11-02 23:18:43 |
| 10761 |          10 | 2017-11-02 23:18:44 | 2017-11-02 23:18:44 |
| 11105 |          10 | 2017-11-02 23:18:45 | 2017-11-02 23:18:45 |
| 11623 |          10 | 2017-11-02 23:18:46 | 2017-11-02 23:18:46 |
| 12113 |          10 | 2017-11-02 23:18:47 | 2017-11-02 23:18:47 |
| 12544 |           9 | 2017-11-02 23:18:48 | 2017-11-02 23:18:48 |
+-------+-------------+---------------------+---------------------+

mysql> select * from optimism_lock_test;
+-------+-------------+---------------------+---------------------+
| id    | total_count | create_time         | update_time         |
+-------+-------------+---------------------+---------------------+
| 13974 |          10 | 2017-11-03 00:14:04 | 2017-11-03 00:14:04 |
| 14324 |          10 | 2017-11-03 00:14:05 | 2017-11-03 00:14:05 |
| 14782 |          10 | 2017-11-03 00:14:06 | 2017-11-03 00:14:06 |
| 15314 |          10 | 2017-11-03 00:14:07 | 2017-11-03 00:14:07 |
| 15756 |          10 | 2017-11-03 00:14:08 | 2017-11-03 00:14:08 |
+-------+-------------+---------------------+---------------------+

總結

簡單幾行代碼就可以實現、不太依賴其他中間件、比較輕量的實現
有一個不太好點的limit在代碼和xml都出現了、尤其是在xml里是沒辦法熱配的、修改需要重新發(fā)布

其實本質來講當前這一秒執(zhí)行了幾次、如果每個機器達成共識、然后協(xié)調來做保證不超過次數
但是這樣有點過度使用了、不過作為實踐還是值得一試
這幾個月太忙了、以后有時間用類似mul paxos的方式讓服務器對當前數目達成共識、比如raft