基于Atomic实现的流量控制管理器

mex

[Java]代码

1. package flowcontrol;
   
2. 
   
3. import java.util.concurrent.atomic.AtomicInteger;
   
4. 
   
5. public class AtomicFlowCounter {
   
6. 
   
7.     private final int maxflow;
   
8.     private AtomicInteger currentflow;
   
9.     private final String name;
   
10. 
    
11.     public AtomicFlowCounter(String name) {
    
12.         this(-1, name);
    
13.     }
    
14. 
    
15.     public AtomicFlowCounter(int maxflow, String name) {
    
16.         this.maxflow = maxflow;
    
17.         currentflow = new AtomicInteger(0);
    
18.         this.name = name;
    
19.     }
    
20. 
    
21.     public int getMaxflow() {
    
22.         return maxflow;
    
23.     }
    
24. 
    
25.     public int getCurrentflow() {
    
26.         return currentflow.get();
    
27.     }
    
28. 
    
29.     public void setCurrentflow(int currentflow) {
    
30.         this.currentflow = new AtomicInteger(currentflow);
    
31.     }
    
32. 
    
33.     public String getName() {
    
34.         return name;
    
35.     }
    
36. 
    
37.     public boolean incCounter() {
    
38.         for (;;) {
    
39.             int oldvalue = currentflow.get();
    
40.             int current = oldvalue + 1;
    
41.             if (current > maxflow) {
    
42.                 return false;
    
43.             }
    
44.             if (currentflow.compareAndSet(oldvalue, current)) {
    
45.                 System.out.println(Thread.currentThread() + "加流量"+current);
    
46.                 return true;
    
47.             }
    
48.         }
    
49. 
    
50. //       if (currentflow.getAndIncrement() < maxflow) {
    
51. //       System.out.println(Thread.currentThread() + "加流量");
    
52. //       return true;
    
53. //       } else {
    
54. //       currentflow.decrementAndGet();
    
55. //       }
    
56. //       return false;
    
57.     }
    
58. 
    
59.     public boolean decCounter() {
    
60.         for (;;) {
    
61.             int oldvalue = currentflow.get();
    
62.             int current = oldvalue - 1;
    
63.             if (current < 0) {
    
64.                 return false;
    
65.             }
    
66.             if (currentflow.compareAndSet(oldvalue, current)) {
    
67.                 System.out.println(Thread.currentThread() + "减流量"+current);
    
68.                 return true;
    
69.             }
    
70.         }
    
71.         // if (currentflow.getAndDecrement() > 0) {
    
72.         // System.out.println(Thread.currentThread() + "减流量");
    
73.         // return true;
    
74.         // } else {
    
75.         // currentflow.incrementAndGet();
    
76.         // }
    
77.         // return false;
    
78.     }
    
79. }

复制代码

[Java]代码

1. package flowcontrol;
   
2. 
   
3. import java.util.Random;
   
4. import java.util.concurrent.ConcurrentHashMap;
   
5. import java.util.concurrent.ConcurrentMap;
   
6. import java.util.concurrent.CountDownLatch;
   
7. import java.util.concurrent.ExecutorService;
   
8. import java.util.concurrent.Executors;
   
9. import java.util.concurrent.ThreadFactory;
   
10. import java.util.concurrent.TimeUnit;
    
11. import java.util.concurrent.atomic.AtomicInteger;
    
12. import java.util.concurrent.atomic.AtomicLong;
    
13. import java.util.concurrent.locks.Condition;
    
14. import java.util.concurrent.locks.ReentrantLock;
    
15. 
    
16. public class AtomicFlowControlManager {
    
17. 
    
18.     private static final ConcurrentMap<String, AtomicFlowCounter> flowMap = new ConcurrentHashMap<String, AtomicFlowCounter>();
    
19.     private static final ReentrantLock lock = new ReentrantLock();
    
20.     private static final Condition condition = lock.newCondition();
    
21.     private static final AtomicInteger count = new AtomicInteger(0);
    
22. 
    
23.     public static boolean isExceed(String name) {
    
24.         AtomicFlowCounter counter = (AtomicFlowCounter) flowMap.get(name);
    
25.         if (counter != null) {
    
26.             return !counter.incCounter();
    
27.         }
    
28.         return true;
    
29.     }
    
30. 
    
31.     public static boolean releaseCounter(String name) {
    
32.         AtomicFlowCounter counter = (AtomicFlowCounter) flowMap.get(name);
    
33.         if (counter != null) {
    
34.             return counter.decCounter();
    
35.         }
    
36.         return false;
    
37.     }
    
38. 
    
39.     public static int getMaxFlow(String name) {
    
40.         AtomicFlowCounter counter = (AtomicFlowCounter) flowMap.get(name);
    
41.         if (counter != null) {
    
42.             return counter.getMaxflow();
    
43.         }
    
44.         return 0;
    
45.     }
    
46. 
    
47.     public static void setMaxflow(String name, int maxflow) {
    
48.         AtomicFlowCounter counter = null;
    
49.         lock.lock();
    
50.         try {
    
51.             counter = (AtomicFlowCounter) flowMap.get(name);
    
52.             if (counter == null) {
    
53.                 counter = new AtomicFlowCounter(maxflow, name);
    
54.                 flowMap.put(name, counter);
    
55.             }
    
56.         } finally {
    
57.             lock.unlock();
    
58.         }
    
59.     }
    
60. 
    
61.     public static void main(String[] args) {
    
62.         AtomicFlowControlManager.setMaxflow("b2b", 50);
    
63.         ExecutorService cacheThreadPool = Executors.newFixedThreadPool(100, new ThreadFactory() {
    
64. 
    
65.             private ThreadGroup threadGroup = System.getSecurityManager() == null ? Thread.currentThread()
    
66.                     .getThreadGroup() : System.getSecurityManager().getThreadGroup();
    
67.             private AtomicLong seq = new AtomicLong(0);
    
68.             private String THREADNAME = "AtomicFlowControlManager-Thread-";
    
69. 
    
70.             @Override
    
71.             public Thread newThread(Runnable r) {
    
72. 
    
73.                 Thread threadAdapter = new Thread(threadGroup, r, THREADNAME + seq.getAndIncrement(), 0);
    
74.                 return threadAdapter;
    
75.             }
    
76. 
    
77.         });
    
78.         long t1 = System.currentTimeMillis();
    
79.         final CountDownLatch cdl = new CountDownLatch(1000);
    
80.         for (int i = 0; i < 1000; i++) {
    
81.             cacheThreadPool.execute(new Runnable() {
    
82. 
    
83.                 @Override
    
84.                 public void run() {
    
85.                     try {
    
86.                         // 1000笔业务1s内不定时到来
    
87.                         TimeUnit.MILLISECONDS.sleep(new Random().nextInt(1000));
    
88.                     } catch (InterruptedException e1) {
    
89. 
    
90.                         e1.printStackTrace();
    
91.                     }
    
92.                     boolean isExceed = AtomicFlowControlManager.isExceed("b2b");
    
93. 
    
94.                     // 等待0.1s
    
95.                     if (isExceed) {
    
96.                         System.out.println(Thread.currentThread() + "超出流量等待最多0.1s");
    
97.                         lock.lock();
    
98.                         try {
    
99.                             condition.await(100, TimeUnit.MILLISECONDS);
    
100.                         } catch (InterruptedException e) {
     
101. 
     
102.                             e.printStackTrace();
     
103.                         } finally {
     
104.                             lock.unlock();
     
105.                         }
     
106.                         isExceed = AtomicFlowControlManager.isExceed("b2b");
     
107.                         if (!isExceed) {
     
108.                             // 睡眠300ms模拟服务时间
     
109.                             try {
     
110.                                 Thread.sleep(300);
     
111.                             } catch (InterruptedException e) {
     
112.                                 e.printStackTrace();
     
113.                             }
     
114.                             System.out.println(Thread.currentThread() + "执行服务成功");
     
115.                             AtomicFlowControlManager.releaseCounter("b2b");
     
116.                             lock.lock();
     
117.                             try {
     
118.                                 condition.signal();
     
119.                             } finally {
     
120.                                 lock.unlock();
     
121.                             }
     
122.                             count.incrementAndGet();
     
123.                         }
     
124.                     } else {
     
125.                         // 睡眠200ms模拟服务时间
     
126.                         long t1 = System.currentTimeMillis();
     
127.                         try {
     
128.                             Thread.sleep(200);
     
129.                         } catch (InterruptedException e) {
     
130.                             e.printStackTrace();
     
131.                         }
     
132.                         long t2 = System.currentTimeMillis();
     
133.                         System.out.println(Thread.currentThread() + "执行服务成功耗时[" + (t2 - t1) + "]");
     
134.                         AtomicFlowControlManager.releaseCounter("b2b");
     
135.                         lock.lock();
     
136.                         try {
     
137.                             condition.signal();
     
138.                         } finally {
     
139.                             lock.unlock();
     
140.                         }
     
141.                         count.incrementAndGet();
     
142.                     }
     
143.                     cdl.countDown();
     
144.                 }
     
145.             });
     
146.         }
     
147.         try {
     
148.             cdl.await();
     
149.         } catch (InterruptedException e) {
     
150. 
     
151.             e.printStackTrace();
     
152.         }
     
153.         cacheThreadPool.shutdown();
     
154.         long t2 = System.currentTimeMillis();
     
155.         System.out.println("耗时： " + (t2 - t1) + " 业务成功笔数： " + count.get() + "  " + (count.get() * 1000 / (t2 - t1))
     
156.                 + " 笔/秒" + "  业务处理成功率： " + (count.get() * 100 / 1000.0D) + "%");
     
157. 
     
158.     }
     
159. }

复制代码