需求3：优化应用侧分页查询收藏接口

利用上watchTime字段，xxljob只更新上一周查看过的条目，可以减少带宽。

1.改造判断逻辑

保存watchTime，首先需要获得该用户的accountId，还要获取某一条推荐对象的旧watchTime，判断是否是今天，如果是一天那就不用更新，如果是一天就需要更新。

1
2
3
4
if (!pageRes.getContent().isEmpty()){
    Date oldWatchTime = pageRes.getContent().get(0).getWatchTime();
    watchTimeJudgeAndUpdate(Long.parseLong(AuthInfoUtils.getContext().getAuthInfo().getAccId()),oldWatchTime);
}

这个方法按道理来说要变成异步的，看后续用多线程或者@Async解决

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
private void watchTimeJudgeAndUpdate(Long accountId,Date oldWatchTime){
    try {
        //一天只保存一次watchTime
        if (oldWatchTime == null || !isSameDay(oldWatchTime,new Date())){
            tbGlobalCollectEntityService.updateWatchTime(accountId);
        }
        log.info("watchTimeUpdate success , accountId : {}",accountId);
    }
    catch (Throwable e){
        log.error("watchTimeUpdate failed , accountId : {}",accountId,e);
    }
}

private boolean isSameDay(Date date1, Date date2) {
    SimpleDateFormat sdf = new SimpleDateFormat("yyyyMMdd");
    return sdf.format(date1).equals(sdf.format(date2));
}

这里刚开始是用了jpa的抽象仓库一个一个更新，但是mentor说效率太低了，不如直接写sql，在这个方法里面支持跟mybatis那样直接执行sql。

1
2
3
4
5
6
7
8
9
@Service
public class TbGlobalCollectEntityService extends AbstractEasyEntityService<TbGlobalCollectEntity> {

    @Autowired
    private TbGlobalCollectRepository repository;
    public void updateWatchTime(Long accountId){
        repository.updateWatchTime(accountId);
    }
}

sql是更新所有当前accountId下，没有取消掉的推荐条目，把他们的watchTime都改为现在。

这里不需要返回当前时间，就更加说明这个判断逻辑跟前端没什么关系了，所以应该用异步。

1
2
3
4
@Modifying
@Transactional
@Query(value = "update TbGlobalCollectEntity p set p.watchTime=NOW() where p.status != -1 and p.accountId = :accountId")
int updateWatchTime(@Param("accountId") Long accountId);

2.xxljob改造

保存了watchTime，每周执行一次的定时任务需要改成查询所有watchTime为上周的条目。

DateUtil这个工具可以找到今天0点和上周0点的Date对象，然后再用jpa进行查询，把所有的数据投入kafka里。

优化的本意是减少部分挖掘facebook的流量。但是因为在这个topic里面不止有这个任务，还有部分通讯录挖掘的任务。减少了这部分流量同时也会减少通讯录挖掘的流量，但是后者更加有价值，并不应该减少。所以按照道理来说应该分为两个topic进行操作。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
public void collectLogUpdate(int param) {
    Date endDate = DateUtil.trimAndIncreDate(new Date(),0);
    Date startDate = DateUtil.trimAndDecreDate(new Date(), param);
    List<Specification<TbGlobalCollectEntity>> specifications = Lists.newArrayList();
    specifications.add(of(TbGlobalCollectEntity.class, "status", NEQ, GlobalSubscribeStatusEnum.DELETE.getStatus()));
    specifications.add(of(TbGlobalCollectEntity.class, "watchTime", GTE, startDate));
    specifications.add(of(TbGlobalCollectEntity.class, "watchTime", LTE, endDate));
    List<TbGlobalCollectEntity> records = tbGlobalCollectEntityService.findAll(specifications);
    if (CollectionUtils.isEmpty(records)) {
        return;
    }

    for (TbGlobalCollectEntity record : records) {
        kafkaProducer.send(KafkaConstants.COWORK_GLOBAL_COLLECT_DETAIL_CHANGE, String.valueOf(record.getId()), String.valueOf(record.getId()));
    }
}

需求4：查验500条数据

这个的工作量不在代码上而在于excel上，学到了一些技巧。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
GET cowork_global_domain_info/_search
{
  "query": {
    "bool": {
      " must ": [
        {
          "exists": {
            "field": "recommendData"
          }
        }
      ]
    }
  }
    ,"sort": {
    "_script": {
      "script": "Math.random()",
      "type": "number",
      "order": "asc"
    }
  }
  ,"_source": ["recommendData"]
}

需要注意的点：

  1. 随机从917w中找，这里sort要用random脚本
  2. 最好不要用滚动查询，有多少条是多少条
  3. searchHit对象是一个大的对象DomainIndexBO，recommendData只是其中的一个字段。所以getSourceAsString包装的是DomainIndexBO，刚开始类型转换没弄明白耽误很久。
  4. excel小技巧：在这种需要保留很多字段的时候可以用￥(dollar符)分开，因为很少有业务里面会用，防止误操作。例如在这个场景里面日志可以打印成”{recommendData.domain}￥{recomendData.other}”，然后在excel按照￥分行。
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
public void esGet() {
    Script script = new Script("Math.random()");
    ScriptSortBuilder sortBuilder = new ScriptSortBuilder(script, ScriptSortBuilder.ScriptSortType.NUMBER)
            .order(SortOrder.ASC);
    BoolQueryBuilder boolQuery = QueryBuilders.boolQuery();
    boolQuery.must(QueryBuilders.existsQuery("recommendData"));
    SearchSourceBuilder sourceBuilder = new SearchSourceBuilder()
            .query(boolQuery) // 查询参数
            .fetchSource(new String[]{"recommendData"}, null)
            .sort(sortBuilder)
            .size(500);
    SearchHit[] searchHits = domainDataIndex.query(sourceBuilder);
    int i = 1;
    for (SearchHit searchHit : searchHits) {
        DomainIndexBO domainIndexBO = JSON.parseObject(searchHit.getSourceAsString(),DomainIndexBO.class);
        RecommendDataBO recommendData = domainIndexBO.getRecommendData();
        StringBuilder stringBuilder = new StringBuilder();
        stringBuilder.append("$");
        stringBuilder.append(recommendData.getDomain()).append("$");
        stringBuilder.append(String.join(",", recommendData.getProducts())).append("$");
        if (recommendData.getAliCategories().isEmpty()){
            stringBuilder.append("null");
        }
        else {
            for (RecommendDataBO.AliCategory aliCategory : recommendData.getAliCategories()) {
                stringBuilder.append(aliCategory.getName()).append(":").append(aliCategory.getScore()).append(",");
            }
        }

        stringBuilder.append("$").append(recommendData.getIsBizP()).append("$");
        stringBuilder.append(recommendData.getIsBizVersion()).append("$");
        stringBuilder.append(recommendData.getProductsVersion()).append("$");
        if (StringUtils.isBlank(recommendData.getAliCategoryVersion())){
            stringBuilder.append("null");
        }
        else {
            stringBuilder.append(recommendData.getAliCategoryVersion());
        }


        String result = stringBuilder.toString();
        log.info("第"+i+"个:"+result);
        i++;
    }
}

。。。未完待续