微信给所有用户群发模板消息
业务需求需要给所有关注公众号用户发送模板消息,单线程发送人数过多发送时间也随之增加,多线程能都大幅度减少处理时间
获取微信所有关注用户工具类
@Componentpublic class WeChatSubscribePublicUtil {private static final String APPID = "";@Resourceprivate AccessTokenUtil accessTokenUtil;public Set<String> wxSubscribe(){Set<String> set = null;String nextOpendId = null;try {set = setWx(set, nextOpendId);} catch (IOException e) {e.printStackTrace();}return set;}private Set<String> setWx(Set<String> set, String nextOpendId) throws IOException {String url = "https://api./cgi-bin/user/get?access_token=%s";String url2 = "https://api./cgi-bin/user/get?access_token=%s&next_openid=%s";JSONObject res;if (set == null) {set = new HashSet<>();res = HttpRequestUtils.getC(String.format(url, accessTokenUtil.getAccessToken(APPID)));} else {res = HttpRequestUtils.getC(String.format(url2, accessTokenUtil.getAccessToken(APPID), nextOpendId));}if (res != null) {if (set.size() < res.getInteger("total")) {set.addAll(JSONObject.parseArray(res.getJSONObject("data").getJSONArray("openid").toJSONString(),String.class));if(res.getString("next_openid")!=null){setWx(set,res.getString("next_openid"));}}}return set;}}
下面给所有用户发送消息
public void TemplateSendThread() {System.out.println("模板消息线程开始");//获取用户Set<String> set = weChatSubscribePublicUtil.wxSubscribe();List<String> openIds= new ArrayList<>(set);//单个线程处理数量int threadSize = 5000;int remainder = openIds.size() % threadSize;//线程数int threadNum;if (remainder == 0) {threadNum = openIds.size() / threadSize;} else {threadNum = openIds.size() / threadSize + 1;}//创建一个线程池ExecutorService executorService = Executors.newFixedThreadPool(threadNum);final String taskCode = "203012";List<Callable<String>> callableList = new ArrayList<>();//线程任务Callable<String> task;//分批处理信息 泛型与获取到总数据类型一致List<String> cutList;for (int i = 0; i < threadNum; i++) {//数据进行分批if (i == threadNum - 1) {cutList = openIds.subList(i * threadSize, openIds.size());} else {cutList = openIds.subList(i * threadSize, (i + 1) * threadSize);}//待处理信息final List<String> handlerList = cutList;int finalI = i;AtomicReference<String> accessToken = new AtomicReference<>("");try {accessToken.set(sendTemplateMessageUtil.getAccessToken(APPID));} catch (IOException e) {e.printStackTrace();}task = () -> {String threadName = "线程-" + finalI;for (int j = 0; j < handlerList.size(); j++) {synchronized (handlerList.get(j)) {try {String temp = handlerList.get(j);MessageRecord messageRecord1 = messageRecordMapper.selectByOpenIdAndTaskCode(temp, taskCode);if (messageRecord1 == null) {JSONObject res = sendTemplateMessageUtil.SendTemplateMessages1(temp, templateId, link, appid, pagepath, keyword1, keyword2, keyword3, keyword4, keyword5,remark, msg, accessToken.get());//token失效重新获取if (res.getIntValue("errcode") == 40001) {accessToken.set(sendTemplateMessageUtil.getAccessToken(APPID));res = sendTemplateMessageUtil.SendTemplateMessages1(temp, templateId, link, appid, pagepath, keyword1, keyword2, keyword3, keyword4, keyword5,remark, msg, accessToken.get());}//记录发送返回信息if (res != null) {MessageRecord messageRecord2 = new MessageRecord();messageRecord2.setId(codeUtil.UUID());messageRecord2.setOpenId(temp);messageRecord2.setErrcode(res.getInteger("errcode"));messageRecord2.setErrmsg(res.getString("errmsg"));messageRecord2.setTaskCode(taskCode);messageRecordMapper.insertSelective(messageRecord2);}}} catch (Exception e) {e.printStackTrace();}}}return threadName;};callableList.add(task);}List<Future<String>> results = executorService.invokeAll(callableList);for (Future<String> o : results) {//线程执行o.get();}executorService.shutdown();System.out.println("模板消息线程结束");}
