现在我们考察一下在避免共享数据之间的竞争条件时，内核控制路径是如何交错执行的。表5-2列出了Linux内核使用的同步技术。”适用范围”一栏表示同步技术适用于系统中的所有CPU还是单个CPU。例如，本地中断的禁止只适用于一个CPU；相反原子操作影响系统中的所有CPU。
现在，让我们简要地描述每种同步技术。在后面”对内核数据结构的同步访问”一节，我们会说明如何把这些同步技术组合在一起来有效地保护内核数据结构。
每CPU变量
最好的同步技术是把设计不需要同步的内核放在首位。正如我们将要看到的，事实上每一种显式的同步有的放矢都有不容忽视的性能开销。
最简单也是最重要的同步技术包括把内核变量声明为每CPU变量。每CPU变量主要是数据结构的数组，系统的每个CPU对应数组的一个元素。
一个CPU不应该访问与其它CPU对应的数组元素，另外，它可以随意读或修改它自己元素而不用担心出现竞争条件，因为它是唯一有资格这么做的CPU。但是，这也意味着每CPU变量基本上只能在特殊情况下使用，也就是当它确定在系统的CPU上的数据在逻辑上是独立的时候。
每CPU的数组元素在主存中被排列以使每个数据结构存放在硬件高速缓存的不同行，因此，对每CPU数组的并发访问不会导致高速缓存行的窃用和失效。
虽然每CPU变量为来自不同CPU的并发访问提供保护，但对来自异步函数的访问不提供保护，在这种情况下需要另外的同步原语。
此外，在单处理器和多处理器系统中，内核抢占都可能使每CPU变量产生竞争条件。总的原则是内核控制路径应该在禁用抢占的情况下访问每CPU变量。作为一个例子，简单地考虑一下在下面这种情况下会产生什么后果--------一个内核控制路径获得了它的每CPU变量本地副本的地址，然后它因被抢占而转移到另外一个CPU上，但仍然引用原来CPU元素的地址。
表5-3列出了内核提供使用每CPU变量的函数和宏。

6.修改锁的公平性
ReentrantLock和ReentrantReadWriteLock类的构造器都含有一个布尔参数fair，它允许你控制这两个类的行为。默认fair值是false，他称为非公平模式。在非公平模式下，当有很多线程在等待锁时，锁将选择他们中的一个来访问临界区，这个选择是没有任何约束的。如果fair值是true，则称为公平模式，锁会选择等待时间最长的访问临界区。这两种模式只适用于lock()和unlock()方法。而Lock接口的tryLock()方法没有将线程置于休眠，fair属性并不影响这个方法。
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

public class PrintQueue {
	private final Lock queueLock = new ReentrantLock(true);
	public void printJob(Object object) {
		queueLock.lock();
		try {
			long duration = (long)(Math.random()*10000);
			System.out.println(Thread.currentThread().getName()+":PrintQueue: Printing a Job during "+(duration/1000)+" seconds");
			Thread.sleep(duration);
		} catch (InterruptedException e) {
			// TODO: handle exception
		}finally {
			queueLock.unlock();
		}
		queueLock.lock();
		try {
			long duration = (long)(Math.random()*10000);
			System.out.println(Thread.currentThread().getName()+":PrintQueue: Printing a Job during "+(duration/1000)+" seconds");
			Thread.sleep(duration);
		} catch (InterruptedException e) {
			// TODO: handle exception
		}finally {
			queueLock.unlock();
		}
	}
}
public class Job implements Runnable{
	private PrintQueue printQueue;
	

	public Job(PrintQueue printQueue) {
		this.printQueue = printQueue;
	}


	@Override
	public void run() {
		// TODO Auto-generated method stub
		System.out.println(Thread.currentThread().getName()+": Going to print a document");
		printQueue.printJob(new Object());
		System.out.println(Thread.currentThread().getName()+": The document has been printed");
	}

}
public class Main {
	public static void main(String[] args) {
		PrintQueue printQueue = new PrintQueue();
		Thread[] threads = new Thread[10];
		for (int i = 0; i < 10; i++) {
			threads[i] = new Thread(new Job(printQueue),"Thread "+i);
		}
		
		for (int i = 0; i < 10; i++) {
			threads[i].start();
			try {
				Thread.sleep(100);
			} catch (InterruptedException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			}
		}
	}
}

7.在锁中使用多条件
一个锁可能关联一个或者多个条件，这些条件通过Condition接口声明。目的是允许线程获取锁并且查看等待的某一个条件是否满足，如果不满足就挂起直到某个线程唤醒他们。Condition接口提供了挂起线程和唤起线程的机制。
public class FileMock {
	private String[] content;
	private int index;
	public FileMock(int size, int length) {
		content = new String[size];
		StringBuffer buffer = new StringBuffer(length);
		for (int i = 0; i < content.length; i++) {
			for (int j = 0; j < length; j++) {
				int indice = (int)Math.random()*255;
				buffer.append((char)indice);
			}	
			content[i] = buffer.toString();
		}
		index = 0;
	}
	public boolean hasMoreLines() {
		return index<content.length;
	}
	public String  getLine() {
		if (this.hasMoreLines()) {
			System.out.println("Mock: "+(content.length-index));
			return content[index+1];
		}
		return null;
	}
}
import java.util.LinkedList;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;

public class Buffer {
	private LinkedList<String> buffer;
	private int maxSize;
	private ReentrantLock lock;
	private Condition lines;
	private Condition space;
	private boolean pendingLines;
	public Buffer(int maxSize) {
		this.maxSize = maxSize;
		buffer = new LinkedList<>();
		lock = new ReentrantLock();
		lines = lock.newCondition();
		space = lock.newCondition();
		pendingLines = true;
	}

	public void insert(String line) {
		lock.lock();
		try {
			while(buffer.size()==maxSize) {
				space.await();
			}
			buffer.offer(line);
			System.out.println(Thread.currentThread().getName()+": Inserted Line: "+buffer.size());
			lines.signalAll();
		} catch (InterruptedException e) {
			e.printStackTrace();
		}finally {
			lock.unlock();
		}
	}
	public String  get() {
		String line = null;
		lock.lock();
		try {
			while((buffer.size()==0)&&(hasPendingLines())) {
				lines.await();
			}
			if (hasPendingLines()) {
				line = buffer.poll();
				System.out.println(Thread.currentThread().getName()+": Line Readed: "+buffer.size());
				space.signalAll();
			}
		} catch (InterruptedException e) {
			// TODO: handle exception
			e.printStackTrace();
		}finally {
			lock.unlock();
		}
		return line;
	}
	public void setPendingLines(boolean pendingLines) {
		this.pendingLines = pendingLines;
	}
	public boolean hasPendingLines() {
		return pendingLines || buffer.size()>0;
	}
}
public class Producer implements Runnable{

	private FileMock mock;
	private Buffer buffer;
	
	public Producer(FileMock mock, Buffer buffer) {
		this.mock = mock;
		this.buffer = buffer;
	}

	@Override
	public void run() {
		// TODO Auto-generated method stub
		buffer.setPendingLines(true);
		while(mock.hasMoreLines()) {
			String line = mock.getLine();
			buffer.insert(line);
		}
		buffer.setPendingLines(false);
	}

}
public class Consumer implements Runnable{

	private Buffer buffer;
	public Consumer(Buffer buffer) {
		this.buffer = buffer;
	}
	@Override
	public void run() {
		// TODO Auto-generated method stub
		while(buffer.hasPendingLines()) {
			String line = buffer.get();
			processLine(line);
		}
	}
	private void processLine(String line) {
		try {
			Random random = new Random();
			Thread.sleep(random.nextInt(100));
		} catch (InterruptedException e) {
			// TODO: handle exception
			e.printStackTrace();
		}
	}

}
public class Main {
	public static void main(String[] args) {
		FileMock mock = new FileMock(100, 10);
		Buffer buffer = new Buffer(20);
		Producer producer = new Producer(mock, buffer);
		Thread thread1 = new Thread(producer,"Producer");
		Consumer[] consumers= new Consumer[3];
		Thread[] threads = new Thread[3];
		for (int i = 0; i < 3; i++) {
			consumers[i] = new Consumer(buffer);
			threads[i] = new Thread(consumers[i],"Consumer "+i);
		}
		
		thread1.start();
		for (int i = 0; i < 3; i++) {
			threads[i].start();
		}
	}
}
与锁绑定的所有条件对象都是通过Lock接口声明的newCondition()方法创建的。在使用条件的时候，必须获取这个条件绑定的锁，所以带条件的代码必须在调用Lock对象的lock()方法和unlock()方法之间。
当线程调用条件的await()方法时，他将自动释放这个条件绑定的锁，其他某个线程才可以获取这个锁并且执行相同的操作，或者执行这个锁保护的另一个临界区代码。
Condition接口还提供了await()方法的其他形式：



await(long time,TimeUnit unit)，直到发生以下情况之一前，线程将一直处于休眠状态


其他某个线程中断当前线程
其他某个线程调用了将当前线程挂起的条件的signal()或signalAll()方法
指定的等待时间已经过去



awaitUninterruptibly()：它是不可中断的。这个线程将休眠直到其他某个线程调用了将他挂起的条件的signal()或signalAll()方法

awaitUntil(Date date):直到发生以下情况之一前，线程将一直处于休眠状态

其他的某个线程中断当前线程
其他某个线程调用了将他挂起的条件的signal()或signalAll()方法
指定的最后期限到了


也可以将条件和读写锁一起使用

前言
RateLimiter通过线程锁控制同步，只适用于单机应用，在分布式环境下，虽然有像阿里Sentinel的限流开源框架，但对于一些小型应用来说未免过重，但限流的需求在小型项目中也是存在的，比如获取手机验证码的控制，对资源消耗较大操作的访问频率控制等。本文介绍最近写的一个基于RateLimiter，适用于分布式环境下的限流实现，并使用spring-boot-starter的形式发布，比较轻量级且“开箱即用”。
限流的两种形式

基于RateLimiter令牌桶算法的限速控制（严格限制访问速度）
基于Lua脚本的限量控制（限制一个时间窗口内的访问量，对访问速度没有严格限制）

- 限速控制
1. 令牌桶模型
首先定义令牌桶模型，与RateLimiter中类似，包括几个关键属性与关键方法。其中关键属性定义如下，
@Data
public class RedisPermits {

    /**
     * 最大存储令牌数
     */
    private double maxPermits;
    /**
     * 当前存储令牌数
     */
    private double storedPermits;
    /**
     * 添加令牌的时间间隔/毫秒
     */
    private double intervalMillis;
    /**
     * 下次请求可以获取令牌的时间，可以是过去（令牌积累）也可以是将来的时间（令牌预消费）
     */
    private long nextFreeTicketMillis;

    //...

关键方法定义与RateLimiter也大同小异，方法注释基本已描述各方法用途。
/**
     * 构建Redis令牌数据模型
     *
     * @param permitsPerSecond     每秒放入的令牌数
     * @param maxBurstSeconds      maxPermits由此字段计算，最大存储maxBurstSeconds秒生成的令牌
     * @param nextFreeTicketMillis 下次请求可以获取令牌的起始时间，默认当前系统时间
     */
    public RedisPermits(double permitsPerSecond, double maxBurstSeconds, Long nextFreeTicketMillis) {
        this.maxPermits = permitsPerSecond * maxBurstSeconds;
        this.storedPermits = maxPermits;
        this.intervalMillis = TimeUnit.SECONDS.toMillis(1) / permitsPerSecond;
        this.nextFreeTicketMillis = nextFreeTicketMillis;
    }

    /**
     * 基于当前时间，若当前时间晚于nextFreeTicketMicros，则计算该段时间内可以生成多少令牌，将生成的令牌加入令牌桶中并更新数据
     */
    public void resync(long nowMillis) {
        if (nowMillis > nextFreeTicketMillis) {
            double newPermits = (nowMillis - nextFreeTicketMillis) / intervalMillis;
            storedPermits = Math.min(maxPermits, storedPermits + newPermits);
            nextFreeTicketMillis = nowMillis;
        }
    }

    /**
    * 保留指定数量令牌，并返回需要等待的时间
    */
    public long reserveAndGetWaitLength(long nowMillis, int permits) {
        resync(nowMillis);
        double storedPermitsToSpend = Math.min(permits, storedPermits); // 可以消耗的令牌数
        double freshPermits = permits - storedPermitsToSpend; // 需要等待的令牌数
        long waitMillis = (long) (freshPermits * intervalMillis); // 需要等待的时间

        nextFreeTicketMillis = LongMath.saturatedAdd(nextFreeTicketMillis, waitMillis);
        storedPermits -= storedPermitsToSpend;
        return waitMillis;
    }

    /**
    * 在超时时间内，是否有指定数量的令牌可用
    */
    public boolean canAcquire(long nowMillis, int permits, long timeoutMillis) {
        return queryEarliestAvailable(nowMillis, permits) <= timeoutMillis;
    }

    /**
     * 指定数量令牌数可用需等待的时间
     *
     * @param permits 需保留的令牌数
     * @return 指定数量令牌可用的等待时间，如果为0或负数，表示当前可用
     */
    private long queryEarliestAvailable(long nowMillis, int permits) {
        resync(nowMillis);
        double storedPermitsToSpend = Math.min(permits, storedPermits); // 可以消耗的令牌数
        double freshPermits = permits - storedPermitsToSpend; // 需要等待的令牌数
        long waitMillis = (long) (freshPermits * intervalMillis); // 需要等待的时间

        return LongMath.saturatedAdd(nextFreeTicketMillis - nowMillis, waitMillis);
    }

2. 令牌桶控制类
Guava RateLimiter中的控制都在RateLimiter及其子类中（如SmoothBursty），本处涉及到分布式环境下的同步，因此将其解耦，令牌桶模型存储于Redis中，对其同步操作的控制放置在如下控制类，其中同步控制使用到了前面介绍的分布式锁（参考基于Redis分布式锁的正确打开方式）
@Slf4j
public class RedisRateLimiter {

    /**
     * 获取一个令牌，阻塞一直到获取令牌，返回阻塞等待时间
     *
     * @return time 阻塞等待时间/毫秒
     */
    public long acquire(String key) throws IllegalArgumentException {
        return acquire(key, 1);
    }

    /**
     * 获取指定数量的令牌，如果令牌数不够，则一直阻塞，返回阻塞等待的时间
     *
     * @param permits 需要获取的令牌数
     * @return time 等待的时间/毫秒
     * @throws IllegalArgumentException tokens值不能为负数或零
     */
    public long acquire(String key, int permits) throws IllegalArgumentException {
        long millisToWait = reserve(key, permits);
        log.info("acquire {} permits for key[{}], waiting for {}ms", permits, key, millisToWait);
        try {
            Thread.sleep(millisToWait);
        } catch (InterruptedException e) {
            log.error("Interrupted when trying to acquire {} permits for key[{}]", permits, key, e);
        }
        return millisToWait;
    }

    /**
     * 在指定时间内获取一个令牌，如果获取不到则一直阻塞，直到超时
     *
     * @param timeout 最大等待时间（超时时间），为0则不等待立即返回
     * @param unit    时间单元
     * @return 获取到令牌则true，否则false
     * @throws IllegalArgumentException
     */
    public boolean tryAcquire(String key, long timeout, TimeUnit unit) throws IllegalArgumentException {
        return tryAcquire(key, 1, timeout, unit);
    }

    /**
     * 在指定时间内获取指定数量的令牌，如果在指定时间内获取不到指定数量的令牌，则直接返回false，
     * 否则阻塞直到能获取到指定数量的令牌
     *
     * @param permits 需要获取的令牌数
     * @param timeout 最大等待时间（超时时间）
     * @param unit    时间单元
     * @return 如果在指定时间内能获取到指定令牌数，则true,否则false
     * @throws IllegalArgumentException tokens为负数或零，抛出异常
     */
    public boolean tryAcquire(String key, int permits, long timeout, TimeUnit unit) throws IllegalArgumentException {
        long timeoutMillis = Math.max(unit.toMillis(timeout), 0);
        checkPermits(permits);

        long millisToWait;
        boolean locked = false;
        try {
            locked = lock.lock(key + LOCK_KEY_SUFFIX, WebUtil.getRequestId(), 60, 2, TimeUnit.SECONDS);
            if (locked) {
                long nowMillis = getNowMillis();
                RedisPermits permit = getPermits(key, nowMillis);
                if (!permit.canAcquire(nowMillis, permits, timeoutMillis)) {
                    return false;
                } else {
                    millisToWait = permit.reserveAndGetWaitLength(nowMillis, permits);
                    permitsRedisTemplate.opsForValue().set(key, permit, expire, TimeUnit.SECONDS);
                }
            } else {
                return false;  //超时获取不到锁，也返回false
            }
        } finally {
            if (locked) {
                lock.unLock(key + LOCK_KEY_SUFFIX, WebUtil.getRequestId());
            }
        }
        if (millisToWait > 0) {
            try {
                Thread.sleep(millisToWait);
            } catch (InterruptedException e) {

            }
        }
        return true;
    }

    /**
     * 保留指定的令牌数待用
     *
     * @param permits 需保留的令牌数
     * @return time 令牌可用的等待时间
     * @throws IllegalArgumentException tokens不能为负数或零
     */
    private long reserve(String key, int permits) throws IllegalArgumentException {
        checkPermits(permits);
        try {
            lock.lock(key + LOCK_KEY_SUFFIX, WebUtil.getRequestId(), 60, 2, TimeUnit.SECONDS);
            long nowMillis = getNowMillis();
            RedisPermits permit = getPermits(key, nowMillis);
            long waitMillis = permit.reserveAndGetWaitLength(nowMillis, permits);
            permitsRedisTemplate.opsForValue().set(key, permit, expire, TimeUnit.SECONDS);
            return waitMillis;
        } finally {
            lock.unLock(key + LOCK_KEY_SUFFIX, WebUtil.getRequestId());
        }
    }

    /**
     * 获取令牌桶
     *
     * @return
     */
    private RedisPermits getPermits(String key, long nowMillis) {
        RedisPermits permit = permitsRedisTemplate.opsForValue().get(key);
        if (permit == null) {
            permit = new RedisPermits(permitsPerSecond, maxBurstSeconds, nowMillis);
        }
        return permit;
    }

    /**
     * 获取redis服务器时间
     */
    private long getNowMillis() {
        String luaScript = "return redis.call('time')";
        DefaultRedisScript<List> redisScript = new DefaultRedisScript<>(luaScript, List.class);
        List<String> now = (List<String>)stringRedisTemplate.execute(redisScript, null);
        return now == null ? System.currentTimeMillis() : Long.valueOf(now.get(0))*1000+Long.valueOf(now.get(1))/1000;
    }

    //...
}

其中：
1、acquire 是阻塞方法，如果没有可用的令牌，则一直阻塞直到获取到令牌。
2、tryAcquire 则是非阻塞方法，如果在指定超时时间内获取不到指定数量的令牌，则直接返回false，不阻塞等待。
3、getNowMillis 获取Redis服务器时间，避免业务服务器时间不一致导致的问题，如果业务服务器能保障时间同步，则可从本地获取提高效率。

3. 令牌桶控制工厂类
工厂类负责管理令牌桶控制类，将其缓存在本地，这里使用了Guava中的Cache，一方面避免每次都新建控制类提高效率，另一方面通过控制缓存的最大容量来避免像用户粒度的限流占用过多的内存。
public class RedisRateLimiterFactory {

    private PermitsRedisTemplate permitsRedisTemplate;
    private StringRedisTemplate stringRedisTemplate;
    private DistributedLock distributedLock;

    private Cache<String, RedisRateLimiter> cache = CacheBuilder.newBuilder()
            .initialCapacity(100)  //初始大小
            .maximumSize(10000) // 缓存的最大容量
            .expireAfterAccess(5, TimeUnit.MINUTES) // 缓存在最后一次访问多久之后失效
            .concurrencyLevel(Runtime.getRuntime().availableProcessors()) // 设置并发级别
            .build();

    public RedisRateLimiterFactory(PermitsRedisTemplate permitsRedisTemplate, StringRedisTemplate stringRedisTemplate, DistributedLock distributedLock) {
        this.permitsRedisTemplate = permitsRedisTemplate;
        this.stringRedisTemplate = stringRedisTemplate;
        this.distributedLock = distributedLock;
    }

    /**
     * 创建RateLimiter
     *
     * @param key              RedisRateLimiter本地缓存key
     * @param permitsPerSecond 每秒放入的令牌数
     * @param maxBurstSeconds  最大存储maxBurstSeconds秒生成的令牌
     * @param expire           该令牌桶的redis tty/秒
     * @return RateLimiter
     */
    public RedisRateLimiter build(String key, double permitsPerSecond, double maxBurstSeconds, int expire) {
        if (cache.getIfPresent(key) == null) {
            synchronized (this) {
                if (cache.getIfPresent(key) == null) {
                    cache.put(key, new RedisRateLimiter(permitsRedisTemplate, stringRedisTemplate, distributedLock, permitsPerSecond,
                            maxBurstSeconds, expire));
                }
            }
        }
        return cache.getIfPresent(key);
    }
}

4. 注解支持
定义注解 @RateLimit 如下，表示以每秒rate的速率放置令牌，最多保留burst秒的令牌，取令牌的超时时间为timeout，limitType用于控制key类型，目前支持：
1、IP, 根据客户端IP限流
2、USER, 根据用户限流，对于Spring Security可从SecurityContextHolder中获取当前用户信息，如userId
3、METHOD, 根据方法名全局限流，className.methodName，注意避免同时对同一个类中的同名方法做限流控制，否则需要修改获取key的逻辑
4、CUSTOM，自定义，支持表达式解析，如#{id}, #{user.id}

@Retention(RetentionPolicy.RUNTIME)
@Target(ElementType.METHOD)
public @interface RateLimit {
    String key() default "";
    String prefix() default "rateLimit:"; //key前缀
    int expire() default 60; // 表示令牌桶模型RedisPermits redis key的过期时间/秒
    double rate() default 1.0; // permitsPerSecond值
    double burst() default 1.0; // maxBurstSeconds值
    int timeout() default 0; // 超时时间/秒
    LimitType limitType() default LimitType.METHOD;
}

通过切面的前置增强来为添加了 @RateLimit 注解的方法提供限流控制，如下
@Aspect
@Slf4j
public class RedisLimitAspect {
    //...

    @Before(value = "@annotation(rateLimit)")
    public void rateLimit(JoinPoint  point, RateLimit rateLimit) throws Throwable {
        String key = getKey(point, rateLimit.limitType(), rateLimit.key(), rateLimit.prefix());
        RedisRateLimiter redisRateLimiter = redisRateLimiterFactory.build(key, rateLimit.rate(), rateLimit.burst(), rateLimit.expire());
        if(!redisRateLimiter.tryAcquire(key, rateLimit.timeout(), TimeUnit.SECONDS)){
            ExceptionUtil.rethrowClientSideException(LIMIT_MESSAGE);
        }
    }

    //...

- 限量控制
1. 限量控制类
限制一个时间窗口内的访问量，可使用计数器算法，借助Lua脚本执行的原子性来实现。
Lua脚本逻辑：
以需要控制的对象为key（如方法，用户ID，或IP等），当前访问次数为Value，时间窗口值为缓存的过期时间

如果key存在则将其增1，判断当前值是否大于访问量限制值，如果大于则返回0，表示该时间窗口内已达访问量上限，如果小于则返回1表示允许访问

如果key不存在，则将其初始化为1，并设置过期时间，返回1表示允许访问
public class RedisCountLimiter {

    private StringRedisTemplate stringRedisTemplate;

    private static final String LUA_SCRIPT = "local c \nc = redis.call('get',KEYS[1]) \nif c and redis.call('incr',KEYS[1]) > tonumber(ARGV[1]) then return 0 end"
            + " \nif c then return 1 else \nredis.call('set', KEYS[1], 1) \nredis.call('expire', KEYS[1], tonumber(ARGV[2])) \nreturn 1 end";

    private static final int SUCCESS_RESULT = 1;
    private static final int FAIL_RESULT = 0;

    public RedisCountLimiter(StringRedisTemplate stringRedisTemplate) {
        this.stringRedisTemplate = stringRedisTemplate;
    }

    /**
     * 是否允许访问
     *
     * @param key redis key
     * @param limit 限制次数
     * @param expire 时间段/秒
     * @return 获取成功true，否则false
     * @throws IllegalArgumentException
     */
    public boolean tryAcquire(String key, int limit, int expire) throws IllegalArgumentException {
        RedisScript<Number> redisScript = new DefaultRedisScript<>(LUA_SCRIPT, Number.class);
        Number result = stringRedisTemplate.execute(redisScript, Collections.singletonList(key), String.valueOf(limit), String.valueOf(expire));
        if(result != null && result.intValue() == SUCCESS_RESULT) {
            return true;
        }
        return false;
    }

}

2. 注解支持
定义注解 @CountLimit 如下，表示在period时间窗口内，最多允许访问limit次，limitType用于控制key类型，取值与 @RateLimit 同。
@Retention(RetentionPolicy.RUNTIME)
@Target(ElementType.METHOD)
public @interface CountLimit {
    String key() default "";
    String prefix() default "countLimit:"; //key前缀
    int limit() default 1;  // expire时间段内限制访问次数
    int period() default 1; // 表示时间段/秒
    LimitType limitType() default LimitType.METHOD;
}

同样采用前值增强来为添加了 @CountLimit 注解的方法提供限流控制，如下
@Before(value = "@annotation(countLimit)")
public void countLimit(JoinPoint  point, CountLimit countLimit) throws Throwable {
    String key = getKey(point, countLimit.limitType(), countLimit.key(), countLimit.prefix());
    if (!redisCountLimiter.tryAcquire(key, countLimit.limit(), countLimit.period())) {
        ExceptionUtil.rethrowClientSideException(LIMIT_MESSAGE);
    }

示例演示
1. 添加依赖
<dependencies>
    <dependency>
        <groupId>cn.jboost.springboot</groupId>
        <artifactId>limiter-spring-boot-starter</artifactId>
        <version>1.3-SNAPSHOT</version>
    </dependency>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-web</artifactId>
    </dependency>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-data-redis</artifactId>
    </dependency>
</dependencies>


配置redis相关参数

spring:
  application:
    name: limiter-demo
  redis:
    #数据库索引
    database: 0
    host: 192.168.40.92
    port: 6379
    password: password
    #连接超时时间
    timeout: 2000


测试类

@RestController
@RequestMapping("limiter")
public class LimiterController {

    /**
     * 注解形式
     * @param key
     * @return
     */
    @GetMapping("/count")
    @CountLimit(key = "#{key}", limit = 2, period = 10, limitType = LimitType.CUSTOM)
    public String testCountLimit(@RequestParam("key") String key){
        return "test count limiter...";
    }

    /**
     * 注解形式
     * @param key
     * @return
     */
    @GetMapping("/rate")
    @RateLimit(rate = 1.0/5, burst = 5.0, expire = 120, timeout = 0)
    public String testRateLimit(@RequestParam("key") String key){
        return "test rate limiter...";
    }

    @Autowired
    private RedisRateLimiterFactory redisRateLimiterFactory;
    /**
     * 代码段形式
     * @param
     * @return
     */
    @GetMapping("/rate2")
    public String testRateLimit(){
        RedisRateLimiter limiter = redisRateLimiterFactory.build("LimiterController.testRateLimit", 1.0/30, 30, 120);
        if(!limiter.tryAcquire("app.limiter", 0, TimeUnit.SECONDS)) {
            System.out.println(LocalDateTime.now());
            ExceptionUtil.rethrowClientSideException("您的访问过于频繁，请稍后重试");
        }
        return "test rate limiter 2...";
    }
}


验证


启动测试项目，浏览器中访问 http://localhost:8080/limiter/rate?key=test ，第一次访问成功，如图

持续刷新，将返回如下错误，直到5s之后再返回成功，限制5秒1次的访问速度


注解的使用
限流类型LimitType支持IP（客户端IP）、用户（userId）、方法（className.methodName）、自定义（CUSTOM）几种形式，默认为METHOD

LimitType为CUSTOM时，需要手动指定key（其它key自动为ip，userid，或methodname），key支持表达式形式，如#{id}, #{user.id}

针对某个时间窗口内限制访问一次的场景，既可以使用 @CountLimit， 也可以使用 @RateLimit，比如验证码一分钟内只允许获取一次，以下两种形式都能达到目的
//同一个手机号码60s内最多访问一次
@CountLimit(key = "#{params.phone}", limit = 1, period = 60, limitType = LimitType.CUSTOM)
//以1/60的速度放置令牌，最多保存60s的令牌（也就是最多保存一个），控制访问速度为1/60个每秒（1个每分钟）
@RateLimit(key = "#{params.phone}", rate = 1.0/60, burst = 60, expire = 120, limitType = LimitType.CUSTOM)

写在最后
本文介绍了适用于分布式环境的基于RateLimiter令牌桶算法的限速控制与基于计数器算法的限量控制，可应用于中小型项目中有相关需求的场景（注：本实现未做压力测试，如果用户并发量较大需验证效果）。
源代码地址，目录 spring-boot-autoconfigure/src/main/java/cn/jboost/springboot/autoconfig/limiter
如果觉得有用，别忘了给个star ，并关注我哦_

1.并行通信：是指数据的各位同时在多根数据线上发送或接收。如下图



并行通信的特点：控制简单，传输速度快；由于传输线较多，适用于短距离通信。

2.串行通信：是指数据的各位在同一根数据线上逐位发送和接收。如下图



串行通信的特点：控制复杂，传输速度慢；只需要一根数据线，适用于远距离通信。
串行通信方式分类

在串行通信中，根据对数据流的分界、定时以及同步方案方法不同，可分为和同步串行通信方式和异步通信方式。
同步通信

同步通信是指发送端和接收端必须使用同一时钟，是一种连续传送数据的通信方式，一次通讯传送多个字符数据（一帧数据）。下图是同步串行通信帧格式：



同步串行通信方式：把许多字符组成一个信息组（信息振），每帧的开始用字符来指示。并且发送和接收的双方必须采用同一时钟，这样接收方就可以通过时钟信号来确定每个信息位。（如下图所示）



同步串行通信帧：是将许多字符组成一个信息帧，字符可以一个接一个传输。但是，需要在每帧信息的开始加上同步字符，在没有信息要传输时，要填上空字符，因为同步传输不允许有间隙。
同步串行通信的特点：必须有同步时钟，传输信息量大，传输速率高，但是传输设备则为复杂，技术要求高。
异步通信

异步通信是指发送和接收端使用的是各自的时钟，并且它是一种不连续的传输通信方式，一次通信只能传输一个字符数据（字符帧）。字符帧之间的间隙可以是任意的，在STM32中USART就是使用的是异步通信方式，下图是异步串行通信帧格式：



异步串行通信方式：是指通信双方以一个自读（包括特定附加位）作为数据传输单位且发送方传送字符的间隔时间是不定的，具有不规则数据段传输特性，下图是异步串行通信：



异步串行同信帧：将一个字节数据加上起始位、校验位以及停止位构成字符帧。由于异步通信没有同步时钟，所以接收端要时刻处于接收状态。
起始位：在没有数据传送时（空闲状态），此时通信线上为逻辑“1”。当发送端要发送一个数据时，首先发送一个逻辑“0”，这个低电平就是帧格式的起始位。作用是告诉接收端要开始发送一帧数据。接收端检测到这个低电平之后，就准备接收数据信号。
数据位：在起始位之后，发送端发出的就是数据位，数据位的位数没有严格限制（5-8位都可以）。低位在前，高位在后。由低位向高位逐位发送。
校验位：数据位发送完成之后，可以发送以为用来校验数据在传送过程中是否出错。校验位是收发双方预先约定好的有限制差错检验的方式之一（可不用）。
停止位：字符帧格式的最后部分是停止位，逻辑“1”有效，它的占位有1/2位、1位或者2位。停止位表示传送一帧信息的结束，也作为发送下一帧数据信息做准备。
异步串行通信特点：不需要同步时钟，通信实现简单，设备简单。但是传输速率不高。
串行通信数据传送方向

根据串行数据的传输方向，我们可以将通信分为单工，半双工，双工。
单工：是指数据传输仅能沿一个方向，不能实现反向传输。
半双工：是指数据传输可以沿两个方向，但需要分时进行传输。
全双工：是指数据可以同时进行双向传输。
下图是单工、半双工以及全双工的示意图


串行通信传输速率

比特率：每秒钟传送的二进制位数。bps
波特率：每秒钟调制信号变化的次数。Baud
串行通信常用波特率表示数据传输率。
波特率与比特率的关系为：
比特率 = 波特率x单个调制状态对应的二进制位数
单个调制状态对应1个二进制位时，或对于数字信号1或0直接用两种不同电压表示的基带传输，此时 比特率==波特率。串行通信双方识别位的时间间隔要相同，所以通信双方的波特率必须一致。