1、在storm中，实时的计算出瞬间出现的热点。
某个storm task，上面算出了1万个商品的访问次数，LRUMap
频率高一些，每隔5秒，去遍历一次LRUMap，将其中的访问次数进行排序，统计出往后排的95%的商品访问次数的平均值
比如说，95%的商品，访问次数的平均值是100
从最前面开始，往后遍历，去找有没有瞬间出现的热点数据
1000，95%的平均值（100）的10倍，这个时候要设定一个阈值，比如说超出95%平均值得n倍，5倍
我们就认为是瞬间出现的热点数据，判断其可能在短时间内继续扩大的访问量，甚至达到平均值几十倍，或者几百倍
当遍历，发现说第一个商品的访问次数，小于平均值的5倍，就安全了，就break掉这个循环
热点数据，热数据，不是一个概念
有100个商品，前10个商品比较热，都访问量在500左右，其他的普通商品，访问量都在200左右，就说前10个商品是热数据
统计出来
预热的时候，将这些热数据放在缓存中去预热就可以了
热点，前面某个商品的访问量，瞬间超出了普通商品的10倍，或者100倍，1000倍，热点
2、storm这里，会直接发送http请求到nginx上，nginx上用lua脚本去处理这个请求
storm会将热点本身对应的productId，发送到流量分发的nginx上面去，放在本地缓存中
storm会将热点对应的完整的缓存数据，发送到所有的应用nginx服务器上去，直接放在本地缓存中
3、流量分发nginx的分发策略降级
流量分发nginx，加一个逻辑，就是每次访问一个商品详情页的时候，如果发现它是个热点，那么立即做流量分发策略的降级
hash策略，同一个productId的访问都同一台应用nginx服务器上
降级成对这个热点商品，流量分发采取随机负载均衡发送到所有的后端应用nginx服务器上去
瞬间将热点缓存数据的访问，从hash分发，全部到一台nginx，变成了，负载均衡发送到多台nginx上去
避免说大量的流量全部集中到一台机器，50万的访问量到一台nginx，5台应用nginx，每台就可以承载10万的访问量
4、storm还需要保存下来上次识别出来的热点list
下次去识别的时候，这次的热点list跟上次的热点list做一下diff，看看可能有的商品已经不是热点了
热点的取消的逻辑，发送http请求到流量分发的nginx上去，取消掉对应的热点数据，从nginx本地缓存中，删除
帮助类：
import java.io.BufferedReader;
import java.io.InputStream;
import java.io.InputStreamReader;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry; import org.apache.http.HttpEntity;
import org.apache.http.HttpResponse;
import org.apache.http.NameValuePair;
import org.apache.http.client.HttpClient;
import org.apache.http.client.entity.UrlEncodedFormEntity;
import org.apache.http.client.methods.HttpGet;
import org.apache.http.client.methods.HttpPost;
import org.apache.http.impl.client.DefaultHttpClient;
import org.apache.http.message.BasicNameValuePair;
import org.apache.http.util.EntityUtils; @SuppressWarnings("deprecation")
public class HttpClientUtils { /**
* 发送GET请求
* @param url 请求URL
* @return 响应结果
*/
@SuppressWarnings("resource")
public static String sendGetRequest(String url) {
String httpResponse = null; HttpClient httpclient = null;
InputStream is = null;
BufferedReader br = null; try {
// 发送GET请求
httpclient = new DefaultHttpClient();
HttpGet httpget = new HttpGet(url);
HttpResponse response = httpclient.execute(httpget); // 处理响应
HttpEntity entity = response.getEntity();
if (entity != null) {
is = entity.getContent();
br = new BufferedReader(new InputStreamReader(is)); StringBuffer buffer = new StringBuffer("");
String line = null; while ((line = br.readLine()) != null) {
buffer.append(line + "\n");
} httpResponse = buffer.toString();
}
} catch (Exception e) {
e.printStackTrace();
} finally {
try {
if(br != null) {
br.close();
}
if(is != null) {
is.close();
}
} catch (Exception e2) {
e2.printStackTrace();
}
} return httpResponse;
} /**
* 发送post请求
* @param url URL
* @param map 参数Map
* @return
*/
@SuppressWarnings({ "rawtypes", "unchecked", "resource" })
public static String sendPostRequest(String url, Map<String,String> map){
HttpClient httpClient = null;
HttpPost httpPost = null;
String result = null; try{
httpClient = new DefaultHttpClient();
httpPost = new HttpPost(url); //设置参数
List<NameValuePair> list = new ArrayList<NameValuePair>();
Iterator iterator = map.entrySet().iterator();
while(iterator.hasNext()){
Entry<String,String> elem = (Entry<String, String>) iterator.next();
list.add(new BasicNameValuePair(elem.getKey(), elem.getValue()));
}
if(list.size() > 0){
UrlEncodedFormEntity entity = new UrlEncodedFormEntity(list, "utf-8");
httpPost.setEntity(entity);
} HttpResponse response = httpClient.execute(httpPost);
if(response != null){
HttpEntity resEntity = response.getEntity();
if(resEntity != null){
result = EntityUtils.toString(resEntity, "utf-8");
}
}
} catch(Exception ex){
ex.printStackTrace();
} finally { } return result;
} }
private class HotProductFindThread implements Runnable {
@SuppressWarnings("deprecation")
public void run() {
List<Map.Entry<Long, Long>> productCountList = new ArrayList<Map.Entry<Long, Long>>();
List<Long> hotProductIdList = new ArrayList<Long>();
List<Long> lastTimeHotProductIdList = new ArrayList<Long>();
while(true) {
// 1、将LRUMap中的数据按照访问次数，进行全局的排序
// 2、计算95%的商品的访问次数的平均值
// 3、遍历排序后的商品访问次数，从最大的开始
// 4、如果某个商品比如它的访问量是平均值的10倍，就认为是缓存的热点
try {
productCountList.clear();
hotProductIdList.clear();
if(productCountMap.size() == 0) {
Utils.sleep(100);
continue;
}
LOGGER.info("【HotProductFindThread打印productCountMap的长度】size=" + productCountMap.size());
// 1、先做全局的排序
for(Map.Entry<Long, Long> productCountEntry : productCountMap.entrySet()) {
if(productCountList.size() == 0) {
productCountList.add(productCountEntry);
} else {
// 比较大小，生成最热topn的算法有很多种
// 但是我这里为了简化起见，不想引入过多的数据结构和算法的的东西
// 很有可能还是会有漏洞，但是我已经反复推演了一下了，而且也画图分析过这个算法的运行流程了
boolean bigger = false;
for(int i = 0; i < productCountList.size(); i++){
Map.Entry<Long, Long> topnProductCountEntry = productCountList.get(i);
if(productCountEntry.getValue() > topnProductCountEntry.getValue()) {
int lastIndex = productCountList.size() < productCountMap.size() ? productCountList.size() - 1 : productCountMap.size() - 2;
for(int j = lastIndex; j >= i; j--) {
if(j + 1 == productCountList.size()) {
productCountList.add(null);
}
productCountList.set(j + 1, productCountList.get(j));
}
productCountList.set(i, productCountEntry);
bigger = true;
break;
}
}
if(!bigger) {
if(productCountList.size() < productCountMap.size()) {
productCountList.add(productCountEntry);
}
}
}
}
LOGGER.info("【HotProductFindThread全局排序后的结果】productCountList=" + productCountList);
// 2、计算出95%的商品的访问次数的平均值
int calculateCount = (int)Math.floor(productCountList.size() * 0.95);
Long totalCount = 0L;
for(int i = productCountList.size() - 1; i >= productCountList.size() - calculateCount; i--) {
totalCount += productCountList.get(i).getValue();
}
Long avgCount = totalCount / calculateCount;
LOGGER.info("【HotProductFindThread计算出95%的商品的访问次数平均值】avgCount=" + avgCount);
// 3、从第一个元素开始遍历，判断是否是平均值得10倍
for(Map.Entry<Long, Long> productCountEntry : productCountList) {
if(productCountEntry.getValue() > 10 * avgCount) {
LOGGER.info("【HotProductFindThread发现一个热点】productCountEntry=" + productCountEntry);
hotProductIdList.add(productCountEntry.getKey());
if(!lastTimeHotProductIdList.contains(productCountEntry.getKey())) {
// 将缓存热点反向推送到流量分发的nginx中
String distributeNginxURL = "http://192.168.31.227/hot?productId=" + productCountEntry.getKey();
HttpClientUtils.sendGetRequest(distributeNginxURL);
// 将缓存热点，那个商品对应的完整的缓存数据，发送请求到缓存服务去获取，反向推送到所有的后端应用nginx服务器上去
String cacheServiceURL = "http://192.168.31.179:8080/getProductInfo?productId=" + productCountEntry.getKey();
String response = HttpClientUtils.sendGetRequest(cacheServiceURL);
List<NameValuePair> params = new ArrayList<NameValuePair>();
params.add(new BasicNameValuePair("productInfo", response));
String productInfo = URLEncodedUtils.format(params, HTTP.UTF_8);
String[] appNginxURLs = new String[]{
"http://192.168.31.187/hot?productId=" + productCountEntry.getKey() + "&" + productInfo,
"http://192.168.31.19/hot?productId=" + productCountEntry.getKey() + "&" + productInfo
};
for(String appNginxURL : appNginxURLs) {
HttpClientUtils.sendGetRequest(appNginxURL);
}
}
}
}
// 4、实时感知热点数据的消失
if(lastTimeHotProductIdList.size() == 0) {
if(hotProductIdList.size() > 0) {
for(Long productId : hotProductIdList) {
lastTimeHotProductIdList.add(productId);
}
LOGGER.info("【HotProductFindThread保存上次热点数据】lastTimeHotProductIdList=" + lastTimeHotProductIdList);
}
} else {
for(Long productId : lastTimeHotProductIdList) {
if(!hotProductIdList.contains(productId)) {
LOGGER.info("【HotProductFindThread发现一个热点消失了】productId=" + productId);
// 说明上次的那个商品id的热点，消失了
// 发送一个http请求给到流量分发的nginx中，取消热点缓存的标识
String url = "http://192.168.31.227/cancel_hot?productId=" + productId;
HttpClientUtils.sendGetRequest(url);
}
}
if(hotProductIdList.size() > 0) {
lastTimeHotProductIdList.clear();
for(Long productId : hotProductIdList) {
lastTimeHotProductIdList.add(productId);
}
LOGGER.info("【HotProductFindThread保存上次热点数据】lastTimeHotProductIdList=" + lastTimeHotProductIdList);
} else {
lastTimeHotProductIdList.clear();
}
}
Utils.sleep(5000);
} catch (Exception e) {
e.printStackTrace();
}
}
}
}
流量分发 local uri_args = ngx.req.get_uri_args() local product_id = uri_args["productId"] local cache_ngx = ngx.shared.my_cache local hot_product_cache_key = "hot_product_"..product_id cache_ngx:set(hot_product_cache_key, "true", 60 * 60) 后端应用 local uri_args = ngx.req.get_uri_args()
local product_id = uri_args["productId"]
local product_info = uri_args["productInfo"] local product_cache_key = "product_info_"..product_id local cache_ngx = ngx.shared.my_cache cache_ngx:set(product_cache_key,product_info,60 * 60)
math.randomseed(tostring(os.time()):reverse():sub(1, 7))
math.random(1, 2) local uri_args = ngx.req.get_uri_args()
local productId = uri_args["productId"]
local shopId = uri_args["shopId"] local hosts = {"192.168.31.187", "192.168.31.19"}
local backend = "" local hot_product_key = "hot_product_"..productId local cache_ngx = ngx.shared.my_cache
local hot_product_flag = cache_ngx:get(hot_product_key) if hot_product_flag == "true" then
math.randomseed(tostring(os.time()):reverse():sub(1, 7))
local index = math.random(1, 2)
backend = "http://"..hosts[index]
else
local hash = ngx.crc32_long(productId)
local index = (hash % 2) + 1
backend = "http://"..hosts[index]
end local requestPath = uri_args["requestPath"]
requestPath = "/"..requestPath.."?productId="..productId.."&shopId="..shopId local http = require("resty.http")
local httpc = http.new() local resp, err = httpc:request_uri(backend,{
method = "GET",
path = requestPath
}) if not resp then
ngx.say("request error: ", err)
return
end ngx.say(resp.body) httpc:close()