问题背景

背景是调用一个外部接口要做集群限流、想到不依赖过多中间件的方法就是用db的唯一性、比如集群限制并发限制每秒最多调用10次、之前用的方法是db悲观锁、可以参考mysql + spring transaction

悲观锁适用于非常严格的场景、同时只能有一个调用并且每秒累积不超10次
相对来讲乐观锁比较简单、并发度更好、比如zk transactional和非transactional请求并发被执行、为了尽量并行、zk不是一个请求followe一个请求的执行、对transactional应用就是乐观锁
通常乐观锁实现如下：

image

变种乐观锁

这个问题不是版本一致而是当前时间内调用次数、为了避免不同服务器上的时间误差、统一使用db的时间

实现如下 image
在本机安装db后、执行下面的 sql

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 '创建时间',
  `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还在记录上造成后续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
统一用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个线程并发执行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里是没办法热配的、修改需要重新发布

其实本质来讲当前这一秒执行了几次、如果每个机器达成共识、然后协调来做保证不超过次数
但是这样有点过度使用了、不过作为实践还是值得一试
这几个月太忙了、以后有时间用类似mul paxos的方式让服务器对当前数目达成共识、比如raft