背景
我们在流量对比的过程中，对比规则是不同的，可能需要：
- 全字段对比；
- 忽略空值对比；
- 排序后对比；
- 指定字段对比；
- 忽略字段对比；
- 字段加工对比；
如果要实现脚本对比，需要解决的难点：嵌套对象如何获取参数对应值（扁平化：xx.xx获取）；
扁平化对象
public class FlowData implements Serializable {
private static final long serialVersionUID = -8677073254096521327L;
private String dataId;
private final Map<String, Object> data = Maps.newHashMap();
public FlowData() {
this.dataId = null;
}
public FlowData(String dataId) {
this.dataId = dataId;
}
public FlowData(FlowData flowData) {
this.dataId = flowData.dataId;
this.data.putAll(flowData.data);
}
public FlowData(String dataId, FlowData flowData) {
this.dataId = dataId;
this.data.putAll(flowData.data);
}
public String getDataId() {
return dataId;
}
public void setDataId(String dataId) {
this.dataId = dataId;
}
/**
* 按照路径找到相应值
*/
public Object getData(String fieldPath) {
if (StringUtils.isEmpty(fieldPath)) {
return null;
}
String[] splitFields = fieldPath.split("\\.");
Object curData = data;
for (String splitField : splitFields) {
if (Objects.isNull(curData)) {
return null;
}
if (curData instanceof Map) {
curData = ((Map<?, ?>) curData).get(splitField);
continue;
}
if (curData instanceof List && StringUtils.isNumeric(splitField)
&& Integer.parseInt(splitField) < ((List<?>) curData).size()) {
curData = ((List<?>) curData).get(Integer.parseInt(splitField));
continue;
}
return null;
}
return curData;
}
public Map<String, Object> getData() {
return data;
}
public void putData(String fieldPath, Object data) {
Map<String, Object> updateMap = Maps.newHashMap();
updateMap.put(fieldPath, data);
putData(updateMap);
}
public void putData(Map<String, Object> updateMap) {
updateMap(data, ParamUtils.foldData(updateMap));
}
public void putData(FlowData updateData) {
updateMap(data, updateData.data);
}
public synchronized void removeData(String fieldPath) {
if (StringUtils.isEmpty(fieldPath)) {
return;
}
String[] splitFields = fieldPath.split("\\.");
Object curData = data;
for (int i = 0; i < splitFields.length - 1; i++) {
if (Objects.isNull(curData)) {
return;
}
String splitField = splitFields[i];
if (curData instanceof Map) {
curData = ((Map<?, ?>) curData).get(splitField);
continue;
}
if (curData instanceof List && StringUtils.isNumeric(splitField)
&& Integer.parseInt(splitField) < ((List<?>) curData).size()) {
curData = ((List<?>) curData).get(Integer.parseInt(splitField));
continue;
}
return;
}
String lastSplitField = splitFields[splitFields.length - 1];
if (curData instanceof Map) {
((Map<?, ?>) curData).remove(lastSplitField);
}
if (curData instanceof List && StringUtils.isNumeric(lastSplitField)
&& Integer.parseInt(lastSplitField) < ((List<?>) curData).size()) {
((List<?>) curData).remove(Integer.parseInt(lastSplitField));
}
}
private synchronized void updateMap(Map<String, Object> mainMap, Map<String, Object> updateMap) {
if (Objects.isNull(mainMap)) {
return;
}
for (Entry<String, Object> entry : updateMap.entrySet()) {
String key = entry.getKey();
Object updateValue = entry.getValue();
if (mainMap.containsKey(key)) {
Object mainValue = mainMap.get(key);
if (mainValue instanceof Map && updateValue instanceof Map) {
updateMap((Map<String, Object>) mainValue, (Map<String, Object>) updateValue);
continue;
}
if (mainValue instanceof List && updateValue instanceof List) {
mainMap.put(key, updateValue);
continue;
}
}
mainMap.put(key, updateValue);
}
}
/**
* 找到最后一个有效路径
*/
private String getValidPath(String fieldPath) {
if (Objects.isNull(data)) {
return null;
}
List<String> result = Lists.newArrayList();
String[] splitFields = fieldPath.split("\\.");
Object curData = data;
for (String field : splitFields) {
Object nextData = null;
if (curData instanceof Map) {
nextData = ((Map<?, ?>) curData).get(field);
}
if (curData instanceof List && StringUtils.isNumeric(field)) {
nextData = ((List<?>) curData).get(Integer.parseInt(field));
}
if (Objects.isNull(nextData)) {
break;
}
result.add(field);
curData = nextData;
}
if (CollectionUtils.isEmpty(result)) {
return null;
}
return StringUtils.join(result, ".");
}
@Override
public String toString() {
return ObjectMapperUtils.toJSON(data);
}
}
工具类：
public class ParamUtils {
public static String convertField(String field) {
return CaseFormat.LOWER_CAMEL.to(CaseFormat.LOWER_UNDERSCORE, field);
}
public static Map<String, Object> foldData(Map<String, Object> data) {
Map<String, Object> result = Maps.newHashMap();
if (MapUtils.isEmpty(data)) {
return result;
}
data.forEach((k, v) -> {
String[] splitKey = k.split("\\.");
int index = 0;
Map<String, Object> indexMap = result;
while (index < splitKey.length - 1) {
String subKey = splitKey[index];
indexMap = foldGetFromMap(indexMap, subKey);
index++;
}
indexMap.put(splitKey[index], v);
});
return convertDataArray(result);
}
private static Map<String, Object> foldGetFromMap(Map<String, Object> map, String key) {
Object subNode = map.get(key);
if (Objects.nonNull(subNode) && subNode instanceof Map) {
return (Map<String, Object>) subNode;
}
Map<String, Object> result = Maps.newHashMap();
map.put(key, result);
return result;
}
private static Map<String, Object> convertDataArray(Map<String, Object> map) {
Map<String, Object> result = Maps.newHashMap();
if (MapUtils.isEmpty(map)) {
return map;
}
for (Map.Entry<String, Object> entry : map.entrySet()) {
String k = entry.getKey();
Object v = entry.getValue();
if (!(v instanceof Map)) {
result.put(k, v);
continue;
}
Map<String, Object> valueMap = convertDataArray((Map<String, Object>) v);
if (valueMap.keySet().stream().allMatch(StringUtils::isNumeric)) {
result.put(k, valueMap.values());
continue;
}
result.put(k, valueMap);
}
return result;
}
/**
* 打平map结构，获取入参元数据
*
* @param prefix 打平的前缀
* @param data {"name":"tom","age":123}
* @return 打平后的结构{"name":"string","age":"int"}
*/
public static Map<String, String> unfoldSampleData(String prefix, Map<String, Object> data) {
Map<String, String> result = Maps.newHashMap();
if (MapUtils.isEmpty(data)) {
return result;
}
data.forEach((k, v) -> {
String unfoldKey;
if (StringUtils.isNotBlank(prefix)) {
unfoldKey = prefix + "." + k;
} else {
unfoldKey = k;
}
if (v instanceof Map) {
result.put(unfoldKey, "MAP");
result.putAll(unfoldSampleData(unfoldKey, (Map<String, Object>) v));
return;
}
if (v instanceof Collection) {
result.put(unfoldKey, "LIST");
result.putAll(unfoldSampleData(unfoldKey, convertToMap((Collection) v)));
return;
}
result.put(unfoldKey, v.getClass().getSimpleName());
});
return result;
}
/**
* 打平map结构，获取入参元数据
*
* @param prefix 打平的前缀
* @param data {"name":"string","age":"int"}
* @return 打平后的结构
*/
public static Map<String, Object> unfoldData(String prefix, Map<String, Object> data) {
Map<String, Object> result = Maps.newHashMap();
if (MapUtils.isEmpty(data)) {
return result;
}
data.forEach((k, v) -> {
String unfoldKey;
if (StringUtils.isNotBlank(prefix)) {
unfoldKey = prefix + "." + k;
} else {
unfoldKey = k;
}
if (v instanceof Map) {
result.put(unfoldKey, "MAP");
result.putAll(unfoldData(unfoldKey, (Map<String, Object>) v));
return;
}
if (v instanceof Collection) {
result.put(unfoldKey, "LIST");
result.putAll(unfoldData(unfoldKey, convertToMap((Collection) v)));
return;
}
result.put(unfoldKey, v);
});
return result;
}
public static Map<String, Object> unfoldAllPathData(String prefix, Map<String, Object> data) {
Map<String, Object> result = Maps.newHashMap();
if (MapUtils.isEmpty(data)) {
return result;
}
data.forEach((k, v) -> {
String unfoldKey;
if (StringUtils.isNotBlank(prefix)) {
unfoldKey = prefix + "." + k;
} else {
unfoldKey = k;
}
if (v instanceof Map) {
result.put(unfoldKey, v);
result.putAll(unfoldAllPathData(unfoldKey, (Map<String, Object>) v));
return;
}
if (v instanceof Collection) {
result.put(unfoldKey, v);
result.putAll(unfoldAllPathData(unfoldKey, convertToMap((Collection) v)));
return;
}
result.put(unfoldKey, v);
});
return result;
}
private static Map<String, Object> convertToMap(Collection<?> collection) {
Map<String, Object> result = Maps.newHashMap();
if (CollectionUtils.isEmpty(collection)) {
return result;
}
Iterator<?> iterator = collection.iterator();
int i = 0;
while (iterator.hasNext()) {
result.put(String.valueOf(i), iterator.next());
i++;
}
return result;
}
//double?
public static String getFiledType(Object o) {
if (Objects.isNull(o)) { //校验schema，不能有null
throw new RuntimeException("jsonSchema配置有误，不能有null");
}
if (o instanceof Boolean) {
return "Boolean";
}
if (o instanceof Integer || o instanceof Long) {
return "Long";
}
if (o instanceof String) {
if (StringUtils.isNumeric((String) o)) {
return "Long";
}
return "String";
}
return null;
}
}
表达式构造器：
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Set;
import org.apache.commons.lang3.StringUtils;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import com.google.common.collect.Sets;
import lombok.AllArgsConstructor;
import lombok.Builder;
import lombok.Getter;
import lombok.Setter;
import lombok.ToString;
public class ExpressBuildHandler {
private static final Set<String> DEFAULT_KEYWORD = Sets.newHashSet("null", "true", "false");
public static final int EXPRESS_PARAM = 1;
public static final int EXPRESS_METHOD = 2;
public static ExpressInfo convertExpress(String express, FlowData dataInfo) throws Exception {
ExpressInfo expressInfo = new ExpressInfo();
//根据字符位置，解析总的表达式，获取到方法名、参数名
List<ExpressVar> expressVarList = getVarFromExpress(express);
//重写表达式（替换表达式中的.为_）
expressInfo.setExpress(replaceSeparatorForExpress(express, expressVarList));
//构建表达式入参(将需要的入参，从流参数中解析出来)
expressInfo.setParamMap(getParamMap(expressVarList, dataInfo));
//构建脚本函数
return expressInfo;
}
public static Map<String, Object> getParamMap(List<ExpressVar> expressVarList, FlowData dataInfo) {
Map<String, Object> paramMap = Maps.newHashMap();
for (ExpressVar entry : expressVarList) {
if (entry.getType() == EXPRESS_PARAM) {
String valKey = replaceSeparatorForKey(entry.getContext());
if (!DEFAULT_KEYWORD.contains(valKey)) {
paramMap.put(valKey, getExpressReplaceValue(entry.getContext(), dataInfo));
}
}
}
return paramMap;
}
/**
* 重新表达式，替换分隔符
*/
private static String replaceSeparatorForExpress(String express, List<ExpressVar> expressVarList) {
int appendStart = 0;
StringBuilder resultBuilder = new StringBuilder();
//处理表达式
for (ExpressVar entry : expressVarList) {
//替换
if (entry.getType() == EXPRESS_PARAM) {
String valKey = replaceSeparatorForKey(entry.getContext());
Integer start = entry.getStart();
resultBuilder.append(express, appendStart, start);
resultBuilder.append(valKey);
appendStart = entry.getEnd();
}
}
if (appendStart < express.length()) {
resultBuilder.append(express, appendStart, express.length());
}
return resultBuilder.toString();
}
private static String replaceSeparatorForKey(String paramFieldName) {
return paramFieldName.replace(".", "_");
}
private static Object getExpressReplaceValue(String field, FlowData dataInfo) {
String stringField = field.trim();
if (DEFAULT_KEYWORD.contains(stringField)) {
return stringField;
}
Object result = dataInfo.getData(stringField);
if (Objects.isNull(result)) {
return null;
}
return result;
}
private static List<ExpressVar> getVarFromExpress(String express) {
List<ExpressVar> result = Lists.newArrayList();
if (StringUtils.isBlank(express)) {
return result;
}
boolean strScope = false;
boolean varScope = false;
int varStart = 0;
int varEnd = 0;
char currentChar;
char preChar = 0;
char[] charList = express.toCharArray();
for (int i = 0; i < charList.length; i++) {
currentChar = charList[i];
if (i - 1 >= 0) {
preChar = charList[i - 1];
}
//遇到引号变换字符串scope
if (currentChar == '"' && preChar != '\\') {
strScope = !strScope;
}
//非字符串且非变量scope,遇到字母进入变量scope
if (!strScope && !varScope && Character.isLetter(currentChar)) {
varScope = true;
varStart = i;
varEnd = i + 1;
continue;
}
//当前处于字符串scope,进行累加或结束
if (varScope) {
if (Character.isLetterOrDigit(currentChar) || currentChar == '_' || currentChar == '.') {
varEnd++;
continue;
}
varScope = false;
if (currentChar == '(') {
result.add(new ExpressVar(varStart, varEnd, 2, express.substring(varStart, varEnd)));
} else {
result.add(new ExpressVar(varStart, varEnd, 1, express.substring(varStart, varEnd)));
}
}
}
if (strScope) {
throw new RuntimeException("字符串不完整");
}
if (varScope) {
result.add(new ExpressVar(varStart, varEnd, 1, express.substring(varStart, varEnd)));
}
return result;
}
@Getter
@Setter
@AllArgsConstructor
public static class ExpressVar {
private Integer start;
private Integer end;
private Integer type;
private String context;
}
@Getter
@Builder
@ToString
public static class ExpressFunction {
private String function;
private Object classInstance;
private String methodName;
private Class<?>[] paramTypes;
private String udfFunctionImpl;
}
@Getter
@Setter
public static class ExpressInfo {
private String express;
private Map<String, Object> paramMap;
private Map<String, ExpressFunction> functionMap;
}
}
表达式执行器
import java.util.Collection;
import com.ql.util.express.DefaultContext;
import com.ql.util.express.ExpressRunner;
public class ExpressExecuteHandler {
private static final ExpressExecuteHandler INSTANCE = new ExpressExecuteHandler();
public static final ExpressRunner RUNNER = new ExpressRunner();
static {
try {
//序列化对比
RUNNER.addFunctionOfClassMethod("toJson", InnerMethod.class.getName(), "json", new Class<?>[] {Object.class}, null);
//有序序列化对比
RUNNER.addFunctionOfClassMethod("toJsonIgnoreNull", InnerMethod.class.getName(), "jsonIgnoreNull", new Class<?>[] {Object.class}, null);
//集合有序
RUNNER.addFunctionOfClassMethod("sort", InnerMethod.class.getName(), "sort", new Class<?>[] {Collection.class}, null);
RUNNER.addFunctionOfClassMethod("sortEx", InnerMethod.class.getName(), "sortEx", new Class<?>[] {Collection.class, String.class}, null);
//忽略某些字段
RUNNER.addFunctionOfClassMethod("ignoreList", InnerMethod.class.getName(), "ignoreList",
new Class<?>[] {Collection.class, String[].class}, null);
//命中某些字段
RUNNER.addFunctionOfClassMethod("hitList", InnerMethod.class.getName(), "hitList",
new Class<?>[] {Collection.class, String[].class}, null);
//获取集合长度
RUNNER.addFunctionOfClassMethod("size", InnerMethod.class.getName(), "size", new Class<?>[] {Object.class}, null);
} catch (Exception e) {
throw new RuntimeException(e);
}
}
public static ExpressExecuteHandler getInstance() {
return INSTANCE;
}
/**
* 执行脚本语言
*
* @param express 脚本表达式
* @param defaultContext 环境变量（参数）
*/
private Object execute(String express, DefaultContext<String, Object> defaultContext) throws Exception {
return RUNNER.execute(express, defaultContext, null, true, false);
}
/**
* 执行函数
*/
public Object calculateForResult(ExpressBuildHandler.ExpressInfo expressInfo) throws Exception {
StringBuilder finalExpress = new StringBuilder();
finalExpress.append(expressInfo.getExpress());
DefaultContext<String, Object> defaultContext = new DefaultContext<>();
defaultContext.putAll(expressInfo.getParamMap());
return this.execute(finalExpress.toString(), defaultContext);
}
}
内置函数
@Slf4j
public class InnerMethod {
private static ObjectMapper objectMapper = new ObjectMapper();
static {
objectMapper.setSerializationInclusion(JsonInclude.Include.NON_EMPTY);
}
//toJson策略
public static String json(Object obj) {
return ObjectMapperUtils.toJSON(obj);
}
public static int size(Object obj) {
if (obj == null) {
return 0;
} else if (obj.getClass().isArray()) {
return ((Object[]) obj).length;
} else if (obj instanceof Collection) {
return ((Collection) obj).size();
}
return 1;
}
//忽略null-序列化
public static String jsonIgnoreNull(Object obj) {
try {
//处理单集合场景
if (obj instanceof Collection) {
List arr = (List) obj;
obj = arr.stream()
.filter(r -> !Objects.isNull(r))
.filter(r -> r instanceof String ? StringUtils.isNotEmpty((String) r) : true)
.collect(Collectors.toList());
}
if (obj.getClass().isArray()) {
Object[] arr = (Object[]) obj;
obj = Arrays.stream(arr)
.filter(r -> !Objects.isNull(r))
.filter(r -> r instanceof String ? StringUtils.isNotEmpty((String) r) : true)
.collect(Collectors.toList());
}
return objectMapper.writeValueAsString(obj);
} catch (JsonProcessingException e) {
throw new RuntimeException(e);
}
}
//基础集合的排序
public static List sort(Collection<Object> targets) {
if (CollectionUtils.isEmpty(targets)) {
return new ArrayList();
}
List<Comparable> collect = targets.stream().map(r -> (Comparable) r).collect(Collectors.toList());
return collect.stream().sorted().collect(Collectors.toList());
}
//普通对象的排序
public static List sortEx(Collection<Object> targets, String sortField) {
log.info("targets:{}", targets);
if (CollectionUtils.isEmpty(targets)) {
return new ArrayList<>();
}
Collection<Map> maps = ObjectMapperUtils.fromJSON(ObjectMapperUtils.toJSON(targets), List.class, Map.class);
String field = sortField.replace("_", ".");
return maps.stream().sorted(Comparator.comparing(r -> {
FlowData flowData = new FlowData();
flowData.putData(r);
return String.valueOf(flowData.getData(field));
})).collect(Collectors.toList());
}
public static List ignoreList(Collection<Object> targets, String[] params) {
if (CollectionUtils.isEmpty(targets)) {
return Lists.newArrayList();
}
Collection<Map> maps = ObjectMapperUtils.fromJSON(ObjectMapperUtils.toJSON(targets), List.class, Map.class);
return maps.stream().map(r -> {
FlowData flowData = new FlowData();
if (r instanceof Map) {
flowData.putData(r);
} else {
flowData.putData(ObjectMapperUtils.value(r, Map.class));
}
for (int i = 0; i < params.length; i++) {
String param = params[i].replace("_", ".");
flowData.removeData(param);
}
return flowData.getData();
}).collect(Collectors.toList());
}
public static List hitList(Collection<Object> targets, String[] params) {
if (CollectionUtils.isEmpty(targets)) {
return Lists.newArrayList();
}
Collection<Map> maps = ObjectMapperUtils.fromJSON(ObjectMapperUtils.toJSON(targets), List.class, Map.class);
return maps.stream().map(r -> {
FlowData flowData = new FlowData();
if (r instanceof Map) {
flowData.putData(r);
} else {
flowData.putData(ObjectMapperUtils.value(r, Map.class));
}
FlowData targetData = new FlowData();
for (int i = 0; i < params.length; i++) {
String param = params[i].replace("_", ".");
targetData.putData(param, flowData.getData(param));
}
return targetData.getData();
}).collect(Collectors.toList());
}
}
测试使用
public static void main(String[] args) throws Exception {
List<Address> d1 = Lists.newArrayList();
d1.add(Address.builder().province(Province.builder().id(4L).name(null).build()).city("洛阳").area("白帝小区").phone("13617172120")
.build());
d1.add(Address.builder().province(Province.builder().id(2L).name("河北").build()).city("邯郸").area("邯郸小区").phone("19917172122")
.build());
d1.add(Address.builder().province(Province.builder().id(3L).name("山东").build()).city("德州").area("德州小区").phone("18817172121")
.build());
FlowData flowData = new FlowData();
flowData.putData("this", d1);
ExpressBuildHandler.ExpressInfo e1 =
ExpressBuildHandler.convertExpress("sortEx(address,'province.id')", flowData);
Object o2 = ExpressExecuteHandler.getInstance().calculateForResult(e1);
System.out.println(o2);
}
