记得上下班打卡 | git大法好，push需谨慎

Skip to content

L
liquidnet-bus-v1
Commit 443f53b0 authored by 张国柄's avatar 张国柄
Browse files
Options

DM;

parent da4e849a
Hide whitespace changes
Inline Side-by-side
Showing with 450 additions and 309 deletions
liquidnet-bus-service/liquidnet-service-adam/liquidnet-service-adam-impl/src/main/java/com/liquidnet/service/adam/controller/admin/AdamDMAdminController.java
View file @ 443f53b0
...@@ -5,6 +5,8 @@ import com.github.xiaoymin.knife4j.annotations.ApiSupport; ...@@ -5,6 +5,8 @@ import com.github.xiaoymin.knife4j.annotations.ApiSupport;
import com.liquidnet.service.adam.service.processor.*; import com.liquidnet.service.adam.service.processor.*;
import com.liquidnet.service.base.ResponseDto; import com.liquidnet.service.base.ResponseDto;
import io.swagger.annotations.Api; import io.swagger.annotations.Api;
import io.swagger.annotations.ApiImplicitParam;
import io.swagger.annotations.ApiImplicitParams;
import io.swagger.annotations.ApiOperation; import io.swagger.annotations.ApiOperation;
import lombok.extern.slf4j.Slf4j; import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Autowired;
...@@ -14,6 +16,9 @@ import org.springframework.web.bind.annotation.RequestMapping; ...@@ -14,6 +16,9 @@ import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestParam; import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController; import org.springframework.web.bind.annotation.RestController;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
@ApiSupport(order = 90062) @ApiSupport(order = 90062)
@Api(tags = "DM") @Api(tags = "DM")
@Slf4j @Slf4j
...@@ -42,83 +47,230 @@ public class AdamDMAdminController extends AdamBaseController { ...@@ -42,83 +47,230 @@ public class AdamDMAdminController extends AdamBaseController {
@Autowired @Autowired
DMCollectionProcessor dmCollectionProcessor; DMCollectionProcessor dmCollectionProcessor;
@ApiOperationSupport(order = 10) @ApiOperationSupport(order = 1)
@ApiOperation(value = "user") @ApiOperation(value = "T - user")
@ApiImplicitParams({
@ApiImplicitParam(type = "form", name = "dg", allowableValues = "0"),
@ApiImplicitParam(type = "form", name = "lS", allowableValues = "200"),
})
@GetMapping(value = {"user"}) @GetMapping(value = {"user"})
public ResponseDto<Object> dataMigrationForUser() { public ResponseDto<Object> dataMigrationForUser(@RequestParam(required = false) Integer dg,
dmUserInformationProcessor.handler(); @RequestParam(required = false) Integer lS,
return ResponseDto.success(); @RequestParam(required = false) String incrDt,
@RequestParam(required = false) String flg,
@RequestParam(required = false) String flu
) {
if (null != dg) {
ExecutorService exec = Executors.newFixedThreadPool(lS);
for (int i = 0; i < dg; i++) {
final int k = i;
exec.execute(() -> {
log.info("==================" + k);
dmUserInformationProcessor.dataProcessing(k, lS, null, null, k == 0 ? "1" : "");
});
}
exec.shutdown();
} else {
dmUserInformationProcessor.dataProcessing(null, null, incrDt, flg, flu);
}
return ResponseDto.success(dg);
} }
@ApiOperationSupport(order = 20) @ApiOperationSupport(order = 2)
@ApiOperation(value = "tpa") @ApiOperation(value = "T - tpa")
@ApiImplicitParams({
@ApiImplicitParam(type = "form", name = "dg", allowableValues = "1,2,3"),
})
@GetMapping(value = {"tpa"}) @GetMapping(value = {"tpa"})
public ResponseDto<Object> dataMigrationForTpa() { public ResponseDto<Object> dataMigrationForTpa(@RequestParam(required = false) Integer dg,
dmThirdPartsProcessor.handler(); @RequestParam(required = false) Integer lS,
return ResponseDto.success(); @RequestParam(required = false) String incrDt,
@RequestParam(required = false) String flg,
@RequestParam(required = false) String flu
) {
if (null != dg) {
ExecutorService exec = Executors.newFixedThreadPool(lS);
for (int i = 0; i < dg; i++) {
final int k = i;
exec.execute(() -> {
log.info("==================" + k);
dmThirdPartsProcessor.dataProcessing(k, lS, null, null, k == 0 ? "1" : "");
});
}
exec.shutdown();
} else {
dmThirdPartsProcessor.dataProcessing(null, null, incrDt, flg, flu);
}
return ResponseDto.success(dg);
} }
@ApiOperationSupport(order = 30) @ApiOperationSupport(order = 3)
@ApiOperation(value = "real") @ApiOperation(value = "T - real")
@GetMapping(value = {"real"}) @GetMapping(value = {"real"})
public ResponseDto<Object> dataMigrationForReal() { public ResponseDto<Object> dataMigrationForReal(@RequestParam(required = false) Integer dg,
dmRealNameProcessor.handler(); @RequestParam(required = false) Integer lS,
return ResponseDto.success(); @RequestParam(required = false) String incrDt,
} @RequestParam(required = false) String flg,
@RequestParam(required = false) String flu
@ApiOperationSupport(order = 40) ) {
@ApiOperation(value = "vip") if (null != dg) {
@GetMapping(value = {"vip"}) ExecutorService exec = Executors.newFixedThreadPool(lS);
public ResponseDto<Object> dataMigrationForVip() { for (int i = 0; i < dg; i++) {
dmUserMemberProcessor.handler(); final int k = i;
return ResponseDto.success(); exec.execute(() -> {
log.info("==================" + k);
dmRealNameProcessor.dataProcessing(k, lS, null, null, k == 0 ? "1" : "");
});
}
exec.shutdown();
} else {
dmRealNameProcessor.dataProcessing(null, null, incrDt, flg, flu);
}
return ResponseDto.success(dg);
} }
@ApiOperationSupport(order = 50) @ApiOperationSupport(order = 4)
@ApiOperation(value = "traces") @ApiOperation(value = "T - collect")
@GetMapping(value = {"traces"}) @GetMapping(value = {"collect"})
public ResponseDto<Object> dataMigrationForTraces() { public ResponseDto<Object> dataMigrationForCollect(@RequestParam(required = false) Integer dg,
dmTracesInfoProcessor.handler(); @RequestParam(required = false) Integer lS,
return ResponseDto.success(); @RequestParam(required = false) String incrDt,
@RequestParam(required = false) String flg,
@RequestParam(required = false) String flu
) {
if (null != dg) {
ExecutorService exec = Executors.newFixedThreadPool(lS);
for (int i = 0; i < dg; i++) {
final int k = i;
exec.execute(() -> {
log.info("==================" + k);
dmCollectionProcessor.dataProcessing(k, lS, null, null, k == 0 ? "1" : "");
});
}
exec.shutdown();
} else {
dmCollectionProcessor.dataProcessing(null, null, incrDt, flg, flu);
}
return ResponseDto.success(dg);
} }
@ApiOperationSupport(order = 60) @ApiOperationSupport(order = 5)
@ApiOperation(value = "enters") @ApiOperation(value = "T - enters")
@ApiImplicitParams({
@ApiImplicitParam(type = "form", name = "dg", allowableValues = "1,2,3,4,5"),
})
@GetMapping(value = {"enters"}) @GetMapping(value = {"enters"})
public ResponseDto<Object> dataMigrationForEnters() { public ResponseDto<Object> dataMigrationForEnters(@RequestParam(required = false) Integer dg,
dmEntersProcessor.handler(); @RequestParam(required = false) Integer lS,
return ResponseDto.success(); @RequestParam(required = false) String incrDt,
@RequestParam(required = false) String flg,
@RequestParam(required = false) String flu
) {
if (null != dg) {
ExecutorService exec = Executors.newFixedThreadPool(lS);
for (int i = 0; i < dg; i++) {
final int k = i;
exec.execute(() -> {
log.info("==================" + k);
dmEntersProcessor.dataProcessing(k, lS, null, null, k == 0 ? "1" : "");
});
}
exec.shutdown();
} else {
dmEntersProcessor.dataProcessing(null, null, incrDt, flg, flu);
}
return ResponseDto.success(dg);
} }
@ApiOperationSupport(order = 70) @ApiOperationSupport(order = 6)
@ApiOperation(value = "addr") @ApiOperation(value = "T - addr")
@ApiImplicitParams({
@ApiImplicitParam(type = "form", name = "dg", allowableValues = "1,2"),
})
@GetMapping(value = {"addr"}) @GetMapping(value = {"addr"})
public ResponseDto<Object> dataMigrationForAddr() { public ResponseDto<Object> dataMigrationForAddr(@RequestParam(required = false) Integer dg,
dmAddressesProcessor.handler(); @RequestParam(required = false) Integer lS,
return ResponseDto.success(); @RequestParam(required = false) String incrDt,
@RequestParam(required = false) String flg,
@RequestParam(required = false) String flu
) {
if (null != dg) {
ExecutorService exec = Executors.newFixedThreadPool(lS);
for (int i = 0; i < dg; i++) {
final int k = i;
exec.execute(() -> {
log.info("==================" + k);
dmAddressesProcessor.dataProcessing(k, lS, null, null, k == 0 ? "1" : "");
});
}
exec.shutdown();
} else {
dmAddressesProcessor.dataProcessing(null, null, incrDt, flg, flu);
}
return ResponseDto.success(dg);
} }
@ApiOperationSupport(order = 80) @ApiOperationSupport(order = 7)
@ApiOperation(value = "T - vip")
@GetMapping(value = {"vip"})
public ResponseDto<Object> dataMigrationForVip(@RequestParam(required = false) Integer dg,
@RequestParam(required = false) Integer lS,
@RequestParam(required = false) String incrDt,
@RequestParam(required = false) String flg,
@RequestParam(required = false) String flu
) {
if (null != dg) {
ExecutorService exec = Executors.newFixedThreadPool(lS);
for (int i = 0; i < dg; i++) {
final int k = i;
exec.execute(() -> {
log.info("==================" + k);
dmUserMemberProcessor.dataProcessing(k, lS, null, null, k == 0 ? "1" : "");
});
}
exec.shutdown();
} else {
dmUserMemberProcessor.dataProcessing(null, null, incrDt, flg, flu);
}
return ResponseDto.success(dg);
}
@ApiOperationSupport(order = 8)
@ApiOperation(value = "morder") @ApiOperation(value = "morder")
@GetMapping(value = {"morder"}) @GetMapping(value = {"morder"})
public ResponseDto<Object> dataMigrationForMOrder() { public ResponseDto<Object> dataMigrationForMOrder(@RequestParam(required = false) Integer dg,
dmMemberOrderProcessor.handler(); @RequestParam(required = false) Integer lS,
return ResponseDto.success(); @RequestParam(required = false) String incrDt,
@RequestParam(required = false) String flg,
@RequestParam(required = false) String flu
) {
dmMemberOrderProcessor.dataProcessing(null, null, incrDt, flg, flu);
return ResponseDto.success(dg);
} }
@ApiOperationSupport(order = 90) @ApiOperationSupport(order = 9)
@ApiOperation(value = "mcode") @ApiOperation(value = "mcode")
@GetMapping(value = {"mcode"}) @GetMapping(value = {"mcode"})
public ResponseDto<Object> dataMigrationForMCode() { public ResponseDto<Object> dataMigrationForMCode(@RequestParam(required = false) Integer dg,
dmMemberCodeProcessor.handler(); @RequestParam(required = false) Integer lS,
return ResponseDto.success(); @RequestParam(required = false) String incrDt,
@RequestParam(required = false) String flg,
@RequestParam(required = false) String flu
) {
dmMemberCodeProcessor.dataProcessing(null, null, incrDt, flg, flu);
return ResponseDto.success(dg);
} }
@ApiOperationSupport(order = 100) @ApiOperationSupport(order = 10)
@ApiOperation(value = "collect") @ApiOperation(value = "traces")
@GetMapping(value = {"collect"}) @GetMapping(value = {"traces"})
public ResponseDto<Object> dataMigrationForCollect() { public ResponseDto<Object> dataMigrationForTraces(@RequestParam(required = false) Integer dg,
dmCollectionProcessor.handler(); @RequestParam(required = false) Integer lS,
return ResponseDto.success(); @RequestParam(required = false) String incrDt,
@RequestParam(required = false) String flg,
@RequestParam(required = false) String flu
) {
dmTracesInfoProcessor.dataProcessing(null, null, incrDt, flg, flu);
return ResponseDto.success(dg);
} }
} }
liquidnet-bus-service/liquidnet-service-adam/liquidnet-service-adam-impl/src/main/java/com/liquidnet/service/adam/service/DataMigrationProcessorService.java
View file @ 443f53b0
package com.liquidnet.service.adam.service; package com.liquidnet.service.adam.service;
import com.liquidnet.common.cache.redis.util.RedisUtil;
import com.liquidnet.service.adam.mapper.AdamUserMapper; import com.liquidnet.service.adam.mapper.AdamUserMapper;
import com.liquidnet.service.adam.service.processor.DMAddressesProcessor;
import com.liquidnet.service.adam.service.processor.DMEntersProcessor;
import com.mysql.cj.jdbc.result.ResultSetImpl;
import lombok.SneakyThrows;
import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.mongodb.core.MongoTemplate; import org.springframework.data.mongodb.core.MongoTemplate;
import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.util.ArrayList; import java.util.ArrayList;
import java.util.Arrays; import java.util.Arrays;
import java.util.List; import java.util.List;
public abstract class DataMigrationProcessorService { public abstract class DataMigrationProcessorService {
public static final String DM_INCR_DT = "incrDt";
public static final String DM_LS = "lS";
public static final String DM_FLG = "flg";
public static final String DM_FLUSH = "flu";
public static final String MEMBER_NAME = "摩登天空会员"; public static final String MEMBER_NAME = "摩登天空会员";
public static final String MEMBER_ID = "1"; public static final String MEMBER_ID = "1";
public static List<String> reUidList = new ArrayList<>(); public static List<String> reUidList = new ArrayList<>();
...@@ -32,49 +19,7 @@ public abstract class DataMigrationProcessorService { ...@@ -32,49 +19,7 @@ public abstract class DataMigrationProcessorService {
@Autowired @Autowired
public MongoTemplate mongoTemplate; public MongoTemplate mongoTemplate;
@Autowired @Autowired
public RedisUtil redisUtil;
@Autowired
public AdamUserMapper userMapper; public AdamUserMapper userMapper;
public Connection connection = null; protected abstract void dataProcessing(Integer dg, Integer lS, String incrDt, String flg, String flu);
public PreparedStatement statement = null;
public ResultSetImpl row = null;
public void handler() {
preHandler();
try {
dataProcessing();
} catch (Exception e) {
e.printStackTrace();
}
postHandler();
}
@SneakyThrows
private void preHandler() {
Class.forName("com.mysql.cj.jdbc.Driver");
String url = "jdbc:mysql://pc-2ze6z2hxkug559q79.mysql.polardb.rds.aliyuncs.com:3306/";
// String url = "jdbc:mysql://zhengzai.mysql.polardb.rds.aliyuncs.com:3306/";
if (this instanceof DMEntersProcessor || this instanceof DMAddressesProcessor) {
url += "mall";
} else {
url += "passport";
}
connection = DriverManager.getConnection(url, "readonly", "ZWDsf8Fy");
}
protected abstract void dataProcessing();
private void postHandler() {
try {
assert row != null;
row.close();
statement.close();
connection.close();
} catch (SQLException throwables) {
throwables.printStackTrace();
}
}
} }
liquidnet-bus-service/liquidnet-service-adam/liquidnet-service-adam-impl/src/main/java/com/liquidnet/service/adam/service/processor/DMAddressesProcessor.java
View file @ 443f53b0
package com.liquidnet.service.adam.service.processor; package com.liquidnet.service.adam.service.processor;
import com.liquidnet.commons.lang.util.ServletUtils;
import com.liquidnet.service.adam.dto.vo.AdamAddressesVo; import com.liquidnet.service.adam.dto.vo.AdamAddressesVo;
import com.liquidnet.service.adam.entity.AdamAddresses; import com.liquidnet.service.adam.entity.AdamAddresses;
import com.liquidnet.service.adam.service.DataMigrationProcessorService; import com.liquidnet.service.adam.service.DataMigrationProcessorService;
...@@ -15,7 +14,9 @@ import org.springframework.data.mongodb.core.query.Query; ...@@ -15,7 +14,9 @@ import org.springframework.data.mongodb.core.query.Query;
import org.springframework.stereotype.Service; import org.springframework.stereotype.Service;
import org.springframework.util.CollectionUtils; import org.springframework.util.CollectionUtils;
import javax.servlet.http.HttpServletRequest; import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.util.ArrayList; import java.util.ArrayList;
import java.util.List; import java.util.List;
...@@ -27,16 +28,16 @@ public class DMAddressesProcessor extends DataMigrationProcessorService { ...@@ -27,16 +28,16 @@ public class DMAddressesProcessor extends DataMigrationProcessorService {
@SneakyThrows @SneakyThrows
@Override @Override
protected void dataProcessing() { public void dataProcessing(Integer dg, Integer lS, String incrDt, String flg, String flu) {
HttpServletRequest request = ServletUtils.getRequest(); if (StringUtils.isNotEmpty(flu)) {
String lS = request.getParameter(DM_LS), incrDt = request.getParameter(DM_INCR_DT),
flg = StringUtils.defaultIfBlank(request.getParameter(DM_FLG), "");
if (StringUtils.isNotEmpty(request.getParameter(DM_FLUSH))) {
log.info("DM.flush.addresses:{}", userMapper.executeForDM("adam_addresses")); log.info("DM.flush.addresses:{}", userMapper.executeForDM("adam_addresses"));
log.info("DM.flush.AdamAddressesVo:{}", mongoTemplate.remove(Query.query(Criteria.where("_id").exists(true)), AdamAddressesVo.class.getSimpleName()).getDeletedCount()); log.info("DM.flush.AdamAddressesVo:{}", mongoTemplate.remove(Query.query(Criteria.where("_id").exists(true)), AdamAddressesVo.class.getSimpleName()).getDeletedCount());
} }
String sqlCount = "select count(1) from addresses where created_at" + (StringUtils.isBlank(incrDt) ? "<" : ">=") + "curdate()"; String sqlCount = "select count(1) from addresses where created_at" + (StringUtils.isBlank(incrDt) ? "<" : ">=") + "curdate()";
if (null != dg) {
sqlCount = sqlCount + " and id%" + lS + "=" + dg;
}
String field = "id,user_id,user_name,user_phone,province,city,county,user_address,defaultDelivery,created_at"; String field = "id,user_id,user_name,user_phone,province,city,county,user_address,defaultDelivery,created_at";
// 1-id,2-user_id,3-user_name,4-user_phone,5-province,6-city,7-county,8-user_address,9-defaultDelivery,10-created_at // 1-id,2-user_id,3-user_name,4-user_phone,5-province,6-city,7-county,8-user_address,9-defaultDelivery,10-created_at
String sql = sqlCount.replace("count(1)", field); String sql = sqlCount.replace("count(1)", field);
...@@ -46,15 +47,20 @@ public class DMAddressesProcessor extends DataMigrationProcessorService { ...@@ -46,15 +47,20 @@ public class DMAddressesProcessor extends DataMigrationProcessorService {
} }
sql += " order by id limit ?,?"; sql += " order by id limit ?,?";
statement = connection.prepareStatement(sqlCount); Class.forName("com.mysql.cj.jdbc.Driver");
row = (ResultSetImpl) statement.executeQuery(); // String url = "jdbc:mysql://pc-2ze6z2hxkug559q79.mysql.polardb.rds.aliyuncs.com:3306/mall";
String url = "jdbc:mysql://zhengzai.mysql.polardb.rds.aliyuncs.com:3306/mall";
Connection connection = DriverManager.getConnection(url, "readonly", "ZWDsf8Fy");
PreparedStatement statement = connection.prepareStatement(sqlCount);
ResultSetImpl row = (ResultSetImpl) statement.executeQuery();
row.first(); row.first();
int ct = row.getInt(1), pSize = 1000, num = 0, tl = 0, pl = ct; int ct = row.getInt(1), pSize = 1000, num = 0, tl = 0, pl = ct;
log.info("DM.execute.counts:{}/{},{} ~ {} ----------------------------> start.sql:{}", pl, num, pSize, tl, sql); log.info("DM.execute.counts:{}/{},{} ~ {} ----------------------------> start.sql:{}", pl, num, pSize, tl, sql);
while (ct > 0) { while (ct > 0) {
statement = connection.prepareStatement(sql); statement = connection.prepareStatement(sql);
int ls = (num == 0 && StringUtils.isNotEmpty(lS)) ? Integer.parseInt(lS) : pSize * num; int ls = pSize * num;
statement.setInt(1, ls); statement.setInt(1, ls);
statement.setInt(2, pSize); statement.setInt(2, pSize);
row = (ResultSetImpl) statement.executeQuery(); row = (ResultSetImpl) statement.executeQuery();
...@@ -96,5 +102,8 @@ public class DMAddressesProcessor extends DataMigrationProcessorService { ...@@ -96,5 +102,8 @@ public class DMAddressesProcessor extends DataMigrationProcessorService {
ct -= pSize; ct -= pSize;
} }
log.info("DM.execute.counts:{}/{},{} ~ {} ----------------------------> end", pl, num, pSize, tl); log.info("DM.execute.counts:{}/{},{} ~ {} ----------------------------> end", pl, num, pSize, tl);
row.close();
statement.close();
connection.close();
} }
} }
liquidnet-bus-service/liquidnet-service-adam/liquidnet-service-adam-impl/src/main/java/com/liquidnet/service/adam/service/processor/DMCollectionProcessor.java
View file @ 443f53b0
package com.liquidnet.service.adam.service.processor; package com.liquidnet.service.adam.service.processor;
import com.liquidnet.commons.lang.util.ServletUtils;
import com.liquidnet.service.adam.dto.vo.AdamCollectVo; import com.liquidnet.service.adam.dto.vo.AdamCollectVo;
import com.liquidnet.service.adam.entity.AdamCollection; import com.liquidnet.service.adam.entity.AdamCollection;
import com.liquidnet.service.adam.service.DataMigrationProcessorService; import com.liquidnet.service.adam.service.DataMigrationProcessorService;
...@@ -15,7 +14,9 @@ import org.springframework.data.mongodb.core.query.Query; ...@@ -15,7 +14,9 @@ import org.springframework.data.mongodb.core.query.Query;
import org.springframework.stereotype.Service; import org.springframework.stereotype.Service;
import org.springframework.util.CollectionUtils; import org.springframework.util.CollectionUtils;
import javax.servlet.http.HttpServletRequest; import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.util.ArrayList; import java.util.ArrayList;
import java.util.List; import java.util.List;
...@@ -27,17 +28,17 @@ public class DMCollectionProcessor extends DataMigrationProcessorService { ...@@ -27,17 +28,17 @@ public class DMCollectionProcessor extends DataMigrationProcessorService {
@SneakyThrows @SneakyThrows
@Override @Override
protected void dataProcessing() { public void dataProcessing(Integer dg, Integer lS, String incrDt, String flg, String flu) {
HttpServletRequest request = ServletUtils.getRequest(); if (StringUtils.isNotEmpty(flu)) {
String lS = request.getParameter(DM_LS), incrDt = request.getParameter(DM_INCR_DT),
flg = StringUtils.defaultIfBlank(request.getParameter(DM_FLG), "");
if (StringUtils.isNotEmpty(request.getParameter(DM_FLUSH))) {
log.info("DM.flush.collection:{}", userMapper.executeForDM("adam_collection")); log.info("DM.flush.collection:{}", userMapper.executeForDM("adam_collection"));
log.info("DM.flush.AdamCollectVo:{}", mongoTemplate.remove(Query.query(Criteria.where("_id").exists(true)), AdamCollectVo.class.getSimpleName()).getDeletedCount()); log.info("DM.flush.AdamCollectVo:{}", mongoTemplate.remove(Query.query(Criteria.where("_id").exists(true)), AdamCollectVo.class.getSimpleName()).getDeletedCount());
} }
String sqlCount = "select count(1) from user_collections where type='TICKET' and status=1 and content_id>=5722 and created_at"+(StringUtils.isBlank(incrDt) ? "<" : ">=")+"curdate()"; String sqlCount = "select count(1) from user_collections where type='TICKET' and status=1 and content_id>=5722 and created_at"+(StringUtils.isBlank(incrDt) ? "<" : ">=")+"curdate()";
if (null != dg) {
sqlCount = sqlCount + " and id%" + lS + "=" + dg;
}
String field = "uid,content_id,`type`,`status`,created_at,updated_at"; String field = "uid,content_id,`type`,`status`,created_at,updated_at";
// 1-uid,2-content_id,3-`type`,4-`status`,5-created_at,6-updated_at // 1-uid,2-content_id,3-`type`,4-`status`,5-created_at,6-updated_at
String sql = sqlCount.replace("count(1)", field); String sql = sqlCount.replace("count(1)", field);
...@@ -47,15 +48,20 @@ public class DMCollectionProcessor extends DataMigrationProcessorService { ...@@ -47,15 +48,20 @@ public class DMCollectionProcessor extends DataMigrationProcessorService {
} }
sql += " order by id limit ?,?"; sql += " order by id limit ?,?";
statement = connection.prepareStatement(sqlCount); Class.forName("com.mysql.cj.jdbc.Driver");
row = (ResultSetImpl) statement.executeQuery(); // String url = "jdbc:mysql://pc-2ze6z2hxkug559q79.mysql.polardb.rds.aliyuncs.com:3306/passport";
String url = "jdbc:mysql://zhengzai.mysql.polardb.rds.aliyuncs.com:3306/passport";
Connection connection = DriverManager.getConnection(url, "readonly", "ZWDsf8Fy");
PreparedStatement statement = connection.prepareStatement(sqlCount);
ResultSetImpl row = (ResultSetImpl) statement.executeQuery();
row.first(); row.first();
int ct = row.getInt(1), pSize = 1000, num = 0, tl = 0, pl = ct; int ct = row.getInt(1), pSize = 1000, num = 0, tl = 0, pl = ct;
log.info("DM.execute.counts:{}/{},{} ~ {} ----------------------------> start.sql:{}", pl, num, pSize, tl, sql); log.info("DM.execute.counts:{}/{},{} ~ {} ----------------------------> start.sql:{}", pl, num, pSize, tl, sql);
while (ct > 0) { while (ct > 0) {
statement = connection.prepareStatement(sql); statement = connection.prepareStatement(sql);
int ls = (num == 0 && StringUtils.isNotEmpty(lS)) ? Integer.parseInt(lS) : pSize * num; int ls = pSize * num;
statement.setInt(1, ls); statement.setInt(1, ls);
statement.setInt(2, pSize); statement.setInt(2, pSize);
row = (ResultSetImpl) statement.executeQuery(); row = (ResultSetImpl) statement.executeQuery();
...@@ -89,5 +95,8 @@ public class DMCollectionProcessor extends DataMigrationProcessorService { ...@@ -89,5 +95,8 @@ public class DMCollectionProcessor extends DataMigrationProcessorService {
ct -= pSize; ct -= pSize;
} }
log.info("DM.execute.counts:{}/{},{} ~ {} ----------------------------> end", pl, num, pSize, tl); log.info("DM.execute.counts:{}/{},{} ~ {} ----------------------------> end", pl, num, pSize, tl);
row.close();
statement.close();
connection.close();
} }
} }
liquidnet-bus-service/liquidnet-service-adam/liquidnet-service-adam-impl/src/main/java/com/liquidnet/service/adam/service/processor/DMEntersProcessor.java
View file @ 443f53b0
package com.liquidnet.service.adam.service.processor; package com.liquidnet.service.adam.service.processor;
import com.liquidnet.commons.lang.util.ServletUtils;
import com.liquidnet.service.adam.dto.vo.AdamEntersVo; import com.liquidnet.service.adam.dto.vo.AdamEntersVo;
import com.liquidnet.service.adam.entity.AdamEnters; import com.liquidnet.service.adam.entity.AdamEnters;
import com.liquidnet.service.adam.service.DataMigrationProcessorService; import com.liquidnet.service.adam.service.DataMigrationProcessorService;
...@@ -15,7 +14,9 @@ import org.springframework.data.mongodb.core.query.Query; ...@@ -15,7 +14,9 @@ import org.springframework.data.mongodb.core.query.Query;
import org.springframework.stereotype.Service; import org.springframework.stereotype.Service;
import org.springframework.util.CollectionUtils; import org.springframework.util.CollectionUtils;
import javax.servlet.http.HttpServletRequest; import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.util.ArrayList; import java.util.ArrayList;
import java.util.List; import java.util.List;
...@@ -27,17 +28,17 @@ public class DMEntersProcessor extends DataMigrationProcessorService { ...@@ -27,17 +28,17 @@ public class DMEntersProcessor extends DataMigrationProcessorService {
@SneakyThrows @SneakyThrows
@Override @Override
protected void dataProcessing() { public void dataProcessing(Integer dg, Integer lS, String incrDt, String flg, String flu) {
HttpServletRequest request = ServletUtils.getRequest(); if (StringUtils.isNotEmpty(flu)) {
String lS = request.getParameter(DM_LS), incrDt = request.getParameter(DM_INCR_DT),
flg = StringUtils.defaultIfBlank(request.getParameter(DM_FLG), "");
if (StringUtils.isNotEmpty(request.getParameter(DM_FLUSH))) {
log.info("DM.flush.enters:{}", userMapper.executeForDM("adam_enters")); log.info("DM.flush.enters:{}", userMapper.executeForDM("adam_enters"));
log.info("DM.flush.AdamEntersVo:{}", mongoTemplate.remove(Query.query(Criteria.where("_id").exists(true)), AdamEntersVo.class.getSimpleName()).getDeletedCount()); log.info("DM.flush.AdamEntersVo:{}", mongoTemplate.remove(Query.query(Criteria.where("_id").exists(true)), AdamEntersVo.class.getSimpleName()).getDeletedCount());
} }
String sqlCount = "select count(1) from enters where (type <> 1 or (type = 1 and is_certification = 'yes')) and created_at" + (StringUtils.isBlank(incrDt) ? "<" : ">=") + "curdate()"; String sqlCount = "select count(1) from enters where (type <> 1 or (type = 1 and is_certification = 'yes')) and created_at" + (StringUtils.isBlank(incrDt) ? "<" : ">=") + "curdate()";
if (null != dg) {
sqlCount = sqlCount + " and id%" + lS + "=" + dg;
}
String field = "id,user_id,type,`name`,mobile,idcode,if(is_default='no', 0, 1),created_at"; String field = "id,user_id,type,`name`,mobile,idcode,if(is_default='no', 0, 1),created_at";
// 1-id,2-user_id,3-type,4-`name`,5-mobile,6-idcode,7-if(is_default='no', 0, 1),8-created_at // 1-id,2-user_id,3-type,4-`name`,5-mobile,6-idcode,7-if(is_default='no', 0, 1),8-created_at
String sql = sqlCount.replace("count(1)", field); String sql = sqlCount.replace("count(1)", field);
...@@ -47,15 +48,20 @@ public class DMEntersProcessor extends DataMigrationProcessorService { ...@@ -47,15 +48,20 @@ public class DMEntersProcessor extends DataMigrationProcessorService {
} }
sql += " order by id limit ?,?"; sql += " order by id limit ?,?";
statement = connection.prepareStatement(sqlCount); Class.forName("com.mysql.cj.jdbc.Driver");
row = (ResultSetImpl) statement.executeQuery(); // String url = "jdbc:mysql://pc-2ze6z2hxkug559q79.mysql.polardb.rds.aliyuncs.com:3306/mall";
String url = "jdbc:mysql://zhengzai.mysql.polardb.rds.aliyuncs.com:3306/mall";
Connection connection = DriverManager.getConnection(url, "readonly", "ZWDsf8Fy");
PreparedStatement statement = connection.prepareStatement(sqlCount);
ResultSetImpl row = (ResultSetImpl) statement.executeQuery();
row.first(); row.first();
int ct = row.getInt(1), pSize = 1000, num = 0, tl = 0, pl = ct; int ct = row.getInt(1), pSize = 1000, num = 0, tl = 0, pl = ct;
log.info("DM.execute.counts:{}/{},{} ~ {} ----------------------------> start.sql:{}", pl, num, pSize, tl, sql); log.info("DM.execute.counts:{}/{},{} ~ {} ----------------------------> start.sql:{}", pl, num, pSize, tl, sql);
while (ct > 0) { while (ct > 0) {
statement = connection.prepareStatement(sql); statement = connection.prepareStatement(sql);
int ls = (num == 0 && StringUtils.isNotEmpty(lS)) ? Integer.parseInt(lS) : pSize * num; int ls = pSize * num;
statement.setInt(1, ls); statement.setInt(1, ls);
statement.setInt(2, pSize); statement.setInt(2, pSize);
row = (ResultSetImpl) statement.executeQuery(); row = (ResultSetImpl) statement.executeQuery();
...@@ -95,5 +101,8 @@ public class DMEntersProcessor extends DataMigrationProcessorService { ...@@ -95,5 +101,8 @@ public class DMEntersProcessor extends DataMigrationProcessorService {
ct -= pSize; ct -= pSize;
} }
log.info("DM.execute.counts:{}/{},{} ~ {} ----------------------------> end", pl, num, pSize, tl); log.info("DM.execute.counts:{}/{},{} ~ {} ----------------------------> end", pl, num, pSize, tl);
row.close();
statement.close();
connection.close();
} }
} }
liquidnet-bus-service/liquidnet-service-adam/liquidnet-service-adam-impl/src/main/java/com/liquidnet/service/adam/service/processor/DMMemberCodeProcessor.java
View file @ 443f53b0
package com.liquidnet.service.adam.service.processor; package com.liquidnet.service.adam.service.processor;
import com.liquidnet.commons.lang.util.ServletUtils;
import com.liquidnet.service.adam.dto.vo.AdamMemberCodeVo; import com.liquidnet.service.adam.dto.vo.AdamMemberCodeVo;
import com.liquidnet.service.adam.entity.AdamMemberCode; import com.liquidnet.service.adam.entity.AdamMemberCode;
import com.liquidnet.service.adam.service.DataMigrationProcessorService; import com.liquidnet.service.adam.service.DataMigrationProcessorService;
...@@ -15,7 +14,9 @@ import org.springframework.data.mongodb.core.query.Query; ...@@ -15,7 +14,9 @@ import org.springframework.data.mongodb.core.query.Query;
import org.springframework.stereotype.Service; import org.springframework.stereotype.Service;
import org.springframework.util.CollectionUtils; import org.springframework.util.CollectionUtils;
import javax.servlet.http.HttpServletRequest; import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.util.ArrayList; import java.util.ArrayList;
import java.util.HashMap; import java.util.HashMap;
import java.util.List; import java.util.List;
...@@ -180,32 +181,37 @@ public class DMMemberCodeProcessor extends DataMigrationProcessorService { ...@@ -180,32 +181,37 @@ public class DMMemberCodeProcessor extends DataMigrationProcessorService {
@SneakyThrows @SneakyThrows
@Override @Override
protected void dataProcessing() { public void dataProcessing(Integer dg, Integer lS, String incrDt, String flg, String flu) {
HttpServletRequest request = ServletUtils.getRequest(); if (StringUtils.isNotEmpty(flu)) {
String lS = request.getParameter(DM_LS), incrDt = request.getParameter(DM_INCR_DT),
flg = StringUtils.defaultIfBlank(request.getParameter(DM_FLG), "");
if (StringUtils.isNotEmpty(request.getParameter(DM_FLUSH))) {
log.info("DM.flush.member_code:{}", userMapper.executeForDM("adam_member_code")); log.info("DM.flush.member_code:{}", userMapper.executeForDM("adam_member_code"));
log.info("DM.flush.AdamMemberCodeVo:{}", mongoTemplate.remove(Query.query(Criteria.where("_id").exists(true)), AdamMemberCodeVo.class.getSimpleName()).getDeletedCount()); log.info("DM.flush.AdamMemberCodeVo:{}", mongoTemplate.remove(Query.query(Criteria.where("_id").exists(true)), AdamMemberCodeVo.class.getSimpleName()).getDeletedCount());
} }
String sqlCount = "select count(1) from vip_card_codes where buy_order_id <> use_order_id or status=0"; String sqlCount = "select count(1) from vip_card_codes where buy_order_id <> use_order_id or status=0";
if (null != dg) {
sqlCount = sqlCount + " and id%" + lS + "=" + dg;
}
String field = "vip_card_code,`type`+1,card_number,`status`,created_at,updated_at,buy_order_id,buy_uid,buy_time,use_order_id,use_uid,use_time"; String field = "vip_card_code,`type`+1,card_number,`status`,created_at,updated_at,buy_order_id,buy_uid,buy_time,use_order_id,use_uid,use_time";
// 1-vip_card_code,2-`type`+1,3-card_number,4-`status`,5-created_at,6-updated_at, // 1-vip_card_code,2-`type`+1,3-card_number,4-`status`,5-created_at,6-updated_at,
// 7-buy_order_id,8-buy_uid,9-buy_time,10-use_order_id,11-use_uid,12-use_time // 7-buy_order_id,8-buy_uid,9-buy_time,10-use_order_id,11-use_uid,12-use_time
String sql = sqlCount.replace("count(1)", field); String sql = sqlCount.replace("count(1)", field);
sql += " order by id limit ?,?"; sql += " order by id limit ?,?";
statement = connection.prepareStatement(sqlCount); Class.forName("com.mysql.cj.jdbc.Driver");
row = (ResultSetImpl) statement.executeQuery(); // String url = "jdbc:mysql://pc-2ze6z2hxkug559q79.mysql.polardb.rds.aliyuncs.com:3306/passport";
String url = "jdbc:mysql://zhengzai.mysql.polardb.rds.aliyuncs.com:3306/passport";
Connection connection = DriverManager.getConnection(url, "readonly", "ZWDsf8Fy");
PreparedStatement statement = connection.prepareStatement(sqlCount);
ResultSetImpl row = (ResultSetImpl) statement.executeQuery();
row.first(); row.first();
int ct = row.getInt(1), pSize = 1000, num = 0, tl = 0, pl = ct; int ct = row.getInt(1), pSize = 1000, num = 0, tl = 0, pl = ct;
log.info("DM.execute.counts:{}/{},{} ~ {} ----------------------------> start.sql:{}", pl, num, pSize, tl, sql); log.info("DM.execute.counts:{}/{},{} ~ {} ----------------------------> start.sql:{}", pl, num, pSize, tl, sql);
while (ct > 0) { while (ct > 0) {
statement = connection.prepareStatement(sql); statement = connection.prepareStatement(sql);
int ls = (num == 0 && StringUtils.isNotEmpty(lS)) ? Integer.parseInt(lS) : pSize * num; int ls = pSize * num;
statement.setInt(1, ls); statement.setInt(1, ls);
statement.setInt(2, pSize); statement.setInt(2, pSize);
row = (ResultSetImpl) statement.executeQuery(); row = (ResultSetImpl) statement.executeQuery();
...@@ -248,5 +254,8 @@ public class DMMemberCodeProcessor extends DataMigrationProcessorService { ...@@ -248,5 +254,8 @@ public class DMMemberCodeProcessor extends DataMigrationProcessorService {
ct -= pSize; ct -= pSize;
} }
log.info("DM.execute.counts:{}/{},{} ~ {} ----------------------------> end", pl, num, pSize, tl); log.info("DM.execute.counts:{}/{},{} ~ {} ----------------------------> end", pl, num, pSize, tl);
row.close();
statement.close();
connection.close();
} }
} }
liquidnet-bus-service/liquidnet-service-adam/liquidnet-service-adam-impl/src/main/java/com/liquidnet/service/adam/service/processor/DMMemberOrderProcessor.java
View file @ 443f53b0
package com.liquidnet.service.adam.service.processor; package com.liquidnet.service.adam.service.processor;
import com.liquidnet.commons.lang.util.ServletUtils;
import com.liquidnet.service.adam.dto.vo.AdamMemberOrderVo; import com.liquidnet.service.adam.dto.vo.AdamMemberOrderVo;
import com.liquidnet.service.adam.entity.AdamMemberOrder; import com.liquidnet.service.adam.entity.AdamMemberOrder;
import com.liquidnet.service.adam.service.DataMigrationProcessorService; import com.liquidnet.service.adam.service.DataMigrationProcessorService;
...@@ -15,7 +14,9 @@ import org.springframework.data.mongodb.core.query.Query; ...@@ -15,7 +14,9 @@ import org.springframework.data.mongodb.core.query.Query;
import org.springframework.stereotype.Service; import org.springframework.stereotype.Service;
import org.springframework.util.CollectionUtils; import org.springframework.util.CollectionUtils;
import javax.servlet.http.HttpServletRequest; import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.util.ArrayList; import java.util.ArrayList;
import java.util.HashMap; import java.util.HashMap;
import java.util.List; import java.util.List;
...@@ -51,17 +52,17 @@ public class DMMemberOrderProcessor extends DataMigrationProcessorService { ...@@ -51,17 +52,17 @@ public class DMMemberOrderProcessor extends DataMigrationProcessorService {
@SneakyThrows @SneakyThrows
@Override @Override
protected void dataProcessing() { public void dataProcessing(Integer dg, Integer lS, String incrDt, String flg, String flu) {
HttpServletRequest request = ServletUtils.getRequest(); if (StringUtils.isNotEmpty(flu)) {
String lS = request.getParameter(DM_LS), incrDt = request.getParameter(DM_INCR_DT),
flg = StringUtils.defaultIfBlank(request.getParameter(DM_FLG), "");
if (StringUtils.isNotEmpty(request.getParameter(DM_FLUSH))) {
log.info("DM.flush.member_order:{}", userMapper.executeForDM("adam_member_order")); log.info("DM.flush.member_order:{}", userMapper.executeForDM("adam_member_order"));
log.info("DM.flush.AdamMemberOrderVo:{}", mongoTemplate.remove(Query.query(Criteria.where("_id").exists(true)), AdamMemberOrderVo.class.getSimpleName()).getDeletedCount()); log.info("DM.flush.AdamMemberOrderVo:{}", mongoTemplate.remove(Query.query(Criteria.where("_id").exists(true)), AdamMemberOrderVo.class.getSimpleName()).getDeletedCount());
} }
String sqlCount = "select count(1) from user_vip_card_orders where status in (1,5) and created_at" + (StringUtils.isBlank(incrDt) ? "<" : ">=") + "curdate()"; String sqlCount = "select count(1) from user_vip_card_orders where status in (1,5) and created_at" + (StringUtils.isBlank(incrDt) ? "<" : ">=") + "curdate()";
if (null != dg) {
sqlCount = sqlCount + " and id%" + lS + "=" + dg;
}
String field = "order_code,uid,order_sub_type,vip_card_price,price,status,card_number,birthday,pay_type,pay_code,pay_at,created_at,updated_at,client_ip,source,version"; String field = "order_code,uid,order_sub_type,vip_card_price,price,status,card_number,birthday,pay_type,pay_code,pay_at,created_at,updated_at,client_ip,source,version";
// 1-order_code,2-uid,3-order_sub_type,4-vip_card_price,5-price,6-status,7-card_number,8-birthday, // 1-order_code,2-uid,3-order_sub_type,4-vip_card_price,5-price,6-status,7-card_number,8-birthday,
// 9-pay_type,10-pay_code,11-pay_at,12-created_at,13-updated_at,14-client_ip,15-source,16-version // 9-pay_type,10-pay_code,11-pay_at,12-created_at,13-updated_at,14-client_ip,15-source,16-version
...@@ -72,15 +73,20 @@ public class DMMemberOrderProcessor extends DataMigrationProcessorService { ...@@ -72,15 +73,20 @@ public class DMMemberOrderProcessor extends DataMigrationProcessorService {
} }
sql += " order by id limit ?,?"; sql += " order by id limit ?,?";
statement = connection.prepareStatement(sqlCount); Class.forName("com.mysql.cj.jdbc.Driver");
row = (ResultSetImpl) statement.executeQuery(); // String url = "jdbc:mysql://pc-2ze6z2hxkug559q79.mysql.polardb.rds.aliyuncs.com:3306/passport";
String url = "jdbc:mysql://zhengzai.mysql.polardb.rds.aliyuncs.com:3306/passport";
Connection connection = DriverManager.getConnection(url, "readonly", "ZWDsf8Fy");
PreparedStatement statement = connection.prepareStatement(sqlCount);
ResultSetImpl row = (ResultSetImpl) statement.executeQuery();
row.first(); row.first();
int ct = row.getInt(1), pSize = 1000, num = 0, tl = 0, pl = ct; int ct = row.getInt(1), pSize = 1000, num = 0, tl = 0, pl = ct;
log.info("DM.execute.counts:{}/{},{} ~ {} ----------------------------> start.sql:{}", pl, num, pSize, tl, sql); log.info("DM.execute.counts:{}/{},{} ~ {} ----------------------------> start.sql:{}", pl, num, pSize, tl, sql);
while (ct > 0) { while (ct > 0) {
statement = connection.prepareStatement(sql); statement = connection.prepareStatement(sql);
int ls = (num == 0 && StringUtils.isNotEmpty(lS)) ? Integer.parseInt(lS) : pSize * num; int ls = pSize * num;
statement.setInt(1, ls); statement.setInt(1, ls);
statement.setInt(2, pSize); statement.setInt(2, pSize);
row = (ResultSetImpl) statement.executeQuery(); row = (ResultSetImpl) statement.executeQuery();
...@@ -132,5 +138,8 @@ public class DMMemberOrderProcessor extends DataMigrationProcessorService { ...@@ -132,5 +138,8 @@ public class DMMemberOrderProcessor extends DataMigrationProcessorService {
ct -= pSize; ct -= pSize;
} }
log.info("DM.execute.counts:{}/{},{} ~ {} ----------------------------> end", pl, num, pSize, tl); log.info("DM.execute.counts:{}/{},{} ~ {} ----------------------------> end", pl, num, pSize, tl);
row.close();
statement.close();
connection.close();
} }
} }
liquidnet-bus-service/liquidnet-service-adam/liquidnet-service-adam-impl/src/main/java/com/liquidnet/service/adam/service/processor/DMRealNameProcessor.java
View file @ 443f53b0
package com.liquidnet.service.adam.service.processor; package com.liquidnet.service.adam.service.processor;
import com.liquidnet.commons.lang.util.ServletUtils;
import com.liquidnet.service.adam.dto.vo.AdamRealInfoVo; import com.liquidnet.service.adam.dto.vo.AdamRealInfoVo;
import com.liquidnet.service.adam.entity.AdamRealName; import com.liquidnet.service.adam.entity.AdamRealName;
import com.liquidnet.service.adam.service.DataMigrationProcessorService; import com.liquidnet.service.adam.service.DataMigrationProcessorService;
...@@ -15,7 +14,9 @@ import org.springframework.data.mongodb.core.query.Query; ...@@ -15,7 +14,9 @@ import org.springframework.data.mongodb.core.query.Query;
import org.springframework.stereotype.Service; import org.springframework.stereotype.Service;
import org.springframework.util.CollectionUtils; import org.springframework.util.CollectionUtils;
import javax.servlet.http.HttpServletRequest; import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.util.ArrayList; import java.util.ArrayList;
import java.util.List; import java.util.List;
...@@ -27,17 +28,17 @@ public class DMRealNameProcessor extends DataMigrationProcessorService { ...@@ -27,17 +28,17 @@ public class DMRealNameProcessor extends DataMigrationProcessorService {
@SneakyThrows @SneakyThrows
@Override @Override
protected void dataProcessing() { public void dataProcessing(Integer dg, Integer lS, String incrDt, String flg, String flu) {
HttpServletRequest request = ServletUtils.getRequest(); if (StringUtils.isNotEmpty(flu)) {
String lS = request.getParameter(DM_LS), incrDt = request.getParameter(DM_INCR_DT),
flg = StringUtils.defaultIfBlank(request.getParameter(DM_FLG), "");
if (StringUtils.isNotEmpty(request.getParameter(DM_FLUSH))) {
log.info("DM.flush.real_name:{}", userMapper.executeForDM("adam_real_name")); log.info("DM.flush.real_name:{}", userMapper.executeForDM("adam_real_name"));
log.info("DM.flush.AdamRealInfoVo:{}", mongoTemplate.remove(Query.query(Criteria.where("_id").exists(true)), AdamRealInfoVo.class.getSimpleName()).getDeletedCount()); log.info("DM.flush.AdamRealInfoVo:{}", mongoTemplate.remove(Query.query(Criteria.where("_id").exists(true)), AdamRealInfoVo.class.getSimpleName()).getDeletedCount());
} }
String sqlCount = "select count(1) from user_real_name where id not in (9145,7456,7448,7197,7194,4137,4138,4139,269,261,262,263,264,265,266,267,251) and created_at" + (StringUtils.isBlank(incrDt) ? "<" : ">=") + "curdate()"; String sqlCount = "select count(1) from user_real_name where id not in (9145,7456,7448,7197,7194,4137,4138,4139,269,261,262,263,264,265,266,267,251) and created_at" + (StringUtils.isBlank(incrDt) ? "<" : ">=") + "curdate()";
if (null != dg) {
sqlCount = sqlCount + " and id%" + lS + "=" + dg;
}
String field = "id,uid,real_name,id_card,ifnull(created_at, '1970-01-01 00:00:00') created_at,updated_at"; String field = "id,uid,real_name,id_card,ifnull(created_at, '1970-01-01 00:00:00') created_at,updated_at";
// 1-id,2-uid,3-real_name,4-id_card,5-created_at,6-updated_at // 1-id,2-uid,3-real_name,4-id_card,5-created_at,6-updated_at
String sql = sqlCount.replace("count(1)", field); String sql = sqlCount.replace("count(1)", field);
...@@ -47,15 +48,20 @@ public class DMRealNameProcessor extends DataMigrationProcessorService { ...@@ -47,15 +48,20 @@ public class DMRealNameProcessor extends DataMigrationProcessorService {
} }
sql += " order by id limit ?,?"; sql += " order by id limit ?,?";
statement = connection.prepareStatement(sqlCount); Class.forName("com.mysql.cj.jdbc.Driver");
row = (ResultSetImpl) statement.executeQuery(); // String url = "jdbc:mysql://pc-2ze6z2hxkug559q79.mysql.polardb.rds.aliyuncs.com:3306/passport";
String url = "jdbc:mysql://zhengzai.mysql.polardb.rds.aliyuncs.com:3306/passport";
Connection connection = DriverManager.getConnection(url, "readonly", "ZWDsf8Fy");
PreparedStatement statement = connection.prepareStatement(sqlCount);
ResultSetImpl row = (ResultSetImpl) statement.executeQuery();
row.first(); row.first();
int ct = row.getInt(1), pSize = 1000, num = 0, tl = 0, pl = ct; int ct = row.getInt(1), pSize = 1000, num = 0, tl = 0, pl = ct;
log.info("DM.execute.counts:{}/{},{} ~ {} ----------------------------> start.sql:{}", pl, num, pSize, tl, sql); log.info("DM.execute.counts:{}/{},{} ~ {} ----------------------------> start.sql:{}", pl, num, pSize, tl, sql);
while (ct > 0) { while (ct > 0) {
statement = connection.prepareStatement(sql); statement = connection.prepareStatement(sql);
int ls = (num == 0 && StringUtils.isNotEmpty(lS)) ? Integer.parseInt(lS) : pSize * num; int ls = pSize * num;
statement.setInt(1, ls); statement.setInt(1, ls);
statement.setInt(2, pSize); statement.setInt(2, pSize);
row = (ResultSetImpl) statement.executeQuery(); row = (ResultSetImpl) statement.executeQuery();
...@@ -98,5 +104,8 @@ public class DMRealNameProcessor extends DataMigrationProcessorService { ...@@ -98,5 +104,8 @@ public class DMRealNameProcessor extends DataMigrationProcessorService {
ct -= pSize; ct -= pSize;
} }
log.info("DM.execute.counts:{}/{},{} ~ {} ----------------------------> end", pl, num, pSize, tl); log.info("DM.execute.counts:{}/{},{} ~ {} ----------------------------> end", pl, num, pSize, tl);
row.close();
statement.close();
connection.close();
} }
} }
liquidnet-bus-service/liquidnet-service-adam/liquidnet-service-adam-impl/src/main/java/com/liquidnet/service/adam/service/processor/DMThirdPartsProcessor.java
View file @ 443f53b0
package com.liquidnet.service.adam.service.processor; package com.liquidnet.service.adam.service.processor;
import com.liquidnet.commons.lang.util.ServletUtils;
import com.liquidnet.service.adam.dto.vo.AdamThirdPartInfoVo; import com.liquidnet.service.adam.dto.vo.AdamThirdPartInfoVo;
import com.liquidnet.service.adam.entity.AdamThirdParty; import com.liquidnet.service.adam.entity.AdamThirdParty;
import com.liquidnet.service.adam.service.DataMigrationProcessorService; import com.liquidnet.service.adam.service.DataMigrationProcessorService;
...@@ -15,7 +14,9 @@ import org.springframework.data.mongodb.core.query.Query; ...@@ -15,7 +14,9 @@ import org.springframework.data.mongodb.core.query.Query;
import org.springframework.stereotype.Service; import org.springframework.stereotype.Service;
import org.springframework.util.CollectionUtils; import org.springframework.util.CollectionUtils;
import javax.servlet.http.HttpServletRequest; import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.util.ArrayList; import java.util.ArrayList;
import java.util.List; import java.util.List;
...@@ -27,11 +28,8 @@ public class DMThirdPartsProcessor extends DataMigrationProcessorService { ...@@ -27,11 +28,8 @@ public class DMThirdPartsProcessor extends DataMigrationProcessorService {
@SneakyThrows @SneakyThrows
@Override @Override
protected void dataProcessing() { public void dataProcessing(Integer dg, Integer lS, String incrDt, String flg, String flu) {
HttpServletRequest request = ServletUtils.getRequest(); if (StringUtils.isNotEmpty(flu)) {
String lS = request.getParameter(DM_LS), incrDt = request.getParameter(DM_INCR_DT),
flg = StringUtils.defaultIfBlank(request.getParameter(DM_FLG), "");
if (StringUtils.isNotEmpty(request.getParameter(DM_FLUSH))) {
log.info("DM.flush.third_party:{}", userMapper.executeForDM("adam_third_party")); log.info("DM.flush.third_party:{}", userMapper.executeForDM("adam_third_party"));
log.info("DM.flush.AdamThirdPartInfoVo:{}", mongoTemplate.remove(Query.query(Criteria.where("_id").exists(true)), AdamThirdPartInfoVo.class.getSimpleName()).getDeletedCount()); log.info("DM.flush.AdamThirdPartInfoVo:{}", mongoTemplate.remove(Query.query(Criteria.where("_id").exists(true)), AdamThirdPartInfoVo.class.getSimpleName()).getDeletedCount());
...@@ -40,6 +38,9 @@ public class DMThirdPartsProcessor extends DataMigrationProcessorService { ...@@ -40,6 +38,9 @@ public class DMThirdPartsProcessor extends DataMigrationProcessorService {
String sqlCount = "select count(1) from (\n" + String sqlCount = "select count(1) from (\n" +
" select row_number() over (partition by uid,type order by created_at desc) rn,id,uid,open_id,avatar,nickname,type,created_at,updated_at from user_third_parts where `status`=1 and delete_tag in (0,1) and open_id<>'' and uid>0 and created_at"+(StringUtils.isBlank(incrDt) ? "<" : ">=")+"curdate()\n" + " select row_number() over (partition by uid,type order by created_at desc) rn,id,uid,open_id,avatar,nickname,type,created_at,updated_at from user_third_parts where `status`=1 and delete_tag in (0,1) and open_id<>'' and uid>0 and created_at"+(StringUtils.isBlank(incrDt) ? "<" : ">=")+"curdate()\n" +
" ) t where t.rn=1"; " ) t where t.rn=1";
if (null != dg) {
sqlCount = sqlCount + " and id%" + lS + "=" + dg;
}
String field = "uid,open_id,avatar,nickname,type,created_at,updated_at"; String field = "uid,open_id,avatar,nickname,type,created_at,updated_at";
// 1-uid,2-open_id,3-avatar,4-nickname,5-type,6-created_at,7-updated_at // 1-uid,2-open_id,3-avatar,4-nickname,5-type,6-created_at,7-updated_at
String sql = sqlCount.replace("count(1)", field); String sql = sqlCount.replace("count(1)", field);
...@@ -49,15 +50,20 @@ public class DMThirdPartsProcessor extends DataMigrationProcessorService { ...@@ -49,15 +50,20 @@ public class DMThirdPartsProcessor extends DataMigrationProcessorService {
} }
sql += " order by id limit ?,?"; sql += " order by id limit ?,?";
statement = connection.prepareStatement(sqlCount); Class.forName("com.mysql.cj.jdbc.Driver");
row = (ResultSetImpl) statement.executeQuery(); // String url = "jdbc:mysql://pc-2ze6z2hxkug559q79.mysql.polardb.rds.aliyuncs.com:3306/passport";
String url = "jdbc:mysql://zhengzai.mysql.polardb.rds.aliyuncs.com:3306/passport";
Connection connection = DriverManager.getConnection(url, "readonly", "ZWDsf8Fy");
PreparedStatement statement = connection.prepareStatement(sqlCount);
ResultSetImpl row = (ResultSetImpl) statement.executeQuery();
row.first(); row.first();
int ct = row.getInt(1), pSize = 1000, num = 0, tl = 0, pl = ct; int ct = row.getInt(1), pSize = 1000, num = 0, tl = 0, pl = ct;
log.info("DM.execute.counts:{}/{},{} ~ {} ----------------------------> start.sql:{}", pl, num, pSize, tl, sql); log.info("DM.execute.counts:{}/{},{} ~ {} ----------------------------> start.sql:{}", pl, num, pSize, tl, sql);
while (ct > 0) { while (ct > 0) {
statement = connection.prepareStatement(sql); statement = connection.prepareStatement(sql);
int ls = (num == 0 && StringUtils.isNotEmpty(lS)) ? Integer.parseInt(lS) : pSize * num; int ls = pSize * num;
statement.setInt(1, ls); statement.setInt(1, ls);
statement.setInt(2, pSize); statement.setInt(2, pSize);
row = (ResultSetImpl) statement.executeQuery(); row = (ResultSetImpl) statement.executeQuery();
...@@ -103,5 +109,8 @@ public class DMThirdPartsProcessor extends DataMigrationProcessorService { ...@@ -103,5 +109,8 @@ public class DMThirdPartsProcessor extends DataMigrationProcessorService {
ct -= pSize; ct -= pSize;
} }
log.info("DM.execute.counts:{}/{},{} ~ {} ----------------------------> end", pl, num, pSize, tl); log.info("DM.execute.counts:{}/{},{} ~ {} ----------------------------> end", pl, num, pSize, tl);
row.close();
statement.close();
connection.close();
} }
} }
liquidnet-bus-service/liquidnet-service-adam/liquidnet-service-adam-impl/src/main/java/com/liquidnet/service/adam/service/processor/DMTracesInfoProcessor.java
View file @ 443f53b0
package com.liquidnet.service.adam.service.processor; package com.liquidnet.service.adam.service.processor;
import com.liquidnet.common.cache.redis.util.RedisUtil; import com.liquidnet.common.cache.redis.util.RedisUtil;
import com.liquidnet.commons.lang.util.ServletUtils;
import com.liquidnet.service.adam.dto.vo.AdamMemberPriceVo; import com.liquidnet.service.adam.dto.vo.AdamMemberPriceVo;
import com.liquidnet.service.adam.dto.vo.AdamMemberVo; import com.liquidnet.service.adam.dto.vo.AdamMemberVo;
import com.liquidnet.service.adam.dto.vo.AdamThirdPartInfoVo;
import com.liquidnet.service.adam.entity.AdamMember; import com.liquidnet.service.adam.entity.AdamMember;
import com.liquidnet.service.adam.entity.AdamMemberPrice; import com.liquidnet.service.adam.entity.AdamMemberPrice;
import com.liquidnet.service.adam.mapper.AdamMemberMapper; import com.liquidnet.service.adam.mapper.AdamMemberMapper;
...@@ -20,7 +18,9 @@ import org.springframework.data.mongodb.core.query.Criteria; ...@@ -20,7 +18,9 @@ import org.springframework.data.mongodb.core.query.Criteria;
import org.springframework.data.mongodb.core.query.Query; import org.springframework.data.mongodb.core.query.Query;
import org.springframework.stereotype.Service; import org.springframework.stereotype.Service;
import javax.servlet.http.HttpServletRequest; import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.SQLException; import java.sql.SQLException;
@Slf4j @Slf4j
...@@ -36,9 +36,8 @@ public class DMTracesInfoProcessor extends DataMigrationProcessorService { ...@@ -36,9 +36,8 @@ public class DMTracesInfoProcessor extends DataMigrationProcessorService {
AdamMemberPriceMapper memberPriceMapper; AdamMemberPriceMapper memberPriceMapper;
@Override @Override
protected void dataProcessing() { public void dataProcessing(Integer dg, Integer lS, String incrDt, String flg, String flu) {
HttpServletRequest request = ServletUtils.getRequest(); if (StringUtils.isNotEmpty(flu)) {
if (StringUtils.isNotEmpty(request.getParameter(DM_FLUSH))) {
log.info("DM.flush.member:{}", userMapper.executeForDM("adam_member")); log.info("DM.flush.member:{}", userMapper.executeForDM("adam_member"));
log.info("DM.flush.member_price:{}", userMapper.executeForDM("adam_member_price")); log.info("DM.flush.member_price:{}", userMapper.executeForDM("adam_member_price"));
...@@ -54,8 +53,14 @@ public class DMTracesInfoProcessor extends DataMigrationProcessorService { ...@@ -54,8 +53,14 @@ public class DMTracesInfoProcessor extends DataMigrationProcessorService {
try { try {
String sql = "select id,`name`,sub_title,icon,avatar,min_card_number,if(`type`='VIP',1,0),interests_detail,notes,created_at,updated_at,is_notice,vip_notice,vip_notice_title,need_question,display_agreement,status,limitation from vip_cards"; String sql = "select id,`name`,sub_title,icon,avatar,min_card_number,if(`type`='VIP',1,0),interests_detail,notes,created_at,updated_at,is_notice,vip_notice,vip_notice_title,need_question,display_agreement,status,limitation from vip_cards";
// 1-id,2-`name`,3-sub_title,4-icon,5-avatar,6-min_card_number,7-`type`,8-interests_detail,9-notes,10-created_at,11-updated_at,12-is_notice,13-vip_notice,14-vip_notice_title,15-need_question,16-display_agreement,17-status,18-limitation // 1-id,2-`name`,3-sub_title,4-icon,5-avatar,6-min_card_number,7-`type`,8-interests_detail,9-notes,10-created_at,11-updated_at,12-is_notice,13-vip_notice,14-vip_notice_title,15-need_question,16-display_agreement,17-status,18-limitation
statement = connection.prepareStatement(sql);
row = (ResultSetImpl) statement.executeQuery(); Class.forName("com.mysql.cj.jdbc.Driver");
// String url = "jdbc:mysql://pc-2ze6z2hxkug559q79.mysql.polardb.rds.aliyuncs.com:3306/passport";
String url = "jdbc:mysql://zhengzai.mysql.polardb.rds.aliyuncs.com:3306/passport";
Connection connection = DriverManager.getConnection(url, "readonly", "ZWDsf8Fy");
PreparedStatement statement = connection.prepareStatement(sql);
ResultSetImpl row = (ResultSetImpl) statement.executeQuery();
row.first(); row.first();
AdamMember o = new AdamMember(); AdamMember o = new AdamMember();
...@@ -83,6 +88,10 @@ public class DMTracesInfoProcessor extends DataMigrationProcessorService { ...@@ -83,6 +88,10 @@ public class DMTracesInfoProcessor extends DataMigrationProcessorService {
memberMapper.insert(o); memberMapper.insert(o);
mongoTemplate.insert(vo, AdamMemberVo.class.getSimpleName()); mongoTemplate.insert(vo, AdamMemberVo.class.getSimpleName());
row.close();
statement.close();
connection.close();
} catch (SQLException throwables) { } catch (SQLException throwables) {
throwables.printStackTrace(); throwables.printStackTrace();
} }
...@@ -93,8 +102,14 @@ public class DMTracesInfoProcessor extends DataMigrationProcessorService { ...@@ -93,8 +102,14 @@ public class DMTracesInfoProcessor extends DataMigrationProcessorService {
try { try {
String sql = "select id,vip_card_id,status,name,price,days,detail,created_at,updated_at from vip_card_prices"; String sql = "select id,vip_card_id,status,name,price,days,detail,created_at,updated_at from vip_card_prices";
// 1-id,2-vip_card_id,3-status,4-name,5-price,6-days,7-detail,8-created_at,9-updated_at // 1-id,2-vip_card_id,3-status,4-name,5-price,6-days,7-detail,8-created_at,9-updated_at
statement = connection.prepareStatement(sql);
row = (ResultSetImpl) statement.executeQuery(); Class.forName("com.mysql.cj.jdbc.Driver");
// String url = "jdbc:mysql://pc-2ze6z2hxkug559q79.mysql.polardb.rds.aliyuncs.com:3306/passport";
String url = "jdbc:mysql://zhengzai.mysql.polardb.rds.aliyuncs.com:3306/passport";
Connection connection = DriverManager.getConnection(url, "readonly", "ZWDsf8Fy");
PreparedStatement statement = connection.prepareStatement(sql);
ResultSetImpl row = (ResultSetImpl) statement.executeQuery();
row.first(); row.first();
AdamMemberPrice o = new AdamMemberPrice(); AdamMemberPrice o = new AdamMemberPrice();
...@@ -113,51 +128,12 @@ public class DMTracesInfoProcessor extends DataMigrationProcessorService { ...@@ -113,51 +128,12 @@ public class DMTracesInfoProcessor extends DataMigrationProcessorService {
memberPriceMapper.insert(o); memberPriceMapper.insert(o);
mongoTemplate.insert(vo, AdamMemberPriceVo.class.getSimpleName()); mongoTemplate.insert(vo, AdamMemberPriceVo.class.getSimpleName());
row.close();
statement.close();
connection.close();
} catch (SQLException throwables) { } catch (SQLException throwables) {
throwables.printStackTrace(); throwables.printStackTrace();
} }
} }
// @SneakyThrows
// @Override
// protected void dataProcessing(boolean flg) {
// String sqlCount = "select count(1) from addresses where 1=1";
// String field = "id,user_id,user_name,user_phone,province,city,county,user_address,defaultDelivery,created_at";
//// 1-id,2-user_id,3-user_name,4-user_phone,5-province,6-city,7-county,8-user_address,9-defaultDelivery,10-created_at
// String sql = sqlCount.replace("count(1)", field);
//
// if (flg) {
// sqlCount += " and user_id in ('2026628','544327','1997335','773650','809406')";
// sql += " and user_id in ('2026628','544327','1997335','773650','809406')";
// }
//
// sql += " limit ?,?";
//
// statement = connection.prepareStatement(sqlCount);
// resultSet = statement.executeQuery();
// resultSet.first();
// int ct = resultSet.getInt(1), pSize = 2, num = 0;
// log.info("DM数据迁移总数:{}", ct);
//
//
//
// while (ct > 0) {
// statement = connection.prepareStatement(sql);
// statement.setInt(1, pSize * num);
// statement.setInt(2, pSize);
// resultSet = statement.executeQuery();
// log.info("DM.execute.limit:[pNo:{},pSize:{}]", pSize * num, pSize);
// while (resultSet.next()) {
//
// }
//
//
// num++;
// ct -= pSize;
// }
// }
} }
liquidnet-bus-service/liquidnet-service-adam/liquidnet-service-adam-impl/src/main/java/com/liquidnet/service/adam/service/processor/DMUserInformationProcessor.java
View file @ 443f53b0
package com.liquidnet.service.adam.service.processor; package com.liquidnet.service.adam.service.processor;
import com.baomidou.mybatisplus.core.toolkit.Wrappers;
import com.liquidnet.commons.lang.util.DateUtil; import com.liquidnet.commons.lang.util.DateUtil;
import com.liquidnet.commons.lang.util.JsonUtils; import com.liquidnet.commons.lang.util.JsonUtils;
import com.liquidnet.commons.lang.util.ServletUtils;
import com.liquidnet.service.adam.dto.vo.AdamTagVo; import com.liquidnet.service.adam.dto.vo.AdamTagVo;
import com.liquidnet.service.adam.dto.vo.AdamUserInfoVo; import com.liquidnet.service.adam.dto.vo.AdamUserInfoVo;
import com.liquidnet.service.adam.entity.AdamUser; import com.liquidnet.service.adam.entity.AdamUser;
...@@ -22,8 +20,9 @@ import org.springframework.data.mongodb.core.query.Query; ...@@ -22,8 +20,9 @@ import org.springframework.data.mongodb.core.query.Query;
import org.springframework.stereotype.Service; import org.springframework.stereotype.Service;
import org.springframework.util.CollectionUtils; import org.springframework.util.CollectionUtils;
import javax.servlet.http.HttpServletRequest; import java.sql.Connection;
import java.sql.SQLException; import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.util.ArrayList; import java.util.ArrayList;
import java.util.HashMap; import java.util.HashMap;
import java.util.List; import java.util.List;
...@@ -37,8 +36,6 @@ public class DMUserInformationProcessor extends DataMigrationProcessorService { ...@@ -37,8 +36,6 @@ public class DMUserInformationProcessor extends DataMigrationProcessorService {
@Autowired @Autowired
IAdamUserInfoService adamUserInfoService; IAdamUserInfoService adamUserInfoService;
public static final String DM_FAILED_USER = "dm:failed:uid";
public static final String DM_EXISTS_USER = "dm:exists:uid";
public static Map<String, AdamTagVo> sexMap = new HashMap<>(); public static Map<String, AdamTagVo> sexMap = new HashMap<>();
static { static {
...@@ -49,18 +46,19 @@ public class DMUserInformationProcessor extends DataMigrationProcessorService { ...@@ -49,18 +46,19 @@ public class DMUserInformationProcessor extends DataMigrationProcessorService {
@SneakyThrows @SneakyThrows
@Override @Override
protected void dataProcessing() { public void dataProcessing(Integer dg, Integer lS, String incrDt, String flg, String flu) {
HttpServletRequest request = ServletUtils.getRequest(); if (StringUtils.isNotEmpty(flu)) {
String lS = request.getParameter(DM_LS), incrDt = request.getParameter(DM_INCR_DT),
flg = StringUtils.defaultIfBlank(request.getParameter(DM_FLG), "");
if (StringUtils.isNotEmpty(request.getParameter(DM_FLUSH))) {
log.info("DM.flush.user:{}", userMapper.executeForDM("adam_user")); log.info("DM.flush.user:{}", userMapper.executeForDM("adam_user"));
log.info("DM.flush.user_info:{}", userMapper.executeForDM("adam_user_info")); log.info("DM.flush.user_info:{}", userMapper.executeForDM("adam_user_info"));
log.info("DM.flush.AdamUserInfoVo:{}", mongoTemplate.remove(Query.query(Criteria.where("_id").exists(true)), AdamUserInfoVo.class.getSimpleName()).getDeletedCount()); log.info("DM.flush.AdamUserInfoVo:{}", mongoTemplate.remove(Query.query(Criteria.where("_id").exists(true)), AdamUserInfoVo.class.getSimpleName()).getDeletedCount());
} }
String sqlCount = "select count(1) from users where delete_tag in (0,1) and length(mobile)=11 and created_at" + (StringUtils.isBlank(incrDt) ? "<" : ">=") + "curdate()"; // String sqlCount = "select count(1) from users where delete_tag in (0,1) and length(mobile)=11 and created_at" + (StringUtils.isBlank(incrDt) ? "<" : ">=") + "curdate()";
String sqlCount = "select count(1) from users where delete_tag in (0,1) and created_at" + (StringUtils.isBlank(incrDt) ? "<" : ">=") + "curdate()";
if (null != dg) {
sqlCount = sqlCount + " and uid%" + lS + "=" + dg;
}
String field = "uid,mobile,username,sex,birthday,area,sign,face_url,back_url,created_at,updated_at,is_complete,rong_cloud_token,rong_cloud_tag"; String field = "uid,mobile,username,sex,birthday,area,sign,face_url,back_url,created_at,updated_at,is_complete,rong_cloud_token,rong_cloud_tag";
// 1-uid,2-mobile,3-username,4-sex,5-birthday,6-area,7-sign,8-face_url,9-back_url,10-created_at,11-updated_at,12-is_complete,13-rong_cloud_token,14-rong_cloud_tag // 1-uid,2-mobile,3-username,4-sex,5-birthday,6-area,7-sign,8-face_url,9-back_url,10-created_at,11-updated_at,12-is_complete,13-rong_cloud_token,14-rong_cloud_tag
String sql = sqlCount.replace("count(1)", field); String sql = sqlCount.replace("count(1)", field);
...@@ -70,15 +68,20 @@ public class DMUserInformationProcessor extends DataMigrationProcessorService { ...@@ -70,15 +68,20 @@ public class DMUserInformationProcessor extends DataMigrationProcessorService {
} }
sql += " order by uid limit ?,?"; sql += " order by uid limit ?,?";
statement = connection.prepareStatement(sqlCount); Class.forName("com.mysql.cj.jdbc.Driver");
row = (ResultSetImpl) statement.executeQuery(); // String url = "jdbc:mysql://pc-2ze6z2hxkug559q79.mysql.polardb.rds.aliyuncs.com:3306/passport";
String url = "jdbc:mysql://zhengzai.mysql.polardb.rds.aliyuncs.com:3306/passport";
Connection connection = DriverManager.getConnection(url, "readonly", "ZWDsf8Fy");
PreparedStatement statement = connection.prepareStatement(sqlCount);
ResultSetImpl row = (ResultSetImpl) statement.executeQuery();
row.first(); row.first();
int ct = row.getInt(1), pSize = 1000, num = 0, tl = 0, pl = ct; int ct = row.getInt(1), pSize = 1000, num = 0, tl = 0, pl = ct;
log.info("DM.execute.counts:{}/{},{} ~ {} ----------------------------> start.sql:{}", pl, num, pSize, tl, sql); log.info("DM.execute.counts:{}/{},{} ~ {} ----------------------------> start.sql:{}", pl, num, pSize, tl, sql);
while (ct > 0) { while (ct > 0) {
statement = connection.prepareStatement(sql); statement = connection.prepareStatement(sql);
int ls = (num == 0 && StringUtils.isNotEmpty(lS)) ? Integer.parseInt(lS) : pSize * num; int ls = pSize * num;
statement.setInt(1, ls); statement.setInt(1, ls);
statement.setInt(2, pSize); statement.setInt(2, pSize);
row = (ResultSetImpl) statement.executeQuery(); row = (ResultSetImpl) statement.executeQuery();
...@@ -88,58 +91,48 @@ public class DMUserInformationProcessor extends DataMigrationProcessorService { ...@@ -88,58 +91,48 @@ public class DMUserInformationProcessor extends DataMigrationProcessorService {
List<AdamUserInfo> userInfoList = new ArrayList<>(); List<AdamUserInfo> userInfoList = new ArrayList<>();
List<AdamUserInfoVo> vos = new ArrayList<>(); List<AdamUserInfoVo> vos = new ArrayList<>();
while (row.next()) { while (row.next()) {
try { AdamTagVo sex = sexMap.get(row.getString(4));
AdamTagVo sex = sexMap.get(row.getString(4)); String uid = row.getString(1);
String uid = row.getString(1); if (!reUidList.contains(uid)) {
if (!reUidList.contains(uid)) { AdamUser user = new AdamUser();
// if (userMapper.selectCount(Wrappers.lambdaQuery(AdamUser.class).eq(AdamUser::getUid, uid)) > 0) { user.setUid(uid);
// redisUtil.lSet(DM_EXISTS_USER, uid); user.setMobile(row.getString(2));
// } else { user.setCreatedAt(row.getLocalDateTime(10));
AdamUser user = new AdamUser(); user.setUpdatedAt(row.getLocalDateTime(11));
user.setUid(uid); user.setState(1);
user.setMobile(row.getString(2));
user.setCreatedAt(row.getLocalDateTime(10)); AdamUserInfo userInfo = new AdamUserInfo();
user.setUpdatedAt(row.getLocalDateTime(11)); userInfo.setUid(user.getUid());
user.setState(1); userInfo.setNickname(row.getString(3));
userInfo.setSex(JsonUtils.toJson(sex));
AdamUserInfo userInfo = new AdamUserInfo(); userInfo.setBirthday(row.getLocalDate(5));
userInfo.setUid(user.getUid()); userInfo.setArea(row.getString(6));
userInfo.setNickname(row.getString(3)); userInfo.setSignature(row.getString(7));
userInfo.setSex(JsonUtils.toJson(sex)); userInfo.setAvatar(row.getString(8));
userInfo.setBirthday(row.getLocalDate(5)); userInfo.setBackground(row.getString(9));
userInfo.setArea(row.getString(6)); userInfo.setQrCode("lN".concat(userInfo.getUid()).concat(RandomStringUtils.randomAlphanumeric(5).toUpperCase()));
userInfo.setSignature(row.getString(7)); userInfo.setRongCloudToken(row.getString(13));
userInfo.setAvatar(row.getString(8)); userInfo.setRongCloudTag(row.getInt(14));
userInfo.setBackground(row.getString(9));
userInfo.setQrCode("lN".concat(userInfo.getUid()).concat(RandomStringUtils.randomAlphanumeric(5).toUpperCase())); AdamUserInfoVo vo = AdamUserInfoVo.getNew();
userInfo.setRongCloudToken(row.getString(13)); vo.setUid(user.getUid());
userInfo.setRongCloudTag(row.getInt(14)); vo.setMobile(user.getMobile());
vo.setNickname(userInfo.getNickname());
AdamUserInfoVo vo = AdamUserInfoVo.getNew(); vo.setState(user.getState());
vo.setUid(user.getUid()); vo.setSex(sex);
vo.setMobile(user.getMobile()); vo.setBirthday(DateUtil.Formatter.yyyy_MM_dd.format(userInfo.getBirthday()));
vo.setNickname(userInfo.getNickname()); vo.setArea(userInfo.getArea());
vo.setState(user.getState()); vo.setSignature(userInfo.getSignature());
vo.setSex(sex); vo.setAvatar(userInfo.getAvatar());
vo.setBirthday(DateUtil.Formatter.yyyy_MM_dd.format(userInfo.getBirthday())); vo.setBackground(userInfo.getBackground());
vo.setArea(userInfo.getArea()); vo.setCreateAt(user.getCreatedAt());
vo.setSignature(userInfo.getSignature()); vo.setUpdatedAt(user.getUpdatedAt());
vo.setAvatar(userInfo.getAvatar()); vo.setIsComplete(row.getInt(12));
vo.setBackground(userInfo.getBackground()); vo.setQrCode(userInfo.getQrCode());
vo.setCreateAt(user.getCreatedAt());
vo.setUpdatedAt(user.getUpdatedAt()); userList.add(user);
vo.setIsComplete(row.getInt(12)); userInfoList.add(userInfo);
vo.setQrCode(userInfo.getQrCode()); vos.add(vo);
userList.add(user);
userInfoList.add(userInfo);
vos.add(vo);
// }
}
} catch (SQLException throwables) {
redisUtil.lSet(DM_FAILED_USER, row.getString(1));
log.error("DM数据处理异常[uid:{},mobile:{}],{} - {}", row.getString(1), row.getString(2),
throwables.getErrorCode(), throwables.getMessage());
} }
} }
log.info("DM.execute.limit {},{} - handle.counts:{}", ls, pSize, userList.size()); log.info("DM.execute.limit {},{} - handle.counts:{}", ls, pSize, userList.size());
...@@ -154,5 +147,8 @@ public class DMUserInformationProcessor extends DataMigrationProcessorService { ...@@ -154,5 +147,8 @@ public class DMUserInformationProcessor extends DataMigrationProcessorService {
ct -= pSize; ct -= pSize;
} }
log.info("DM.execute.counts:{}/{},{} ~ {} ----------------------------> end", pl, num, pSize, tl); log.info("DM.execute.counts:{}/{},{} ~ {} ----------------------------> end", pl, num, pSize, tl);
row.close();
statement.close();
connection.close();
} }
} }
liquidnet-bus-service/liquidnet-service-adam/liquidnet-service-adam-impl/src/main/java/com/liquidnet/service/adam/service/processor/DMUserMemberProcessor.java
View file @ 443f53b0
package com.liquidnet.service.adam.service.processor; package com.liquidnet.service.adam.service.processor;
import com.liquidnet.commons.lang.util.ServletUtils;
import com.liquidnet.service.adam.dto.vo.AdamUserMemberVo; import com.liquidnet.service.adam.dto.vo.AdamUserMemberVo;
import com.liquidnet.service.adam.entity.AdamUserMember; import com.liquidnet.service.adam.entity.AdamUserMember;
import com.liquidnet.service.adam.service.DataMigrationProcessorService; import com.liquidnet.service.adam.service.DataMigrationProcessorService;
...@@ -15,7 +14,9 @@ import org.springframework.data.mongodb.core.query.Query; ...@@ -15,7 +14,9 @@ import org.springframework.data.mongodb.core.query.Query;
import org.springframework.stereotype.Service; import org.springframework.stereotype.Service;
import org.springframework.util.CollectionUtils; import org.springframework.util.CollectionUtils;
import javax.servlet.http.HttpServletRequest; import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.util.ArrayList; import java.util.ArrayList;
import java.util.List; import java.util.List;
...@@ -27,17 +28,17 @@ public class DMUserMemberProcessor extends DataMigrationProcessorService { ...@@ -27,17 +28,17 @@ public class DMUserMemberProcessor extends DataMigrationProcessorService {
@SneakyThrows @SneakyThrows
@Override @Override
protected void dataProcessing() { public void dataProcessing(Integer dg, Integer lS, String incrDt, String flg, String flu) {
HttpServletRequest request = ServletUtils.getRequest(); if (StringUtils.isNotEmpty(flu)) {
String lS = request.getParameter(DM_LS), incrDt = request.getParameter(DM_INCR_DT),
flg = StringUtils.defaultIfBlank(request.getParameter(DM_FLG), "");
if (StringUtils.isNotEmpty(request.getParameter(DM_FLUSH))) {
log.info("DM.flush.user_member:{}", userMapper.executeForDM("adam_user_member")); log.info("DM.flush.user_member:{}", userMapper.executeForDM("adam_user_member"));
log.info("DM.flush.AdamUserMemberVo:{}", mongoTemplate.remove(Query.query(Criteria.where("_id").exists(true)), AdamUserMemberVo.class.getSimpleName()).getDeletedCount()); log.info("DM.flush.AdamUserMemberVo:{}", mongoTemplate.remove(Query.query(Criteria.where("_id").exists(true)), AdamUserMemberVo.class.getSimpleName()).getDeletedCount());
} }
String sqlCount = "select count(1) from user_cards where created_at" + (StringUtils.isBlank(incrDt) ? "<" : ">=") + "curdate()"; String sqlCount = "select count(1) from user_cards where created_at" + (StringUtils.isBlank(incrDt) ? "<" : ">=") + "curdate()";
if (null != dg) {
sqlCount = sqlCount + " and id%" + lS + "=" + dg;
}
String field = "uid,card_id,card_number,`status`,expired_at,created_at,updated_at"; String field = "uid,card_id,card_number,`status`,expired_at,created_at,updated_at";
// 1-uid,2-card_id,3-card_number,4-status,5-expired_at,6-created_at,7-updated_at // 1-uid,2-card_id,3-card_number,4-status,5-expired_at,6-created_at,7-updated_at
String sql = sqlCount.replace("count(1)", field); String sql = sqlCount.replace("count(1)", field);
...@@ -47,15 +48,20 @@ public class DMUserMemberProcessor extends DataMigrationProcessorService { ...@@ -47,15 +48,20 @@ public class DMUserMemberProcessor extends DataMigrationProcessorService {
} }
sql += " order by id limit ?,?"; sql += " order by id limit ?,?";
statement = connection.prepareStatement(sqlCount); Class.forName("com.mysql.cj.jdbc.Driver");
row = (ResultSetImpl) statement.executeQuery(); // String url = "jdbc:mysql://pc-2ze6z2hxkug559q79.mysql.polardb.rds.aliyuncs.com:3306/passport";
String url = "jdbc:mysql://zhengzai.mysql.polardb.rds.aliyuncs.com:3306/passport";
Connection connection = DriverManager.getConnection(url, "readonly", "ZWDsf8Fy");
PreparedStatement statement = connection.prepareStatement(sqlCount);
ResultSetImpl row = (ResultSetImpl) statement.executeQuery();
row.first(); row.first();
int ct = row.getInt(1), pSize = 1000, num = 0, tl = 0, pl = ct; int ct = row.getInt(1), pSize = 1000, num = 0, tl = 0, pl = ct;
log.info("DM.execute.counts:{}/{},{} ~ {} ----------------------------> start.sql:{}", pl, num, pSize, tl, sql); log.info("DM.execute.counts:{}/{},{} ~ {} ----------------------------> start.sql:{}", pl, num, pSize, tl, sql);
while (ct > 0) { while (ct > 0) {
statement = connection.prepareStatement(sql); statement = connection.prepareStatement(sql);
int ls = (num == 0 && StringUtils.isNotEmpty(lS)) ? Integer.parseInt(lS) : pSize * num; int ls = pSize * num;
statement.setInt(1, ls); statement.setInt(1, ls);
statement.setInt(2, pSize); statement.setInt(2, pSize);
row = (ResultSetImpl) statement.executeQuery(); row = (ResultSetImpl) statement.executeQuery();
...@@ -99,5 +105,8 @@ public class DMUserMemberProcessor extends DataMigrationProcessorService { ...@@ -99,5 +105,8 @@ public class DMUserMemberProcessor extends DataMigrationProcessorService {
ct -= pSize; ct -= pSize;
} }
log.info("DM.execute.counts:{}/{},{} ~ {} ----------------------------> end", pl, num, pSize, tl); log.info("DM.execute.counts:{}/{},{} ~ {} ----------------------------> end", pl, num, pSize, tl);
row.close();
statement.close();
connection.close();
} }
} }
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment