微信公众号（服务号）提供批量获取关注的微信用户信息接口，但每次最多获取 100个，如果一个公司有几个公众号，每个微信公众号又有几万甚至几十万用户，那么使用单线程获取数据显然不合理，需要使用多线程。
1、环境搭建
这里引入一个开源的微信API工具
github 地址：
https://github.com/binarywang/weixin-java-mp-demo-springboot
里面有很相关的Demo 和详细的讲解，我这里就不赘述
POM 引入
<!-- 微信API -->
<dependency>
<groupId>com.github.binarywang</groupId>
<artifactId>weixin-java-mp</artifactId>
<version>3.3.0</version>
<exclusions>
<exclusion>
<groupId>com.fasterxml.jackson.core</groupId>
<artifactId>jackson-databind</artifactId>
</exclusion>
<exclusion>
<groupId>org.apache.httpcomponents</groupId>
<artifactId>httpclient</artifactId>
</exclusion>
</exclusions>
</dependency>
注意: 我这里是因为在其他jar包引入了 json 和 httpclient 所以排除掉了，一般情况下，是不需要做 exclusion 的
2、同步思路
2.1 假设我们有 3 个公众号的用户需要同步，那么我们开启三个线程，每个线程去同步一个公号
2.2 根据微信的接口 https://developers.weixin.qq.com/doc/offiaccount/User_Management/Getting_a_User_List.html
，每次可以获取10000个用户ID，但是每次请求最多获取100个用户，所以我们每获取10000个ID（一页），就再开启一个多线程去拉取用户，达到效率最大化
PS： 简单点讲就是多线程里面再开启一个多线程
上代码：
// 创建一个线程池
ExecutorService executorService = new ThreadPoolExecutor(10, 10,
0L, TimeUnit.MILLISECONDS,
new LinkedBlockingQueue<Runnable>(),
Executors.defaultThreadFactory());
List<WxUser> wxUserList = new ArrayList<>();
// 循环每个公众号
for (String platformCode : platformChannelPromotionsMap.keySet()) {
WxUserComponent totalWxPlatformComponent = new WxUserComponent ();
executorService.execute(totalWxPlatformComponent);
}
// 不再接收新任务
executorService.shutdown();
while (true) {
// 任务都已经执行完
if (executorService.isTerminated()) {
logger.info("统计 {} 公众号渠道累计人数 线程结束", targetDate);
break;
}
// 设置一个超时时间，防止意外事情发生
if (System.currentTimeMillis() - created.getTime() > TIMEOUT) {
logger.error("统计 {} 公众号 线程超时", targetDate);
throw new RuntimeException("统计 " + targetDate + " 公众号 线程超时");
}
Thread.sleep(INTERVAL);
}
/**
* 每个微信公众号的用户
* 实现了Runnable接口，可以使用多线程
*
* @author WangMin
* @version V1.0.0
* @since 2019/9/21
*/
public class WxUserComponent implements Runnable {
private static final Logger logger = LoggerFactory.getLogger(WxUserComponent.class);
/**
* 超时时间
*/
private static final Integer TIMEOUT = 2 * 60 * 60 * 1000;
/**
* 线程等待 sleep 间隔
*/
private static final Integer INTERVAL = 5000;
/**
* 微信平台
*/
private WxPlatform wxPlatform;
@Override
public void run() {
WxMpService wxMpService = WxmpConfig.getWxMpService(wxPlatform);
WxMpUserList wxMpUserList;
try {
wxMpUserList = wxMpService.getUserService().userList(null);
} catch (WxErrorException e) {
logger.error("统计平台公众号：{} 出错,错误： {}", wxPlatform.getCode(), e.getMessage());
throw new RuntimeException("统计平台公众号：" + wxPlatform.getCode() + " 出错,错误:" + e.getMessage());
}
List<String> openIds = wxMpUserList.getOpenids();
// total 关注该公众账号的总用户数
// count 拉取的OPENID个数，最大值为10000
logger.info("平台 {} ,用户列表 total： {}, count: {}", wxPlatform.getCode(), wxMpUserList.getTotal(), wxMpUserList.getCount());
String nextOpenId = wxMpUserList.getNextOpenid();
if (StringUtils.isBlank(nextOpenId)) {
logger.error("统计平台公众号：{} nextOpenId 为空", wxPlatform.getCode());
return;
}
ExecutorService executorService = new ThreadPoolExecutor(10, 10,
0L, TimeUnit.MILLISECONDS,
new LinkedBlockingQueue<Runnable>(),
Executors.defaultThreadFactory());
List<RequestWxUserInfo> threadList = new ArrayList<>();
List<WxMpUser> platformWxMpUserList = Collections.synchronizedList(new ArrayList<WxMpUser>());
while (StringUtils.isNotBlank(nextOpenId)) {
RequestWxUserInfo requestWxUserInfo = new RequestWxUserInfo(openIds);
threadList.add(requestWxUserInfo);
executorService.execute(requestWxUserInfo);
nextOpenId = wxMpUserList.getNextOpenid();
try {
wxMpUserList = wxMpService.getUserService().userList(nextOpenId);
openIds = wxMpUserList.getOpenids();
} catch (WxErrorException e) {
logger.error("统计平台公众号：{} 时，获取 userList 出错,错误： {}", wxPlatform.getCode(), e.getMessage());
throw new RuntimeException("统计平台公众号：" + wxPlatform.getCode() + " 时,获取 userList 出错,错误:" + e.getMessage());
}
}
executorService.shutdown();
while (true) {
if (executorService.isTerminated()) {
logger.info("统计平台公众号 {} 获取用户详情 线程结束", targetDate);
break;
}
if (System.currentTimeMillis() - created.getTime() > TIMEOUT) {
logger.error("统计平台公众号 {} 获取用户详情出错 线程超时", targetDate);
throw new RuntimeException("统计平台公众号 " + targetDate + " 获取用户详情出错 线程超时");
}
try {
Thread.sleep(INTERVAL);
} catch (InterruptedException e) {
// do nothing
}
}
for (RequestWxUserInfo requestWxUserInfo : threadList) {
platformWxMpUserList.addAll(requestWxUserInfo.getWxMpUserList());
}
this.totalWxMpUserChannel(platformWxMpUserList, channelPromotionCodesMap, channelPromotionSubscribeCountMap);
}
/**
* 获取用户信息
*/
class RequestWxUserInfo implements Runnable {
private List<String> openIds;
private List<WxMpUser> wxMpUserList = Collections.synchronizedList(new ArrayList<WxMpUser>());
RequestWxUserInfo(List<String> openIds) {
this.openIds = openIds;
}
@Override
public void run() {
// 每次最多请求100条数据
final int batchGetSize = 100;
int size = openIds.size();
int page = size / batchGetSize;
int remainder = size % batchGetSize;
try {
for (int i = 0; i < page; i++) {
List<String> subOpenIds = openIds.subList(i * batchGetSize, (i + 1) * batchGetSize);
List<WxMpUser> userInfoList = WxmpConfig.getWxMpService(wxPlatform).getUserService().userInfoList(subOpenIds);
wxMpUserList.addAll(userInfoList);
}
if (remainder > 0) {
int lastBeginIndex = page * batchGetSize;
List<String> subOpenIds = openIds.subList(lastBeginIndex, size);
logger.info("统计平台公众号：{}，获取用户详情,remainder: {} 最后一页 {} 条数据", wxPlatform.getCode(), remainder, subOpenIds.size());
List<WxMpUser> userInfoList = WxmpConfig.getWxMpService(wxPlatform).getUserService().userInfoList(subOpenIds);
wxMpUserList.addAll(userInfoList);
}
} catch (Exception e) {
logger.error("统计平台公众号：{} 时，获取用户详情出错,错误： {}", wxPlatform.getCode(), e.getMessage());
throw new RuntimeException("统计平台公众号：" + wxPlatform.getCode() + " 时,获取用户详情出错,错误:" + e.getMessage());
}
}
}
这是一些思路和部分代码，希望对各位有帮助，如果有什么问题，也请各位指出。
