安科网

coding++：Semaphore—RateLimiter-漏桶算法-令牌桶算法

Happyunlimited

Happyunlimited

2019-11-16

关注 关注

java中对于生产者消费者模型，或者小米手机营销 1分钟卖多少台手机等都存在限流的思想在里面。

关于限流 目前存在两大类，从线程个数（jdk1.5 Semaphore）和RateLimiter速率(guava)

Semaphore：从线程个数限流

RateLimiter：从速率限流  目前常见的算法是漏桶算法和令牌算法

令牌桶算法。相比漏桶算法而言区别在于，令牌桶是会去匀速的生成令牌，拿到令牌才能够进行处理，类似于匀速往桶里放令牌

漏桶算法是：生产者消费者模型，生产者往木桶里生产数据，消费者按照定义的速度去消费数据

应用场景：

漏桶算法：必须读写分流的情况下，限制读取的速度

令牌桶算法：必须读写分离的情况下，限制写的速率或者小米手机饥饿营销的场景  只卖1分种抢购1000实现的方法都是一样。

RateLimiter来实现对于多线程问题查找时，很多时候可能使用的类都是原子性的，但是由于代码逻辑的问题，也可能发生线程安全问题

1、关于RateLimter和Semphore简单用法

package concurrent;
 
import com.google.common.util.concurrent.RateLimiter;
 
import java.util.concurrent.*;
import java.util.stream.IntStream;
 
import static java.lang.Thread.currentThread;
 
/**
 * ${DESCRIPTION}
 * 关于限流 目前存在两大类，从线程个数（jdk1.5 Semaphore）和RateLimiter速率(guava)
 * Semaphore：从线程个数限流
 * RateLimiter：从速率限流  目前常见的算法是漏桶算法和令牌算法，下面会具体介绍
 *
 * @author mengxp
 * @version 1.0
 * @create 2018-01-15 22:44
 **/
public class RateLimiterExample {
 
   //Guava  0.5的意思是 1秒中0.5次的操作，2秒1次的操作  从速度来限流，从每秒中能够执行的次数来
    private final static RateLimiter limiter=RateLimiter.create(0.5d);
 
 
    //同时只能有三个线程工作 Java1.5  从同时处理的线程个数来限流
    private final static Semaphore sem=new Semaphore(3);
    private static void testSemaphore(){
        try {
            sem.acquire();
            System.out.println(currentThread().getName()+" is doing work...");
            TimeUnit.MILLISECONDS.sleep(ThreadLocalRandom.current().nextInt(10));
        } catch (InterruptedException e) {
            e.printStackTrace();
        }finally {
            sem.release();
            System.out.println(currentThread().getName()+" release the semephore..other thread can get and do job");
        }
    }
 
    public static void runTestSemaphore(){
        ExecutorService service = Executors.newFixedThreadPool(10);
        IntStream.range(0,10).forEach((i)->{
            //RateLimiterExample::testLimiter 这种写法是创建一个线程
            service.submit(RateLimiterExample::testSemaphore);
        });
    }
 
    /**
     * Guava的RateLimiter
     */
    private static void testLimiter(){
        System.out.println(currentThread().getName()+" waiting  " +limiter.acquire());
    }
 
    //Guava的RateLimiter
    public static void runTestLimiter(){
        ExecutorService service = Executors.newFixedThreadPool(10);
        IntStream.range(0,10).forEach((i)->{
            //RateLimiterExample::testLimiter 这种写法是创建一个线程
            service.submit(RateLimiterExample::testLimiter);
        });
    }
 
 
 
    public static void main(String[] args) {
        IntStream.range(0,10).forEach((a)-> System.out.println(a));//从0-9
        //runTestLimiter();
        runTestSemaphore();
    }
}

2、实现漏桶算法

package concurrent.BucketAl;
 
import com.google.common.util.concurrent.Monitor;
import com.google.common.util.concurrent.RateLimiter;
 
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.TimeUnit;
import java.util.function.Consumer;
 
import static java.lang.Thread.currentThread;
 
/**
 * ${DESCRIPTION}
 *
 * @author mengxp
 * @version 1.0
 * @create 2018-01-20 22:42
 * 实现漏桶算法 实现多线程生产者消费者模型 限流
 **/
public class Bucket {
    //定义桶的大小
    private final ConcurrentLinkedQueue<Integer> container=new ConcurrentLinkedQueue<>();
 
    private final static int  BUCKET_LIMIT=1000;
 
    //消费者 不论多少个线程，每秒最大的处理能力是1秒中执行10次
    private final RateLimiter consumerRate=RateLimiter.create(10d);
 
    //往桶里面放数据时，确认没有超过桶的最大的容量
    private Monitor offerMonitor=new Monitor();
 
    //从桶里消费数据时，桶里必须存在数据
    private Monitor consumerMonitor=new Monitor();
 
 
    /**
     * 往桶里面写数据
     * @param data
     */
    public void submit(Integer data){
        if (offerMonitor.enterIf(offerMonitor.newGuard(()->container.size()<BUCKET_LIMIT))){
            try {
                container.offer(data);
                System.out.println(currentThread()+" submit.."+data+" container size is :["+container.size()+"]");
            } finally {
                offerMonitor.leave();
            }
        }else {
            //这里时候采用降级策略了。消费速度跟不上产生速度时，而且桶满了，抛出异常
            //或者存入MQ DB等后续处理
            throw new IllegalStateException(currentThread().getName()+"The bucket is ful..Pls latter can try...");
        }
    }
 
 
    /**
     * 从桶里面消费数据
     * @param consumer
     */
    public void takeThenConsumer(Consumer<Integer> consumer){
        if (consumerMonitor.enterIf(consumerMonitor.newGuard(()->!container.isEmpty()))){
            try {
                //不打印时 写 consumerRate.acquire();
                System.out.println(currentThread()+"  waiting"+consumerRate.acquire());
                Integer data = container.poll();
                //container.peek() 只是去取出来不会删掉
                consumer.accept(data);
            }finally {
                consumerMonitor.leave();
            }
        }else {
            //当木桶的消费完后，可以消费那些降级存入MQ或者DB里面的数据
            System.out.println("will consumer Data from MQ...");
            try {
                TimeUnit.SECONDS.sleep(10);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }
 
}

2.1 漏桶算法测试类

package concurrent.BucketAl;
 
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.IntStream;
 
import static java.lang.Thread.currentThread;
 
/**
 * ${DESCRIPTION}
 *
 * @author mengxp
 * @version 1.0
 * @create 2018-01-20 23:11
 * 漏桶算法测试
 * 实现漏桶算法 实现多线程生产者消费者模型 限流
 **/
public class BuckerTest {
 
    public static void main(String[] args) {
        final Bucket bucket = new Bucket();
        final AtomicInteger DATA_CREATOR = new AtomicInteger(0);
 
        //生产线程 10个线程 每秒提交 50个数据  1/0.2s*10=50个
        IntStream.range(0, 10).forEach(i -> {
            new Thread(() -> {
                for (; ; ) {
                    int data = DATA_CREATOR.incrementAndGet();
                    try {
                        bucket.submit(data);
                        TimeUnit.MILLISECONDS.sleep(200);
                    } catch (Exception e) {
                        //对submit时，如果桶满了可能会抛出异常
                        if (e instanceof IllegalStateException) {
                            System.out.println(e.getMessage());
                            //当满了后，生产线程就休眠1分钟
                            try {
                                TimeUnit.SECONDS.sleep(60);
                            } catch (InterruptedException e1) {
                                e1.printStackTrace();
                            }
                        }
                    }
                }
            }).start();
        });
 
 
        //消费线程  采用RateLimiter每秒处理10个  综合的比率是5:1
        IntStream.range(0, 10).forEach(i -> {
            new Thread(
                    () -> {
                        for (; ; ) {
                            bucket.takeThenConsumer(x -> {
                                System.out.println(currentThread()+"C.." + x);
                            });
                        }
                    }
            ).start();
        });
 
    }
}

3、令牌桶算法

package concurrent.TokenBucket;
 
import com.google.common.util.concurrent.RateLimiter;
 
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
 
import static java.lang.Thread.currentThread;
import static java.lang.Thread.interrupted;
 
/**
 * ${DESCRIPTION}
 *
 * @author mengxp
 * @version 1.0
 * @create 2018-01-21 0:18
 * 令牌桶算法。相比漏桶算法而言区别在于，令牌桶是会去匀速的生成令牌，拿到令牌才能够进行处理，类似于匀速往桶里放令牌
 * 漏桶算法是：生产者消费者模型，生产者往木桶里生产数据，消费者按照定义的速度去消费数据
 *
 * 应用场景：
 * 漏桶算法：必须读写分流的情况下，限制读取的速度
 * 令牌桶算法：必须读写分离的情况下，限制写的速率或者小米手机饥饿营销的场景  只卖1分种抢购1000
 *
 * 实现的方法都是一样。RateLimiter来实现
 * 对于多线程问题查找时，很多时候可能使用的类都是原子性的，但是由于代码逻辑的问题，也可能发生线程安全问题
 **/
public class TokenBuck {
 
    //可以使用 AtomicInteger+容量  可以不用Queue实现
   private AtomicInteger phoneNumbers=new AtomicInteger(0);
   private RateLimiter rateLimiter=RateLimiter.create(20d);//一秒只能执行五次
   //默认销售500台
   private final static int DEFALUT_LIMIT=500;
   private final int saleLimit;
 
    public TokenBuck(int saleLimit) {
        this.saleLimit = saleLimit;
    }
 
    public TokenBuck() {
        this(DEFALUT_LIMIT);
    }
 
    public int buy(){
        //这个check 必须放在success里面做判断，不然会产生线程安全问题(业务引起)
        //原因当phoneNumbers=99 时 同时存在三个线程进来。虽然phoneNumbers原子性，但是也会发生。如果必须写在这里，在success
        //里面也需要加上double check
       /* if (phoneNumbers.get()>=saleLimit){
            throw new IllegalStateException("Phone has been sale "+saleLimit+" can not  buy more...")
        }*/
 
        //目前设置超时时间，10秒内没有抢到就抛出异常
        //这里的TimeOut*Ratelimiter=总数  这里的超时就是让别人抢几秒，所以设置总数也可以由这里的超时和RateLimiter来计算
         boolean success = rateLimiter.tryAcquire(10, TimeUnit.SECONDS);
         if (success){
             if (phoneNumbers.get()>=saleLimit){
                 throw new IllegalStateException("Phone has been sale "+saleLimit+" can not  buy more...");
             }
             int phoneNo = phoneNumbers.getAndIncrement();
             System.out.println(currentThread()+" user has get :["+phoneNo+"]");
             return phoneNo;
         }else {
             //超时后 同一时间，很大的流量来强时，超时快速失败。
             throw new RuntimeException(currentThread()+"has timeOut can try again...");
         }
 
    }
}

3.1、令牌桶算法的测试类

package concurrent.TokenBucket;
 
import java.util.stream.IntStream;
 
/**
 * ${DESCRIPTION}
 *
 * @author mengxp
 * @version 1.0
 * @create 2018-01-21 0:40
 **/
public class TokenBuckTest {
    public static void main(String[] args) {
        final TokenBuck tokenBuck=new TokenBuck(200);
 
 
        IntStream.range(0,300).forEach(i->{
            //目前测试时，让一个线程抢一次，不用循环抢
            //tokenBuck::buy 这种方式 产生一个Runnable
            new Thread(tokenBuck::buy).start();
        });
    }
}

semaphore 令牌桶算法 coding 线程 算法

Happyunlimited

Happyunlimited

0 关注 0 粉丝 0 动态

关注 关注

相关推荐

深究 Linux 多线程中的信号量 Semaphore

Semaphore，对多线程有过了解的人都听说过，一般我们解释为“信号量”。可是，这个单词对我们来说还是比较陌生，它和另一个单词 Singal(信号)什么关系呢?想要真正理解这个概念，必须得从它的翻译开始。事实上，Semaphore 最好的翻译应该为“

starinshy 2020-11-10

Linux内核源码分析 -- 同步原语 -- 信号量 semaphore

LIST_HEAD_INIT 返回 NULL， 把 等待获取这个锁的进程队列 初始化为链表头，指向 NULL. lockdep_init_map; // 锁验证 这里不用管。down_interruptible 函数：试图去获取一个 信号量。如果被成功获

神龙 2020-06-07

【分布式锁】05-使用Redisson中Semaphore和CountDownLatch原理

剩下还有Semaphore和CountDownLatch两块，我们就趁热打铁，赶紧看看Redisson是如何实现的吧。我们在JDK中都知道Semaphore和CountDownLatch两兄弟，这里就不多赘述，不了解的可以再回头看看。第一次数据为0， 然后

sunzxh 2020-03-23

python,信号量,semaphore

python中的信号量,是通过定义semaphore对象,控制同时可以运行的线程的数量,同时也是一种锁,下面的代码演示了信号量的应用。t.start()这段代码运行后,通过semaphore实现了同时可以允许5个线程同步运行,每隔3秒显示5个‘Alex‘

laityc 2019-12-15

python线程信号量semaphore使用解析

acquire() ― 消耗信号量，内置计数器减一;根据打印的日志可以看出，同一时间只有5个线程运行，间隔两秒之后，再次启动5个线程，直到20个线程全部运行结束为止；如果没有设置信号量Semapaore,创建线程直接start()，输出的时间全部都是一样的

hahalale 2019-11-30

python线程信号量semaphore（33）

通过前面对线程互斥锁lock/线程事件event/线程条件变量condition/线程定时器timer的讲解，相信你对线程threading模块已经有了一定的了解，同时执行多个线程的确可以提高程序的效率，但是并非线程的数量越多越好，可能对于计算机而言，你直

wordmhg 2019-11-30

一文看懂 Mutex vs Semaphore vs Monitor vs SemaphoreSlim

C#开发者（面试者）都会遇到Mutex，Semaphore，Monitor，SemaphoreSlim这四个与锁相关的C#类型，本文期望以最简洁明了的方式阐述四种对象的区别。什么叫线程安全？如果代码在多线程环境中运行的结果与 单线程运行结果一样，其他变量值

gongruitao 2019-10-30

Linux设备驱动程序 之 信号量和互斥体

一个信号量本质是一个整数值，它和一堆函数联合使用，这一对函数通常称为P和V；希望进入临界区的进程将在相关信号量上调用P；如果信号量的值大于零，则该值会减少1，进程可以继续执行；相反，如果信号量的值为0或者更小，则进程必须等待知道其他人释放该信号量；对信号量

GeorgeTH 2019-10-29

Linux内核的并发控制

现代Linux系统中存在大量的并发来源，导致可能的竞态，竞态通常作为对资源的共享访问结果而产生。访问管理的常见技术称为“锁定”或者“互斥”——确保一次只有一个执行单元可操作共享资源。在Linux内核中，主要的竞态发生在如下几种情况：对称多处理器的多个CPU

playlinuxxx 2016-07-30

Semaphore源码分析

Semaphore有两种模式，公平模式和非公平模式。公平模式就是调用acquire的顺序就是获取许可证的顺序，遵循FIFO；而非公平模式是抢占式的，也就是有可能一个新的获取线程恰好在一个许可证释放时得到了这个许可证，而前面还有等待的线程。Semaphore

zhaojiacan 2018-07-12

线程间的同步与通信(8)——Semaphore源码分析

Semaphore（信号量）也是常用的并发工具之一，它常常用于流量控制。通常情况下，公共的资源常常是有限的，例如数据库的连接数。使用Semaphore可以帮助我们有效的管理这些有限资源的使用。Semaphore的结构和ReentrantLock以及Coun

mojianc 2019-06-28

python lock, semaphore, event实现线程同步

lock 机制不管你是java, C#, 还是Python都是常用的线程同步机制, 相比较C# 的锁机制, python的加锁显得比较简单, 直接调用threading 标准库的lock 就可以了. python 的 lock类有两个函数, 分别是acqu

peterlee 2017-09-29

Linux下利用条件变量实现信号量机制

接上一篇《Linux下利用条件变量实现读写锁》，继续讨论如何利用pthread_cond_t来实现sem_t。目前的Linux内核都支持信号量sem_t，但也有一些老的OS，如AIX4，和早期的Solaris并不支持信号量，毕竟semaphore这个东东并

Will0 2011-04-10

Python Library中的Semaphore操作方案详解

PythonLibrary作为计算机语言中常用的语言，很多人都会接触到Python Library中的Semaphore这一计算机语言，一下的文章就是对其实际应用操作方案的具体介绍，希望你浏览完以下的文章会对你有所帮助。Python Library Sem

honghao0 2010-03-17

Python多线程同步Lock、RLock、Semaphore、Event实例

由于CPython的python解释器在单线程模式下执行，所以导致python的多线程在很多的时候并不能很好地发挥多核cpu的资源。大部分情况都推荐使用多进程。Lock & RLock ：用来确保多线程多共享资源的访问。使用acquire()设置为

pythoncream 2019-04-17

Python多进程同步Lock、Semaphore、Event实例

同步的方法基本与多线程相同。当多个进程需要访问共享资源的时候，Lock可以用来避免访问的冲突。nw.join()在上面的例子中，如果两个进程没有使用lock来同步,则他们对同一个文件的写操作可能会出现混乱。Semaphore用来控制对共享资源的访问数量，例

ytp00ytp 2019-04-16

Python 限制线程的最大数量的方法(Semaphore)

with sem: #锁定线程的最大数量

wlpython 2019-02-22

JUC——线程同步辅助工具类（Semaphore，CountDownLatch，CyclicBarrier）

锁的机制从整体的运行转态来讲核心就是：阻塞，解除阻塞，但是如果仅仅是这点功能，那么JUC并不能称为一个优秀的线程开发框架，然而是因为在juc里面提供了大量方便的同步工具辅助类。Semaphore信号量Semaphore通常用于限制可以访问某些资源的线程数目

稀土 2018-05-11

Semaphore信号量

Semaphore是计数信号量。Semaphore管理一系列许可证。每个acquire方法阻塞，直到有一个许可证可以获得然后拿走一个许可证；每个release方法增加一个许可证，这可能会释放一个阻塞的acquire方法。然而，其实并没有实际的许可证这个对象

友心人 2018-05-07

java.util.concurrent常用类（CountDownLatch，Semaphore，CyclicBarrier，F

常用于监听某些初始化操作，等待初始化执行完毕后，通知主线程继续工作。

稀土 2018-02-12
Happyunlimited

Happyunlimited

W3CSchool教程
HTML 教程
CSS 教程
Bootstrap 教程
Javascript 教程
jQuery 教程
后端教程
C 教程
Java 教程
PHP 教程
Python 教程
Go 教程
移动开发
Android 教程
Swift 教程
Kotlin 教程
jQuery Mobile 教程
ionic 教程
关于我们
新闻动态
联系方式
招聘英才
安科实验室
帮助与反馈

安科网(Ancii)，中国第一极客网

Copyright © 2013 - 2019 Ancii.com

京ICP备18063983号 京公网安备11010802014868号