Dubbo加权轮询负载均衡算法应用之推荐产品

Posted 2021-03-11 嘟神子

　　Dubbo加权轮询负载均衡算法，核心点：weight（固定的权重），currentWeight（当前权重，动态变化的），算法逻辑：轮询服务提供者（每个服务提供者都有weight和currentWeight），currentWeight增加weight，取最大的currentWeight，然后取对应的服务提供者，最后将取到的服务提供者的currentWeight减去总的权重（所有服务提供者的weight之和）。示例如下：

　　服务器 [A, B, C] 对应权重 [5, 1, 1] ，现在有7个请求依次进入负载均衡逻辑，选择过程如下：

请求编号	currentWeight 数组	选择结果	减去权重总和后的 currentWeight 数组
1	[5, 1, 1]	A	[-2, 1, 1]
2	[3, 2, 2]	A	[-4, 2, 2]
3	[1, 3, 3]	B	[1, -4, 3]
4	[6, -3, 4]	A	[-1, -3, 4]
5	[4, -2, 5]	C	[4, -2, -2]
6	[9, -1, -1]	A	[2, -1, -1]
7	[7, 0, 0]	A	[0, 0, 0]

　　如上，经过平滑性处理后，得到的服务器序列为 [A, A, B, A, C, A, A]。初始情况下 currentWeight = [0, 0, 0]，第7个请求处理完后，currentWeight 再次变为 [0, 0, 0]。

　　业务实践应用：推荐产品（公司开发了同类型的多种产品，对接的是不同的合作商，现要求按一定的比例，向不同的客户推荐不同合作商的产品，保证各合作商的流量。比如合作商A的产品A：合作商B的产品B：合作商C的产品C=5：1：1，那么如果有7个客户想了解该类型的产品，就向其中5人推荐A，向其中1人推荐B，向其中1人推荐C）。

　　仿照Dubbo加权轮询负载均衡算法，实现推荐产品的算法，代码如下：

 1 public class ProductMessage {
 2     private String productName;
 3     private int weight;
 4 
 5     public ProductMessage(String productName, int weight) {
 6         this.productName = productName;
 7         this.weight = weight;
 8     }
 9 
10     public String getProductName() {
11         return productName;
12     }
13 
14     public void setProductName(String productName) {
15         this.productName = productName;
16     }
17 
18     public int getWeight() {
19         return weight;
20     }
21 
22     public void setWeight(int weight) {
23         this.weight = weight;
24     }
25 }

 1 import java.util.concurrent.atomic.AtomicLong;
 2 
 3 public class WeightedRoundRobin {
 4     private int weight;
 5     private AtomicLong current = new AtomicLong(0);
 6     private long lastUpdate;
 7     public int getWeight() {
 8         return weight;
 9     }
10     public void setWeight(int weight) {
11         this.weight = weight;
12         current.set(0);
13     }
14     public long increaseCurrent() {
15         return current.addAndGet(weight);
16     }
17     public void sel(int total) {
18         current.addAndGet(-1 * total);
19     }
20     public long getLastUpdate() {
21         return lastUpdate;
22     }
23     public void setLastUpdate(long lastUpdate) {
24         this.lastUpdate = lastUpdate;
25     }
26 }

 1 import java.util.Map;
 2 import java.util.List;
 3 import java.util.Iterator;
 4 import java.util.concurrent.ConcurrentMap;
 5 import java.util.concurrent.ConcurrentHashMap;
 6 import java.util.concurrent.atomic.AtomicBoolean;
 7 
 8 /**
 9  * 业务场景
10  * 轮询推荐产品，保证各产品按指定权重被推荐
11  *
12  */
13 public class ProductRoundRobin {
14     private static int RECYCLE_PERIOD = 300000;
15     private AtomicBoolean updateLock = new AtomicBoolean();
16 
17     private ConcurrentMap<String, WeightedRoundRobin> productMap = new ConcurrentHashMap<String, WeightedRoundRobin>();
18 
19     public ProductMessage selectProduct(List<ProductMessage> productMessageList) {
20         int totalWeight = 0;
21         long maxCurrent = Long.MIN_VALUE;
22         long now = System.currentTimeMillis();
23         ProductMessage selectedProduct = null;
24         WeightedRoundRobin selectedWRR = null;
25         for (ProductMessage productMessage : productMessageList) {
26             String identifyString = productMessage.toString();
27             WeightedRoundRobin weightedRoundRobin = productMap.get(identifyString);
28             int weight = productMessage.getWeight();
29             if (weight < 0) {
30                 weight = 0;
31             }
32             if (weightedRoundRobin == null) {
33                 weightedRoundRobin = new WeightedRoundRobin();
34                 weightedRoundRobin.setWeight(weight);
35                 productMap.putIfAbsent(identifyString, weightedRoundRobin);
36                 weightedRoundRobin = productMap.get(identifyString);
37             }
38             if (weight != weightedRoundRobin.getWeight()) {
39                 weightedRoundRobin.setWeight(weight);
40             }
41             long cur = weightedRoundRobin.increaseCurrent();
42             weightedRoundRobin.setLastUpdate(now);
43             if (cur > maxCurrent) {
44                 maxCurrent = cur;
45                 selectedProduct = productMessage;
46                 selectedWRR = weightedRoundRobin;
47             }
48             totalWeight += weight;
49         }
50 
51         if (!updateLock.get() && productMessageList.size() != productMap.size()) {
52             if (updateLock.compareAndSet(false, true)) {
53                 try {
54                     // copy -> modify -> update reference
55                     ConcurrentMap<String, WeightedRoundRobin> newMap = new ConcurrentHashMap<String, WeightedRoundRobin>();
56                     newMap.putAll(productMap);
57                     Iterator<Map.Entry<String, WeightedRoundRobin>> it = newMap.entrySet().iterator();
58                     while (it.hasNext()) {
59                         Map.Entry<String, WeightedRoundRobin> item = it.next();
60                         if (now - item.getValue().getLastUpdate() > RECYCLE_PERIOD) {
61                             it.remove();
62                         }
63                     }
64                 } finally {
65                     updateLock.set(false);
66                 }
67             }
68         }
69 
70         if (selectedProduct != null) {
71             selectedWRR.sel(totalWeight);
72             return selectedProduct;
73         }
74         return productMessageList.get(0);
75     }
76 }

 1 import java.util.Map;
 2 import java.util.List;
 3 import java.util.HashMap;
 4 import java.util.ArrayList;
 5 
 6 public class ProductRoundRobinTest {
 7     public static void main(String[] args) {
 8         /**
 9          * 设定：
10          * 产品A,权重5
11          * 产品B,权重1
12          * 产品C,权重1
13          */
14         ProductMessage product1 = new ProductMessage("产品A", 5);
15         ProductMessage product2 = new ProductMessage("产品B", 1);
16         ProductMessage product3 = new ProductMessage("产品C", 1);
17         List<ProductMessage> productMessageList = new ArrayList<ProductMessage>() {{
18             add(product1);
19             add(product2);
20             add(product3);
21         }};
22 
23         ProductRoundRobin productRoundRobin = new ProductRoundRobin();
24 
25         // 进行7次推荐
26         for (int i = 0; i < 7; i++) {
27             ProductMessage selectedProduct = productRoundRobin.selectProduct(productMessageList);
28             System.out.println("productName:" + selectedProduct.getProductName());
29         }
30 
31         Map<String, Long> countMap = new HashMap<>();
32         for (int i = 0; i < 1000000; i++) {
33             ProductMessage selectedProduct = productRoundRobin.selectProduct(productMessageList);
34             Long count = countMap.get(selectedProduct.getProductName());
35             countMap.put(selectedProduct.getProductName(), count == null ? 1 : ++count);
36         }
37 
38         for (Map.Entry<String, Long> entry : countMap.entrySet()) {
39             System.out.println("introduce productName:" + entry.getKey() + "; introduce count:" + entry.getValue());
40         }
41     }
42 }

　　运行结果如下：

productName:产品A
productName:产品A
productName:产品B
productName:产品A
productName:产品C
productName:产品A
productName:产品A
introduce productName:产品A; introduce count:714286
introduce productName:产品C; introduce count:142857
introduce productName:产品B; introduce count:142857

 

　　总结：技术服务于业务，将算法逻辑，应用到实际业务中，让业务更加智能化。即所谓的科技赋能。