基于Redis、Storm的實時數據查詢實踐

通過算法小組給出的聚合文件，我們需要實現一種業務場景，通過用戶的消費地點的商戶ID與posId，查詢出他所在的商圈，并通過商圈地點查詢出與該區域的做活動的商戶，并與之進行消息匹配，推送相應活動信息到用戶手機。

那么整個流程分為兩步，第一步，將整個聚合文件刷入緩存，文件數據格式如下：

29 1:1 102100156910958 10015691?X有限公司 0 1

29 1:1 102100156910958 10015691 X有限公司?0 1

　　字段含義分別是 地區編號、商圈編號、商戶編號、Pos編號、商戶名稱、合作商戶標志。那么我們可以通過 商戶編號+Post編號 來定位 其所在的商圈， 可以通過 地區編號+商圈編號 來獲取該商圈的所有商戶信息(Redis中直接set)。于是導入Redis是可使用key:商戶編號+Post編號 value:地區編號+商圈編號 。 隨之第二個key 為 地區編號+商圈編號 從而得到 該商圈的所有商戶（Redis中使用hset）。

將聚合文件導入Redis,，部分代碼如下

        String merchantId = StringUtils.join("V_",content[2].trim(),content[3].trim());String areabiz = StringUtils.join(content[0].trim(),content[1].trim());String merchantName = StringUtils.join(content[4].trim());String flag = StringUtils.join(content[5].trim());Map<String,String> MerchantMap = new HashMap<String,String>();MerchantMap.put(merchantName, merchantId);try {for (int i = 0; i < jedisvPools.size(); i++) {JedisPool jp = jedisvPools.get(i);Jedis jedis = null;try {jedis = jp.getResource();//key為商戶編號+PosId value為地區編號area+商圈編號bizAreaId
                    jedis.set(merchantId, areabiz);//key為商圈編號+PosId value為商戶名稱，使用sadd添加相同商圈編號+PosId的商戶if("1".equals(flag)){jedis.hmset(areabiz, MerchantMap);}} catch (Exception e) {logger.error("", e);} finally {jedis.close();}}

將需要匹配的活動商戶文件及信息導入redis,，部分代碼如下

        if (StringUtils.isEmpty(content[4]) || StringUtils.isEmpty(content[5])||StringUtils.isEmpty(content[6])) {logger.warn("數據格式有誤，內容為：{}", line);return;}String merchantId = "";String posIds = StringUtils.join(content[5]);String address = StringUtils.join(content[3]);String[] posIdArray = posIds.split("、");String url = content[6];Map<String,String> MerchantUrlAdress = new HashMap<String,String>();MerchantUrlAdress.put(address,url);for(String posId : posIdArray){merchantId = StringUtils.join("Vir_",content[4].trim(),posId.trim());try {for (int i = 0; i < jedisPools.size(); i++) {JedisPool jp = jedisPools.get(i);Jedis jedis = null;try {jedis = jp.getResource();//key為商戶編號+PosId value為地區編號area+商圈編號bizAreaId
                        jedis.hmset(merchantId,MerchantUrlAdress);} catch (Exception e) {logger.error("", e);} finally {//jedis.close();
                        jp.returnResourceObject(jedis);}}

接入用戶實時刷卡消費信息，流入storm，匹配該用戶所在商圈的活動商戶，并匹配獲取該活動商戶的地址及url信息 通過http的形式推送至支付寶或微信渠道，部分代碼如下：

            String bizAreaName = "";String bizAreaUrl = "";String address = "";//根據活動商戶ID與postId 查詢所在商圈String areabiz = virtualBusinessService.getAreaBiz(MerchantId);if(null == areabiz){resultSets.addValue(ResultSets.OpType.INSERT,"BIZAREALISTNAME",bizAreaName);resultSets.addValue(ResultSets.OpType.INSERT, "BIZAREAURL", bizAreaUrl);resultSets.addValue(ResultSets.OpType.INSERT, "BIZADDRESS", address);logger.info("VirtualTradeAreaAlgorithm="+MerchantId);return resultSets;}//根據活動ID，獲取該活動配置的商戶IdString activityMerchantCode = virtualBusinessService.getActivityConf(activityConfId);//查詢所在商圈的所有商戶信息Map<String,String> bizAreaNameMap = virtualBusinessService.getbizAreaNameSet(areabiz);if(!bizAreaNameMap.isEmpty()){//匹配活動配置的商戶for(String bizName : bizAreaNameMap.keySet()){String mapvalue = bizAreaNameMap.get(bizName).replace("V_", "");if(activityMerchantCode.contains(mapvalue)){bizAreaName = bizName;//根據活動商戶名稱查詢該商戶對應的商戶IDactivityMerchantId = bizAreaNameMap.get(bizName).replace("V_","Vir_");//根據活動商戶Id，查詢該活動商戶的url Vir_89811144816144501080209Map<String,String> bizAreaUrlAdree = virtualBusinessService.getBizUrl(activityMerchantId);if(null == bizAreaUrlAdree){address = "";bizAreaUrl = "";}else{for(String bizAdress : bizAreaUrlAdree.keySet()){address = bizAdress;bizAreaUrl = bizAreaUrlAdree.get(bizAdress);}}break;}}}                                    

?具體還在整理，后續將其補全~