如何对第三方相同请求进行筛选过滤

问题背景

公司内部多个系统共用一套用户体系库，对外(钉钉)我们是两个客户身份(这里是根据系统来的)，例如当第三方服务向我们发起用户同步请求：是一个更新用户操作，它会同时发送一个 delete 和 insert 请求，这两个请求几乎是并发进来的，实际上应该是先发起的delete 再 insert， 实际情况可能和网络延迟也有关系，此时在我们系统中就无法保证这两个请求的顺序执行，即先 delete 处理完之后 再进行 insert 的数据处理(正常流程)，又或者直接把一定时间内同一个用户的 delete 和 insert 操作合并为一个update操作(本质就是更新操作)。

还有一种情况是：第三方系统中添加或者 删除一个用户时，会以两个客户的身份去发送两个相同的用户同步请求，但同一个用户在我们系统内用户数据只有一份，对应的接口肯定也都是同一个，即相同的添加接口会在一瞬间被调用两次，删除即使执行两次的话也没什么问题，问题是添加 即使在添加前判断了用户账号是否存在 并发过来的情况下还是避免不了一些脏数据的产生，加锁的话对整体影响又特别大。

处理思路

根据userId（账号）为每个请求分配一个房间（单独的线程），如果是第一次进来那么就new一个房间（也就是类，里边会有一个单独的线程处理这个用户的行为），后边一定时间内相同的 userId 进来会找到对应已存在的房间，当设置的时间窗口到了之后，判断当前userId的同步行为有哪些，如果有 insert 和 delete，那么直接转为 update 操作。如果是两个insert行为，那么最后就只调用一次insert服务，如果是两个delete行为，那么就只调用一个delete服务。

注意事项

时间窗口的设定，如果时间设置过短，属于同一个操作的请求因为网络波动 请求到接口的时间会有一定间隔，如果你设置的时间间隔小于等待的时间，还是会把本就属于同一批次的操作 多次处理

测试过程：刚开始时间设置的1500ms，也就是当第一个userId进来后，等待1.5秒后根据这段时间内收集到的用户行为再去真正的处理，后来在测试中发现有些本就属于同一批次的请求还是会被处理多次，也就是时间调小了，改成2000ms，测试还是发现同样的问题。最后：采取的是根据最近一个的userId请求的时间 等待1500ms，即相同的userId的请求进来后 在当前时间再重新计算等待1500ms，时间到了之后没有发现新的用户行为即算是一个批次结束

ps：可以创建一个单独的服务来负责对请求进行合理的处理分发，处理之后再去调用对应的业务系统服务

代码实现

定义操作行为枚举

public enum OperationEnum {INSERT("insert"),DELETE("delete"),;private final String value;OperationEnum(String operation) {this.value = operation;}public String getValue() {return value;}
}

定义每个用户所属的房间，房间内存储用户的多个行为（insert、delete）：

public class ActionRoomBean {//用于保存有效事件数据 <insert or delete, data>private Map<String, JSONObject> actionDataMap = new HashMap<>();private String userId;//真正负责处理事件的线程private DispatchTask dispatchTask;/*** 定义操作方法 排队接收* @param action 请求动作：insert 或者 delete* @param data 请求参数*/public void addAction(String action, JSONObject data) {//有新请求进来后 计数器 + 1if(dispatchTask != null){dispatchTask.getIncrementAndGet();}//如果包含直接跳出if (actionDataMap.containsKey(action)) {return;}actionDataMap.put(action, data);}public ActionRoomBean() {}/*** 有参构造* @param userId 用户账号* @param actionDataMap 操作类型，请求参数*/public ActionRoomBean(String userId, Map<String, JSONObject> actionDataMap) {this.actionDataMap = actionDataMap;this.userId = userId;}/*** 创建完这个类的实例后，要先调用startManager方法 启动线程*/public void startManager() {dispatchTask = new DispatchTask(userId, actionDataMap);new Thread(dispatchTask).start();}
}

房间内真正的执行者（子线程）：

public class DispatchTask implements Runnable {//等待的时间窗口private static long sleepTime = 1500;//计数器，用户有新的行为之后 +1，用来控制是否继续等待（sleep）private final AtomicInteger count = new AtomicInteger(0);//用于保存有效事件数据 <insert or delete, data>，与 ActionRoomBean中的 actionDataMap 指向的是同一个地址Map<String, JSONObject> actionDataMap;//用户账号String userId;/*** 有参构造*/public DispatchTask(String userId, Map<String, JSONObject> dataLib) {this.userId = userId;this.actionDataMap = dataLib;}@Overridepublic void run() {try {//线程等待前的数量和休眠后被唤醒的数量做对比，如果不相等说明休眠时间内有新的用户行为，则进入循环继续sleepint afterCount = 0;while (afterCount == 0 || afterCount != count.get()){//每休眠一次 +1，如果下次循环的值与 +1之后的afterCount相等，说明时间窗口内没有新的行为，则不循环afterCount = count.incrementAndGet();Thread.sleep(sleepTime);}} catch (InterruptedException e) {throw new RuntimeException(e);}try {String url = RestTemplateUtil.DD_READING_API_URL;JSONObject param = null;// 只有添加操作if (actionDataMap.containsKey(OperationEnum.INSERT.getValue()) && actionDataMap.size() == 1) {url += "/nc/eduUserInsert";param = actionDataMap.get(OperationEnum.INSERT.getValue());if (param != null) {RestTemplateUtil.postForObject(url, param.toJSONString());}} else if (actionDataMap.containsKey(OperationEnum.DELETE.getValue()) && actionDataMap.size() == 1) {//只有删除操作url += "/nc/eduUserDelete";param = actionDataMap.get(OperationEnum.DELETE.getValue());if (param != null) {RestTemplateUtil.postForObject(url, param.toJSONString());}} else if (actionDataMap.containsKey(OperationEnum.INSERT.getValue()) && actionDataMap.containsKey(OperationEnum.DELETE.getValue())) {//既有添加又有删除，就是更新处理url += "/nc/eduUserUpdate";param = actionDataMap.get(OperationEnum.INSERT.getValue());if (param != null) {RestTemplateUtil.postForObject(url, param.toJSONString());}}} finally {//最后从全局变量中删除userIdDispatchController.closeRoom(userId);}}/*** 计数器 + 1*/public Integer getIncrementAndGet() {return count.incrementAndGet();}
}

控制器：

@RestController
@RequestMapping(value = "/api/dd")
public class DispatchController {//全局map，记录当前有多少个用户正在被处理中private final static Map<String, ActionRoomBean> allMap = new ConcurrentHashMap<>();//简单的配置的密钥，用于接口的身份校验@Value("${url.secret}")private String secret;/*** insert和 delete 操作都会进入这个接口，用 operation 区分当前是什么操作*/@PostMapping(value = "/dispatch")public Result dispatch(@RequestBody JSONObject jsonObject,@RequestHeader(value = "secret") String secret){//进行简单的接口身份校验if(!Objects.equals(secret, this.secret)){return Result.generateError("secret eroor");}String userId = jsonObject.getString("userId");//operation = insert 或者 deleteString operation = jsonObject.getString("operation");if(EmptyUtil.isNotEmpty(userId) && EmptyUtil.isNotEmpty(operation)){//调用进入房间的方法unboltRoom(userId, operation, jsonObject);}return Result.generateSuccess();}/*** 为每个userId创建一个实例（房间）* 这里决定了是创建一个新的房间还是进入到已有的房间中*/private void unboltRoom(String userId, String operation, JSONObject jsonObject) {//加锁处理，由于真正的执行是在子线程中 所以加锁对整体性能影响也不是很大//主要是避免：同一个userId创建了多个实例，即使map中key不可重复，也会造成请求丢失//例如：同一个userId进来insert和delete请求各一个，并发不加锁的情况下就有可能创建了两个实例synchronized (this) {ActionRoomBean room = allMap.get(userId);//如果全局map中没有，说明是这个userId是第一个进来if (room == null) {Map<String, JSONObject> actionMap = new HashMap<>(4);actionMap.put(operation, jsonObject);room = new ActionRoomBean(userId, actionMap);//开启计时room.startManager();//放入到全局map中allMap.put(userId, room);}//如果有，直接调用 addAction方法room.addAction(operation, jsonObject);}}/*** 当前批次处理完之后，从集合中删除用户实例*/public static void closeRoom(String userId) {allMap.remove(userId);}
}

整体核心代码就是上边这些，以上还可以通过线程池去优化一下。

如果涉及到批量导入，同时有大量用户同步数据过来，就需要在测试环境进行反复测试 看是否会丢数据（因为每个用户都是一个独立的子线程），对线程的数量进行优化。