Bagaimana untuk melaksanakan algoritma pengehadan semasa biasa di Java

Mengapa mengehadkan aliran?

Tingkatkan bilangan orang yang masuk sebanyak mungkin sambil memastikan ketersediaan orang lain sedang menunggu dalam barisan, atau kembalikan gesaan mesra untuk memastikan pengguna sistem boleh gunakannya seperti biasa.

Algoritma pengehad semasa

Terdapat banyak algoritma pengehadan semasa, dan terdapat tiga kategori biasa, iaitu algoritma pembilang, algoritma baldi bocor dan algoritma baldi token.

(1) Kaunter:

Dalam tempoh masa, bilangan maksimum permintaan untuk diproses ditetapkan, dan lebihan tidak akan diproses.

(2) Baldi bocor:

Saiz baldi bocor ditetapkan dan kelajuan pemprosesan ditetapkan, tetapi kelajuan kemasukan permintaan tidak tetap (apabila terdapat terlalu banyak permintaan semasa kecemasan , terlalu banyak permintaan akan dibuang).

(3) Baldi token:

Saiz baldi token ditetapkan, kelajuan penjanaan token ditetapkan, tetapi kelajuan penggunaan token (iaitu permintaan) tidak tetap (ia boleh mengatasi dengan beberapa permintaan masa Terlalu banyak); setiap permintaan akan mengambil token daripada baldi token, dan jika tiada token, permintaan itu akan dibuang.

Had semasa balas

Dalam satu tempoh masa, bilangan maksimum permintaan untuk diproses ditetapkan dan sebarang lebihan permintaan tidak akan diproses.

Sebagai contoh, kami menetapkan bahawa untuk antara muka A, kami tidak boleh mengakses lebih daripada 100 kali dalam 1 minit.

Kemudian kita boleh melakukan ini:

Pada mulanya, kita boleh menetapkan kaunter kaunter Setiap kali permintaan datang, kaunter akan meningkat sebanyak 1. Jika nilai kaunter lebih daripada 100 Dan selang antara permintaan ini dan permintaan pertama masih dalam masa 1 minit, maka ini bermakna terdapat terlalu banyak permintaan dan akses ditolak

Jika selang antara permintaan ini dan permintaan pertama lebih daripada 1 minit, dan Nilai pembilang masih dalam julat had semasa, kemudian tetapkan semula pembilang, ia adalah semudah dan kasar seperti itu.

Pelaksanaan kod:

import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;

//计数器 限流
public class CounterLimiter {

    //起始时间
    private static long startTime = System.currentTimeMillis();

    //时间间隔1000ms
    private static long interval = 1000;

    //每个时间间隔内，限制数量
    private static long limit = 3;

    //累加器
    private static AtomicLong accumulator = new AtomicLong();

    /**
     * true 代表放行，请求可已通过
     * false 代表限制，不让请求通过
     */
    public static boolean tryAcquire() {
        long nowTime = System.currentTimeMillis();
        //判断是否在上一个时间间隔内
        if (nowTime < startTime + interval) {
            //如果还在上个时间间隔内
            long count = accumulator.incrementAndGet();
            if (count <= limit) {
                return true;
            } else {
                return false;
            }
        } else {
            //如果不在上一个时间间隔内
            synchronized (CounterLimiter.class) {
                //防止重复初始化
                if (nowTime > startTime + interval) {
                    startTime = nowTime;
                    accumulator.set(0);
                }
            }
            //再次进行判断
            long count = accumulator.incrementAndGet();
            if (count <= limit) {
                return true;
            } else {
                return false;
            }
        }
    }


    // 测试
    public static void main(String[] args) {

        //线程池，用于多线程模拟测试
        ExecutorService pool = Executors.newFixedThreadPool(10);
        // 被限制的次数
        AtomicInteger limited = new AtomicInteger(0);
        // 线程数
        final int threads = 2;
        // 每条线程的执行轮数
        final int turns = 20;
        // 同步器
        CountDownLatch countDownLatch = new CountDownLatch(threads);
        long start = System.currentTimeMillis();
        for (int i = 0; i < threads; i++) {
            pool.submit(() ->
            {
                try {

                    for (int j = 0; j < turns; j++) {

                        boolean flag = tryAcquire();
                        if (!flag) {
                            // 被限制的次数累积
                            limited.getAndIncrement();
                        }
                        Thread.sleep(200);
                    }

                } catch (Exception e) {
                    e.printStackTrace();
                }
                //等待所有线程结束
                countDownLatch.countDown();
            });
        }
        try {
            countDownLatch.await();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        float time = (System.currentTimeMillis() - start) / 1000F;
        //输出统计结果
        System.out.println("限制的次数为：" + limited.get() +
                ",通过的次数为：" + (threads * turns - limited.get()));
        System.out.println("限制的比例为：" + (float) limited.get() / (float) (threads * turns));
        System.out.println("运行的时长为：" + time + "s");
    }

}

Ketidakcukupan had arus kaunter:

Walaupun algoritma ini mudah, terdapat isu kritikal lihat. Gambar:

Daripada gambar di atas, kita dapat melihat bahawa dengan mengandaikan terdapat pengguna yang berniat jahat, dia menghantar 100 permintaan serta-merta pada 0:59, dan pada 1: 00 100 permintaan lagi dihantar serta-merta, jadi sebenarnya, pengguna ini menghantar 200 permintaan serta-merta dalam masa 1 saat.

Apa yang baru kami tetapkan ialah maksimum 100 permintaan seminit (proses pengeluaran yang dirancang), iaitu maksimum 1.7 permintaan sesaat Dengan membuat permintaan pecah pada nod tetapan tetingkap masa, pengguna boleh Melebihi serta-merta had kadar kami.

Pengguna boleh serta-merta mengatasi aplikasi kami melalui kelemahan dalam algoritma ini.

Penghad arus baldi bocor

Prinsip asas pengehadan arus baldi bocor adalah: air (bersesuaian dengan permintaan) memasuki baldi bocor dari salur masuk air, dan baldi bocor mengeluarkan air pada kelajuan tertentu (meminta pelepasan) ), apabila kelajuan aliran masuk air terlalu tinggi dan jumlah isipadu air dalam baldi lebih besar daripada kapasiti baldi, ia akan terus melimpah dan permintaan akan ditolak.

Anggaran peraturan mengehadkan arus baldi bocor adalah seperti berikut:

(1) Saluran masuk air (bersesuaian dengan permintaan pelanggan) mengalir ke dalam baldi bocor pada sebarang kadar.

(2) Kapasiti baldi bocor tetap, dan kadar keluar air (pelepasan) juga tetap.

(3) Kapasiti baldi bocor tidak berubah Jika kelajuan pemprosesan terlalu perlahan, jumlah air dalam baldi akan melebihi kapasiti baldi, dan titisan air yang mengalir kemudian akan melimpah. , menunjukkan bahawa permintaan itu ditolak.

⭐Algoritma baldi bocor sebenarnya sangat mudah ia boleh dianggap sebagai proses kebocoran air pada sebarang kadar dan mengalir keluar kadar tertentu Apabila air melebihi Kapasiti baldi (kapasiti) dibuang, kerana kapasiti baldi tidak berubah, memastikan kadar keseluruhan.

Air mengalir keluar pada kadar tertentu,

Keratan puncak: Apabila sejumlah besar lalu lintas masuk, limpahan akan berlaku, jadi perlindungan pengehad semasa perkhidmatan tersedia

Penimbalan: Jangan minta terus ke pelayan, tekanan penimbal

Pelaksanaan kod:

import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;

//漏斗限流
public class LeakBucketLimiter {

    //桶的大小
    private static long capacity = 10;
    //流出速率,每秒两个
    private static long rate = 2;
    //开始时间
    private static long startTime = System.currentTimeMillis();
    //桶中剩余的水
    private static AtomicLong water = new AtomicLong();

    /**
     * true 代表放行，请求可已通过
     * false 代表限制，不让请求通过
     */
    public synchronized static boolean tryAcquire() {
        //如果桶的余量问0，直接放行
        if (water.get() == 0) {
            startTime = System.currentTimeMillis();
            water.set(1);
            return true;
        }
        //计算从当前时间到开始时间流出的水，和现在桶中剩余的水
        //桶中剩余的水
        water.set(water.get() - (System.currentTimeMillis() - startTime) / 1000 * rate);
        //防止出现<0的情况
        water.set(Math.max(0, water.get()));
        //设置新的开始时间
        startTime += (System.currentTimeMillis() - startTime) / 1000 * 1000;
        //如果当前水小于容量，表示可以放行
        if (water.get() < capacity) {
            water.incrementAndGet();
            return true;
        } else {
            return false;
        }
    }


    // 测试
    public static void main(String[] args) {

        //线程池，用于多线程模拟测试
        ExecutorService pool = Executors.newFixedThreadPool(10);
        // 被限制的次数
        AtomicInteger limited = new AtomicInteger(0);
        // 线程数
        final int threads = 2;
        // 每条线程的执行轮数
        final int turns = 20;
        // 同步器
        CountDownLatch countDownLatch = new CountDownLatch(threads);
        long start = System.currentTimeMillis();
        for (int i = 0; i < threads; i++) {
            pool.submit(() ->
            {
                try {

                    for (int j = 0; j < turns; j++) {

                        boolean flag = tryAcquire();
                        if (!flag) {
                            // 被限制的次数累积
                            limited.getAndIncrement();
                        }
                        Thread.sleep(200);
                    }

                } catch (Exception e) {
                    e.printStackTrace();
                }
                //等待所有线程结束
                countDownLatch.countDown();
            });
        }
        try {
            countDownLatch.await();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        float time = (System.currentTimeMillis() - start) / 1000F;
        //输出统计结果
        System.out.println("限制的次数为：" + limited.get() +
                ",通过的次数为：" + (threads * turns - limited.get()));
        System.out.println("限制的比例为：" + (float) limited.get() / (float) (threads * turns));
        System.out.println("运行的时长为：" + time + "s");
    }

}

Kelemahan baldi bocor:

The kelajuan keluaran air baldi bocor adalah tetap, iaitu permintaan Kelajuan pelepasan ditetapkan.

Salur keluar baldi yang bocor mempunyai kelajuan tetap dan tidak boleh secara fleksibel mengatasi peningkatan keupayaan bahagian belakang. Contohnya, melalui pengembangan dinamik, trafik bahagian belakang ditingkatkan daripada 1000QPS kepada 1WQPS, dan tiada penyelesaian untuk baldi bocor.

Had semasa baldi token

Dalam algoritma baldi token, apabila permintaan baharu tiba, token akan diambil daripada baldi Jika tiada token dalam baldi, perkhidmatan akan diberikan dinafikan. Sudah tentu, bilangan token juga terhad. Bilangan token sangat berkaitan dengan masa dan kadar pengeluaran Semakin lama masa berlalu, semakin banyak token akan ditambahkan pada baldi Jika kelajuan pengeluaran token lebih cepat daripada kelajuan permohonan, baldi token akan diisi dengan token. sehingga token menduduki keseluruhan baldi token.

Peraturan am untuk mengehadkan arus baldi token adalah seperti berikut:

(1) Salur masuk air memasukkan token ke dalam baldi pada kelajuan tertentu.

（2）令牌的容量是固定的，但是放行的速度不是固定的，只要桶中还有剩余令牌，一旦请求过来就能申请成功，然后放行。

（3）如果令牌的发放速度，慢于请求到来速度，桶内就无牌可领，请求就会被拒绝。

总之，令牌的发送速率可以设置，从而可以对突发的出口流量进行有效的应对。

代码实现： 

import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;

//令牌桶
public class TokenBucketLimiter {
    //桶的容量
    private static long capacity = 10;
    //放入令牌的速率,每秒2个
    private static long rate = 2;
    //上次放置令牌的时间
    private static long lastTime = System.currentTimeMillis();
    //桶中令牌的余量
    private static AtomicLong tokenNum = new AtomicLong();

    /**
     * true 代表放行，请求可已通过
     * false 代表限制，不让请求通过
     */
    public synchronized static boolean tryAcquire() {
        //更新桶中剩余令牌的数量
        long now = System.currentTimeMillis();
        tokenNum.addAndGet((now - lastTime) / 1000 * rate);
        tokenNum.set(Math.min(capacity, tokenNum.get()));
        //更新时间
        lastTime += (now - lastTime) / 1000 * 1000;
        //桶中还有令牌就放行
        if (tokenNum.get() > 0) {
            tokenNum.decrementAndGet();
            return true;
        } else {
            return false;
        }
    }


    //测试
    public static void main(String[] args) {

        //线程池，用于多线程模拟测试
        ExecutorService pool = Executors.newFixedThreadPool(10);
        // 被限制的次数
        AtomicInteger limited = new AtomicInteger(0);
        // 线程数
        final int threads = 2;
        // 每条线程的执行轮数
        final int turns = 20;
        // 同步器
        CountDownLatch countDownLatch = new CountDownLatch(threads);
        long start = System.currentTimeMillis();
        for (int i = 0; i < threads; i++) {
            pool.submit(() ->
            {
                try {

                    for (int j = 0; j < turns; j++) {

                        boolean flag = tryAcquire();
                        if (!flag) {
                            // 被限制的次数累积
                            limited.getAndIncrement();
                        }
                        Thread.sleep(200);
                    }

                } catch (Exception e) {
                    e.printStackTrace();
                }
                //等待所有线程结束
                countDownLatch.countDown();
            });
        }
        try {
            countDownLatch.await();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        float time = (System.currentTimeMillis() - start) / 1000F;
        //输出统计结果
        System.out.println("限制的次数为：" + limited.get() +
                ",通过的次数为：" + (threads * turns - limited.get()));
        System.out.println("限制的比例为：" + (float) limited.get() / (float) (threads * turns));
        System.out.println("运行的时长为：" + time + "s");
    }

}

令牌桶的好处： 

令牌桶的好处之一就是可以方便地应对 突发出口流量（后端能力的提升）。

比如，可以改变令牌的发放速度，算法能按照新的发送速率调大令牌的发放数量，使得出口突发流量能被处理。

Atas ialah kandungan terperinci Bagaimana untuk melaksanakan algoritma pengehadan semasa biasa di Java. Untuk maklumat lanjut, sila ikut artikel berkaitan lain di laman web China PHP!