快捷導(dǎo)航

JAVA流控及超流控后的延遲處理實(shí)例

更新時(shí)間：2014年12月31日 09:42:42 投稿：shichen2014

這篇文章主要介紹了JAVA流控及超流控后的延遲處理,以實(shí)例形式較為詳細(xì)的分析了Java進(jìn)行流量控制的技巧,具有一定參考借鑒價(jià)值,需要的朋友可以參考下

本文實(shí)例講述了JAVA流控及超流控后的延遲處理方法。分享給大家供大家參考。具體實(shí)現(xiàn)方法如下：

流控檢查（每半秒累計(jì)，因此最小留空閥值只能做到每秒2條）：

import java.text.SimpleDateFormat;

import java.util.Date;

import java.lang.Thread;

/**

 * 流量控制

 *

 * @author chenx

 */

public class OverflowController {

    private int maxSendCountPerSecend; // 該條鏈路上流控閥值

    private Date sendTime = new Date();

    private int sendCount = 0; // 該條鏈路上發(fā)送的數(shù)量

    public OverflowController(int maxSendCountPerSecend) {

        if (maxSendCountPerSecend < 2) {

            maxSendCountPerSecend = 2;

        }

        this.maxSendCountPerSecend = maxSendCountPerSecend;

    }

    public int getMaxSendCountPerSecend() {

        if (getMilliseconds(new Date()) >= 500) {

            return maxSendCountPerSecend / 2;

        }

        return maxSendCountPerSecend - (maxSendCountPerSecend / 2);

    }

    /**

     * 是否超流控

     */

    public boolean isOverflow(int sendNum) {

        synchronized (this) {

            Date now = new Date();

            if (now.getTime() - sendTime.getTime() >= 500) {

                sendTime = now;

                sendCount = sendNum;

            } else {

                if (sendCount + sendNum > getMaxSendCountPerSecend()) {

                    return true;

                } else {

                    sendCount += sendNum;

                }

            }

            return false;

        }

    }

    /**

     * 獲取指定時(shí)間的毫秒數(shù)

     */

    private int getMilliseconds(Date date) {

        SimpleDateFormat df = new SimpleDateFormat("SSS");

        return Integer.valueOf(df.format(date));

    }

    public static void main(String[] args) throws InterruptedException {

        OverflowController oc = new OverflowController(50);

        SimpleDateFormat df = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss:SSS");

        for (int i = 0; i <= 100; i++) {

            if (oc.isOverflow(1)) {

                System.out.println(i + "-isOverflow-" + df.format(new Date()));

            } else {

                System.out.println(i + "-sendOk-" + df.format(new Date()));

            }

            Thread.sleep(10);

        }

    }

}

超流控后的延遲處理，由于java中沒有.net的“延遲委托”一說：

復(fù)制代碼代碼如下:

ThreadPool.RegisterWaitForSingleObject(

 WaitHandle waitObject,

      WaitOrTimerCallback callBack,

      Object state,

     int millisecondsTimeOutInterval,

     bool executeOnlyOnce

)

Java下需實(shí)現(xiàn)一個(gè)簡(jiǎn)單的延遲隊(duì)列：

復(fù)制代碼代碼如下:

import java.util.concurrent.Delayed;

import java.util.concurrent.TimeUnit;

public class DelayEntry implements Delayed {

    private int count;

    private long dequeuedTimeMillis; // 出隊(duì)列時(shí)間

    public int getCount() {

        return count;

    }

    public void setCount(int count) {

        this.count = count;

    }

    public long getDequeuedTimeMillis() {

        return dequeuedTimeMillis;

    }

    public DelayEntry(long delayMillis) {

        dequeuedTimeMillis = System.currentTimeMillis() + delayMillis;

    }

    @Override

    public int compareTo(Delayed o) {

        DelayEntry de = (DelayEntry) o;

        long timeout = dequeuedTimeMillis - de.dequeuedTimeMillis;

        return timeout > 0 ? 1 : timeout < 0 ? -1 : 0;

    }

    @Override

    public long getDelay(TimeUnit unit) {

        return dequeuedTimeMillis - System.currentTimeMillis();

    }

}

復(fù)制代碼代碼如下:

import java.util.concurrent.DelayQueue;

public class DelayService {

    public void run() {

        DelayQueue<DelayEntry> queue = new DelayQueue<DelayEntry>();

        DelayConsumer delayConsumer = new DelayConsumer(queue);

        delayConsumer.start();

        for (int i = 0; i < 100; i++) {

            DelayEntry de = new DelayEntry(5000);

            de.setCount(i);

            System.out.println(System.currentTimeMillis() + "--------" + de.getCount());

            queue.add(de);

        }

    }

    class DelayConsumer extends Thread {

        DelayQueue<DelayEntry> queue;

        public DelayConsumer(DelayQueue<DelayEntry> queue) {

            this.queue = queue;

        }

        public void run() {

            while (true) {

                try {

                    DelayEntry de = queue.take();

                    System.out.println("queue size=" + queue.size());

                    System.out.println(de.getCount());

                    System.out.println(System.currentTimeMillis());

                } catch (InterruptedException e) {

                    e.printStackTrace();

                }

            }

        }

    }

    public static void main(String[] args) {

        DelayService ds = new DelayService();

        ds.run();

    }

}