♻️ Refactoring code. 重构粉丝同步逻辑，减少操作逻辑，避免大批量耗时

2020-11-25 11:09:10 +08:00
1 parent 2018e571ea
commit cefc39f98d
@@ -18,8 +18,10 @@ package com.pig4cloud.pigx.mp.service.impl;

 import cn.hutool.core.collection.CollUtil;
 import cn.hutool.core.util.BooleanUtil;
+import cn.hutool.core.util.ObjectUtil;
 import cn.hutool.core.util.StrUtil;
 import com.baomidou.mybatisplus.core.toolkit.Wrappers;
+import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
 import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
 import com.pig4cloud.pigx.common.data.tenant.TenantBroker;
 import com.pig4cloud.pigx.mp.config.WxMpInitConfigRunner;
@@ -29,21 +31,20 @@ import com.pig4cloud.pigx.mp.mapper.WxAccountFansMapper;
 import com.pig4cloud.pigx.mp.mapper.WxAccountMapper;
 import com.pig4cloud.pigx.mp.service.WxAccountFansService;
 import lombok.AllArgsConstructor;
-import lombok.SneakyThrows;
 import lombok.extern.slf4j.Slf4j;
-import me.chanjar.weixin.common.error.WxErrorException;
 import me.chanjar.weixin.mp.api.WxMpService;
 import me.chanjar.weixin.mp.api.WxMpUserService;
+import me.chanjar.weixin.mp.bean.result.WxMpUser;
 import me.chanjar.weixin.mp.bean.result.WxMpUserList;
 import org.springframework.scheduling.annotation.Async;
 import org.springframework.stereotype.Service;
-import org.springframework.transaction.annotation.Transactional;

 import java.time.Instant;
 import java.time.LocalDateTime;
 import java.time.ZoneId;
 import java.util.ArrayList;
 import java.util.List;
+import java.util.stream.Collectors;

 /**
 * @author lengleng
@@ -68,8 +69,6 @@ public class WxAccountFansServiceImpl extends ServiceImpl<WxAccountFansMapper, W
 	 */
 	@Async
 	@Override
-	@SneakyThrows
-	@Transactional(rollbackFor = Exception.class)
 	public Boolean syncAccountFans(String appId) {
 		WxAccount wxAccount = wxAccountMapper
 				.selectOne(Wrappers.<WxAccount>query().lambda().eq(WxAccount::getAppid, appId));
@@ -77,67 +76,95 @@ public class WxAccountFansServiceImpl extends ServiceImpl<WxAccountFansMapper, W
 		// 获取操作微信接口类
 		WxMpService wxMpService = WxMpInitConfigRunner.getMpServices().get(appId);
 		WxMpUserService wxMpUserService = wxMpService.getUserService();
-
+		// 根据公众号查询已同步用户openid 查询最新的一条 (注意不能按订阅时间排序)
+		String finalNextOpenId = queryNextOpenId(appId);
 		TenantBroker.runAs(() -> WxMpInitConfigRunner.getTenants().get(appId),
-				(id) -> fetchUser(null, wxAccount, wxMpUserService));
+				(id) -> fetchUser(finalNextOpenId, wxAccount, wxMpUserService));
+
+		log.info("公众号 {} 粉丝同步完成", wxAccount.getName());
 		return Boolean.TRUE;
 	}

 	/**
 	 * 获取微信用户
-	 * @param nextOpenid
-	 * @param wxAccount
-	 * @param wxMpUserService
-	 * @throws WxErrorException
+	 * @param nextOpenid 下一组开始的openid
+	 * @param wxAccount 公众号信息
+	 * @param wxMpUserService mp操作类
 	 */
-	private void fetchUser(String nextOpenid, WxAccount wxAccount, WxMpUserService wxMpUserService)
-			throws WxErrorException {
-		WxMpUserList wxMpUserList = wxMpUserService.userList(nextOpenid);
-
-		// openId 分组 每组 100个 openid
-		CollUtil.split(wxMpUserList.getOpenids(), SIZE).stream().filter(CollUtil::isNotEmpty).forEach(openIdList -> {
-			log.info("开始批量获取用户信息 {}", openIdList);
-			List<WxAccountFans> wxAccountFansList = new ArrayList<>();
-			try {
+	private void fetchUser(String nextOpenid, WxAccount wxAccount, WxMpUserService wxMpUserService) {
+		try {
+			WxMpUserList wxMpUserList = wxMpUserService.userList(nextOpenid);
+			// openId 分组 每组 100个 openid
+			List<List<String>> openIdsList = CollUtil.split(wxMpUserList.getOpenids(), SIZE).stream()
+					.filter(CollUtil::isNotEmpty).collect(Collectors.toList());
+			// 处理每个分组. 调用查询用户信息
+			for (List<String> openIdList : openIdsList) {
+				log.info("开始批量获取用户信息 {}", openIdList);
+				List<WxAccountFans> wxAccountFansList = new ArrayList<>();
 				wxMpUserService.userInfoList(openIdList).forEach(wxMpUser -> {
-					WxAccountFans wxAccountFans = baseMapper.selectOne(Wrappers.<WxAccountFans>query().lambda()
-							.eq(WxAccountFans::getOpenid, wxMpUser.getOpenId()));
-					// 为空初始化一个新的对象
-					if (wxAccountFans == null) {
-						wxAccountFans = new WxAccountFans();
-					}
-
-					wxAccountFans.setOpenid(wxMpUser.getOpenId());
-					wxAccountFans.setSubscribeStatus(String.valueOf(BooleanUtil.toInt(wxMpUser.getSubscribe())));
-					wxAccountFans.setSubscribeTime(LocalDateTime.ofInstant(
-							Instant.ofEpochMilli(wxMpUser.getSubscribeTime() * 1000L), ZoneId.systemDefault()));
-					wxAccountFans.setNickname(wxMpUser.getNickname());
-					wxAccountFans.setGender(String.valueOf(wxMpUser.getSex()));
-					wxAccountFans.setLanguage(wxMpUser.getLanguage());
-					wxAccountFans.setCountry(wxMpUser.getCountry());
-					wxAccountFans.setProvince(wxMpUser.getProvince());
-					wxAccountFans.setCity(wxMpUser.getCity());
-					wxAccountFans.setHeadimgUrl(wxMpUser.getHeadImgUrl());
-					wxAccountFans.setRemark(wxMpUser.getRemark());
-					wxAccountFans.setWxAccountId(wxAccount.getId());
-					wxAccountFans.setWxAccountAppid(wxAccount.getAppid());
-					wxAccountFans.setWxAccountName(wxAccount.getName());
+					WxAccountFans wxAccountFans = buildDbUser(wxAccount, wxMpUser);
 					wxAccountFansList.add(wxAccountFans);
 				});
 				this.saveOrUpdateBatch(wxAccountFansList);
 				log.info("批量插入用户信息完成 {}", openIdList);
 			}
-			catch (WxErrorException e) {
-				log.error("批量同步粉丝失败", e);
+			// 如果nextOpenId 不为空，则继续获取
+			if (StrUtil.isNotBlank(wxMpUserList.getNextOpenid())) {
+				fetchUser(wxMpUserList.getNextOpenid(), wxAccount, wxMpUserService);
 			}
-		});
-
-		// 如果nextOpenId 不为空，则继续获取
-		if (StrUtil.isNotBlank(wxMpUserList.getNextOpenid())) {
-			fetchUser(wxMpUserList.getNextOpenid(), wxAccount, wxMpUserService);
+		}
+		catch (Exception e) {
+			log.warn("同步微信公众号 {} 粉丝异常 {}", wxAccount.getName(), e.getMessage());
+			// 失败重试逻辑
+			String finalNextOpenId = queryNextOpenId(wxAccount.getAppid());
+			fetchUser(finalNextOpenId, wxAccount, wxMpUserService);
 		}

-		log.info("批量同步微信用户信息完成");
+	}
+
+	/**
+	 * 构建数据库存取对象
+	 * @param wxAccount 公众号信息
+	 * @param wxMpUser 用户微信信息
+	 */
+	private WxAccountFans buildDbUser(WxAccount wxAccount, WxMpUser wxMpUser) {
+		WxAccountFans wxAccountFans = new WxAccountFans();
+		wxAccountFans.setOpenid(wxMpUser.getOpenId());
+		wxAccountFans.setSubscribeStatus(String.valueOf(BooleanUtil.toInt(wxMpUser.getSubscribe())));
+
+		// 2020-11-25 部分用户订阅时间为空，跳过此字段处理
+		if (ObjectUtil.isNotEmpty(wxMpUser.getSubscribeTime())) {
+			wxAccountFans.setSubscribeTime(LocalDateTime
+					.ofInstant(Instant.ofEpochMilli(wxMpUser.getSubscribeTime() * 1000L), ZoneId.systemDefault()));
+		}
+		wxAccountFans.setNickname(wxMpUser.getNickname());
+		wxAccountFans.setGender(String.valueOf(wxMpUser.getSex()));
+		wxAccountFans.setLanguage(wxMpUser.getLanguage());
+		wxAccountFans.setCountry(wxMpUser.getCountry());
+		wxAccountFans.setProvince(wxMpUser.getProvince());
+		wxAccountFans.setCity(wxMpUser.getCity());
+		wxAccountFans.setHeadimgUrl(wxMpUser.getHeadImgUrl());
+		wxAccountFans.setRemark(wxMpUser.getRemark());
+		wxAccountFans.setWxAccountId(wxAccount.getId());
+		wxAccountFans.setWxAccountAppid(wxAccount.getAppid());
+		wxAccountFans.setWxAccountName(wxAccount.getName());
+		return wxAccountFans;
+	}
+
+	/**
+	 * 查询当前公众号最后插入的openId
+	 * @param appId 公众号标识
+	 * @return openid / null
+	 */
+	private String queryNextOpenId(String appId) {
+		Page<WxAccountFans> queryPage = new Page<>(0, 1);
+		Page<WxAccountFans> fansPage = baseMapper.selectPage(queryPage, Wrappers.<WxAccountFans>query().lambda()
+				.eq(WxAccountFans::getWxAccountAppid, appId).orderByDesc(WxAccountFans::getCreateTime));
+		String nextOpenId = null;
+		if (fansPage.getTotal() > 0) {
+			nextOpenId = fansPage.getRecords().get(0).getOpenid();
+		}
+		return nextOpenId;
 	}

 }