datasync / yoho-datasync-fullsync · Commits

Go to a project

Authored by 张帅
Commit ac5d43cd6e8721c4ef255317b8c6267ec2d5f4dd 1 parent 60b4912a

用户信息全量同步es

Inline Side-by-side
Showing 5 changed files with 131 additions and 27 deletions
... ... @@ -11,9 +11,9 @@ import java.util.List;
@Repository
public interface UserInfoRepository extends JpaRepository<UserInfo, Integer>{
@Query("select count(user.uid) from UserInfo user")
int countAllBy();
@Query("select count(user.yoho_uid) from UserInfo user where yoho_uid >= ?1 and yoho_uid< ?2 ")
int countAllByYoho_uid(Integer yohoUid, Integer yohoUid2);
@Query("select user from UserInfo user")
List<UserInfo> queryAllBy(Pageable pageable);
@Query("select user from UserInfo user where yoho_uid >= ?1 and yoho_uid< ?2 order by yoho_uid asc")
List<UserInfo> queryAllByYoho_uid(Integer yohoUid, Integer yohoUid2,Pageable pageable);
}
... ...
... ... @@ -5,23 +5,26 @@ import lombok.Data;
import javax.persistence.Column;
import javax.persistence.Entity;
import javax.persistence.Id;
import javax.persistence.Table;
import java.io.Serializable;
@Data
@Entity
@Table(name = "user_info")
public class UserInfo extends BaseEntity implements Serializable {
public class UserInfo implements Serializable {
private static final long serialVersionUID = -7163835024334835792L;
@Column(name = "nick_name")
private String nickName;
private String nick_name;
@Column(name = "head_ico")
private String headIco;
private String head_ico;
@Id
@Column(name = "yoho_uid")
private Integer uid;
private Integer yoho_uid;
@Column(name = "gender")
private Integer gender;
... ...
... ... @@ -3,6 +3,7 @@ package com.yoho.datasync.fullsync.controller;
import com.yoho.datasync.fullsync.service.impl.UserInfoDataSyncServiceImpl;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;
... ... @@ -12,11 +13,15 @@ public class UserInfoController {
@Autowired
private UserInfoDataSyncServiceImpl userInfoDataSyncService;
@RequestMapping("/syncUserInfoToES")
public String syncArticle(){
public String syncArticle(@RequestParam("uid") Integer uid, @RequestParam("flag") Integer flag,@RequestParam("size") Integer size,
@RequestParam("pageSize") Integer pageSize){
return userInfoDataSyncService.sync(uid, flag,size, pageSize);
}
return userInfoDataSyncService.sync();
@RequestMapping("/search")
public String search(){
return userInfoDataSyncService.searchAllUserData();
}
}
... ...
... ... @@ -2,7 +2,18 @@ package com.yoho.datasync.fullsync.service.impl;
import com.alibaba.fastjson.JSON;
import com.google.common.collect.Lists;
import org.elasticsearch.action.delete.DeleteRequest;
import org.elasticsearch.action.delete.DeleteResponse;
import org.elasticsearch.action.search.SearchRequest;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.index.query.BoolQueryBuilder;
import org.elasticsearch.index.query.QueryBuilder;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.index.reindex.BulkByScrollResponse;
import org.elasticsearch.index.reindex.DeleteByQueryRequest;
import org.elasticsearch.search.builder.SearchSourceBuilder;
import org.springframework.data.domain.PageRequest;
import org.springframework.data.domain.Pageable;
import com.yoho.datasync.fullsync.dal.repository.yhcommunity.model.UserInfo;
import org.elasticsearch.action.bulk.BulkRequest;
import org.elasticsearch.action.bulk.BulkResponse;
... ... @@ -17,6 +28,7 @@ import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.stereotype.Service;
import org.springframework.util.CollectionUtils;
import java.io.IOException;
import java.util.List;
... ... @@ -33,34 +45,105 @@ public class UserInfoDataSyncServiceImpl {
@Autowired
private UserInfoRepository userInfoRepository;
public String sync(){
List list = Lists.newArrayList();
UserInfo userInfo = new UserInfo();
userInfo.setNickName("zhangsan");
userInfo.setHeadIco("www");
userInfo.setUid(500031572);
userInfo.setGender(1);
list.add(userInfo);
writeData(list);
return "";
public String sync(Integer uid,Integer flag, Integer size, Integer pageSize){
int endUid;
//flag =0 只跑真实用户uid flag=1 时跑mars 马甲信息
if(flag == 0){
endUid = 1000000000;
}else {
endUid = 2100000000;
}
int count = userInfoRepository.countAllByYoho_uid(uid, endUid);
int totalPage;
String result = "";
if(count > size){
totalPage = size/pageSize;
}else {
totalPage = count%pageSize==0 ? count/pageSize: count/pageSize+1;
}
logger.info("userInfo count is {}, size is {}, pagesize is {}, totalpage is {}",count,size,pageSize,totalPage);
for (int i = 0; i < totalPage; i++) {
Pageable pageReq = PageRequest.of(i, pageSize);
List<UserInfo> userInfoList = userInfoRepository.queryAllByYoho_uid(uid,endUid, pageReq);
if(!CollectionUtils.isEmpty(userInfoList)){
writeData(userInfoList);
}
try {
Thread.sleep(200);
} catch (InterruptedException e) {
e.printStackTrace();
}
if(i == totalPage-1){
result = String.valueOf(userInfoList.get(userInfoList.size()-1).getYoho_uid());
}
}
return result;
}
private void writeData(List<UserInfo> userInfoList){
// 批量增加
BulkRequest bulkAddRequest = new BulkRequest();
for (int i = 0; i < userInfoList.size(); i++) {
UserInfo info = userInfoList.get(i);
IndexRequest indexRequest = new IndexRequest("grass", "userInfo", "");
for (UserInfo info : userInfoList) {
IndexRequest indexRequest = new IndexRequest("grass", "userInfo", String.valueOf(info.getYoho_uid()));
indexRequest.source(JSON.toJSONString(info), XContentType.JSON);
bulkAddRequest.add(indexRequest);
}
BulkResponse bulkAddResponse = null;
try {
bulkAddResponse = rhlClient.bulk(bulkAddRequest, RequestOptions.DEFAULT);
logger.info("bulkAdd: {}" , JSON.toJSONString(bulkAddResponse));
logger.info("writeData: {}" , JSON.toJSONString(bulkAddResponse));
} catch (IOException e) {
logger.info("writeData error :{}" ,e);
}
}
private void deleteData(){
try {
DeleteRequest deleteRequest = new DeleteRequest("grass", "userInfo", "");
DeleteResponse response = rhlClient.delete(deleteRequest, RequestOptions.DEFAULT);
logger.info("deleteData: {}" , JSON.toJSONString(response));
} catch (IOException e) {
e.printStackTrace();
}
}
public void deleteAllUserData(){
try {
BoolQueryBuilder queryBuilder = new BoolQueryBuilder();
queryBuilder.must(QueryBuilders.matchAllQuery());
DeleteByQueryRequest deleteRequest = new DeleteByQueryRequest("grass");
deleteRequest.setQuery(queryBuilder);
deleteRequest.setDocTypes("userInfo");
BulkByScrollResponse response = rhlClient.deleteByQuery(deleteRequest, RequestOptions.DEFAULT);
logger.info("deleteAllUserData: {}" , JSON.toJSONString(response));
} catch (IOException e) {
e.printStackTrace();
}
}
public String searchAllUserData(){
try {
BoolQueryBuilder queryBuilder = new BoolQueryBuilder();
queryBuilder.must(QueryBuilders.matchAllQuery());
SearchSourceBuilder sourceBuilder = new SearchSourceBuilder();
sourceBuilder.query(queryBuilder);
sourceBuilder.from(0);
sourceBuilder.size(200); // 获取记录数，默认10
SearchRequest searchRequest = new SearchRequest("grass");
searchRequest.source(sourceBuilder);
searchRequest.types("userInfo");
SearchResponse response = rhlClient.search(searchRequest, RequestOptions.DEFAULT);
logger.info("searchAllUserData: {}" , JSON.toJSONString(response));
return JSON.toJSONString(response.getHits().totalHits);
} catch (IOException e) {
e.printStackTrace();
}
return "";
}
}
... ...
... ... @@ -24,5 +24,18 @@ spring:
password: Z5BMngayHLUxyWLSv6+koA==
driver-class-name: com.mysql.jdbc.Driver
yhCommunity:
url: jdbc:mysql://192.168.102.219:3306/yh_community?useUnicode=true&amp;characterEncoding=UTF-8
username: yh_test
password: 9nm0icOwt6bMHjMusIfMLw==
driver-class-name: com.mysql.jdbc.Driver
elasticSearch:
host: 192.168.104.246
port: 9200
client:
connectNum: 10
connectPerRoute: 50
logging:
config: classpath:logback-boot.xml
\ No newline at end of file
... ...