package com.ynxbd.push.helper; import com.fasterxml.jackson.core.JsonProcessingException; import com.fasterxml.jackson.databind.ObjectMapper; import com.ynxbd.push.config.IniConfig; import com.ynxbd.push.entity.*; import com.ynxbd.push.entity.tencent.*; import com.ynxbd.push.httpRequest.xbd.wx.WxCacheRequest; import com.ynxbd.push.thread.ThreadConfig; import lombok.SneakyThrows; import lombok.extern.slf4j.Slf4j; import org.apache.http.Header; import org.apache.http.HttpResponse; import org.apache.http.client.methods.HttpUriRequest; import org.apache.http.client.methods.RequestBuilder; import org.apache.http.concurrent.FutureCallback; import org.apache.http.entity.StringEntity; import org.apache.http.impl.nio.client.CloseableHttpAsyncClient; import org.springframework.beans.BeanUtils; import weixin.popular.api.API; import weixin.popular.api.MessageAPI; import weixin.popular.api.UserAPI; import weixin.popular.bean.message.MessageSendResult; import weixin.popular.bean.message.massmessage.Filter; import weixin.popular.bean.message.massmessage.MassTextMessage; import weixin.popular.bean.user.Group; import weixin.popular.client.LocalHttpClient; import java.io.IOException; import java.nio.charset.StandardCharsets; import java.util.*; import java.util.concurrent.CountDownLatch; import java.util.concurrent.ThreadPoolExecutor; /** * 所有消息发送的工厂类 * * @Project WebApi * @Name MessageApiHelper * @Author 李进才 * @Date 2020/5/25 * @Version v1.0.0 * @Copyright @ 2017 云南新八达科技有限公司 All rights reserved. */ @Slf4j public class MessageApiHelper { protected static Header jsonHeader; public static final Integer EVERY_OPENID_COUNT = 10000; public static final String CODE_INVALIDATION = "40001"; public static final String CODE_EXPIRED = "42001"; public static final String MESSAGE_RESULT_ERROR_CODE = "4002"; /** * 三层调用,第一层,传输SendMessageTemplate对象(减少controller代码量) * * @param template * @return */ public static MessageSendResult templateMessageSend(SendMessageTemplate template) { MessageSendResult messageSendResult = templateMessage(template); log.info("TenCent Return Message" + messageSendResult.getErrcode()); return messageSendResult; } /** * 三层调用第二层,转换实体对象为json对象 * * @param sendMessageTemplate 模版数据 * @return 请求接口返回的状态码等信息 */ public static MessageSendResult templateMessage(SendMessageTemplate sendMessageTemplate) { try { String jsonMessage = TemplateSelectHelper.select(sendMessageTemplate); log.info("SendTenCentJson:{}", jsonMessage); return templateMessageSend(jsonMessage); } catch (Exception e) { MessageSendResult messageSendResult = new MessageSendResult(); messageSendResult.setErrcode(MESSAGE_RESULT_ERROR_CODE); return messageSendResult; } } /** * 三层调用最后一层,调起http发送模版消息 * * @param messageJson json字符串 */ private static MessageSendResult templateMessageSend(String messageJson) { HttpUriRequest httpUriRequest = RequestBuilder.post().setHeader(jsonHeader).setUri("https://api.weixin.qq.com/cgi-bin/message/template/send").addParameter("access_token", API.accessToken(WxCacheRequest.getAccessToken())).setEntity(new StringEntity(messageJson, StandardCharsets.UTF_8)).build(); MessageSendResult result = LocalHttpClient.executeJsonResult(httpUriRequest, MessageSendResult.class); String errCode = result.getErrcode(); if (CODE_INVALIDATION.equals(errCode) || CODE_EXPIRED.equals(errCode)) { String errMsg = result.getErrmsg(); log.error("[推送]请求失败 errCode={}, errMsg={}", errCode, errMsg); } return result; } /** * 群发消息给用户 * * @param content 发送消息的内容 * @return 请求接口返回的状态码等信息 */ public static MessageSendResult group(String content) { try { String accessToken = WxCacheRequest.getAccessToken(); Group group = UserAPI.groupsGet(accessToken); String gid = group.getGroups().get(0).getId(); MassTextMessage textMessage = new MassTextMessage(content); textMessage.setFilter(new Filter(false, gid)); return MessageAPI.messageMassSendall(accessToken, textMessage); } catch (Exception e) { log.info(e.getMessage()); MessageSendResult messageSendResult = new MessageSendResult(); messageSendResult.setErrcode(MESSAGE_RESULT_ERROR_CODE); return messageSendResult; } } /** * 根据openid用客服接口发送信息,最多每次发送二十条,每当用户回复自己可以再发送二十条 * * @param openid 用户在该微信公众号的唯一id,也就是需要发送给哪个用户 * @param content 需要发送给用户的内容 * @return 请求接口返回的状态码等信息 */ public static MessageSendResult sendTextForOpenId(String openid, String content) { try { TestMessage testMessage = new TestMessage(); testMessage.setMsgtype("text"); testMessage.setTouser(openid); Map map = new HashMap<>(); map.put("content", content); testMessage.setText(map); String jsonTestMessage = new ObjectMapper().writeValueAsString(testMessage); return messageMassSend(jsonTestMessage); } catch (Exception e) { MessageSendResult messageSendResult = new MessageSendResult(); messageSendResult.setErrcode(MESSAGE_RESULT_ERROR_CODE); return messageSendResult; } } /** * 通过客服的方式发送链接信息给用户 * * @param openid 用户在该微信公众号的唯一id,也就是需要发送给哪个用户 * @param title 标题内容 * @param description 描述 * @param url 跳转链接 * @param thumb_url 图片链接 * @return 请求接口返回的状态码等信息 */ public static MessageSendResult sendLinkForOpenId(String openid, String title, String description, String url, String thumb_url) { try { LinkMessage linkMessage = new LinkMessage(); linkMessage.setMsgtype("link"); linkMessage.setTouser(openid); Map map = new HashMap<>(); map.put("title", title); map.put("description", description); map.put("url", url); map.put("thumb_url", thumb_url); linkMessage.setLink(map); String jsonTestMessage = new ObjectMapper().writeValueAsString(linkMessage); return messageMassSend(jsonTestMessage); } catch (Exception e) { MessageSendResult messageSendResult = new MessageSendResult(); messageSendResult.setErrcode(MESSAGE_RESULT_ERROR_CODE); return messageSendResult; } } /** * 一次性订阅消息发送,因为每家都一样全部都在一个类封装好了,不像模版消息调了好多层解偶 * * @param subscribe * @return * @throws Exception */ public static MessageSendResult subscribeMessage(Subscribe subscribe) throws JsonProcessingException { Map miniProgram = new HashMap<>(); miniProgram.put("appid", subscribe.getAppid()); miniProgram.put("pagepath", subscribe.getPagepath()); Map data = new HashMap<>(); data.put("content", TemplateJsonValue.builder().value(subscribe.getContent()).color("#173177").build()); TemplateJson templateJson = TemplateJson.builder() .template_id(subscribe.getTemplate_id()) .miniprogram(miniProgram) .url(subscribe.getUrl()) .scene(subscribe.getScene()) .title(subscribe.getTitle()) .data(data) .touser(subscribe.getTouser()) .build(); ObjectMapper objectMapper = new ObjectMapper(); String messageJson = objectMapper.writeValueAsString(templateJson); HttpUriRequest httpUriRequest = RequestBuilder.post().setHeader(jsonHeader).setUri("https://api.weixin.qq.com/cgi-bin/message/template/subscribe").addParameter("access_token", API.accessToken(WxCacheRequest.getAccessToken())).setEntity(new StringEntity(messageJson, StandardCharsets.UTF_8)).build(); return LocalHttpClient.executeJsonResult(httpUriRequest, MessageSendResult.class); } /** * 发送文本消息 * * @param messageJson messageJson * @return MessageSendResult */ private static MessageSendResult messageMassSend(String messageJson) { HttpUriRequest httpUriRequest = RequestBuilder.post().setHeader(jsonHeader).setUri("https://api.weixin.qq.com/cgi-bin/message/custom/send").addParameter("access_token", API.accessToken(WxCacheRequest.getAccessToken())).setEntity(new StringEntity(messageJson, StandardCharsets.UTF_8)).build(); return LocalHttpClient.executeJsonResult(httpUriRequest, MessageSendResult.class); } /** * 获取该公众号下所有用户的openId * * @param next_openid next_openid * @return */ public static AllUser getAllUser(String next_openid) { try { HttpUriRequest httpUriRequest = RequestBuilder.get().setHeader(jsonHeader).setUri("https://api.weixin.qq.com/cgi-bin/user/get").addParameter("access_token", API.accessToken(WxCacheRequest.getAccessToken())).addParameter("next_openid", next_openid).build(); AllUser messageSendResult = LocalHttpClient.executeJsonResult(httpUriRequest, AllUser.class); messageSendResult.setErrcode("0"); return messageSendResult; } catch (Exception e) { AllUser allUser = new AllUser(); allUser.setErrcode(MESSAGE_RESULT_ERROR_CODE); return allUser; } } /** * 群发模版消息入口(包括多线程组装) * * @param sendMessageTemplate * @return * @throws Exception */ public static CompleteThreadCount groupTemplateSend(SendMessageTemplate sendMessageTemplate) throws IOException, InterruptedException { final CompleteThreadCount completeThreadCount = CompleteThreadCount.builder().completed(0).failed(0).cancelled(0).build(); final OpenIdList openIdList = getAllUserList(); ThreadPoolExecutor threadPoolExecutor = ThreadConfig.getExecutor(15, 20); log.info("所有用户总数:" + openIdList.getCount()); final CloseableHttpAsyncClient httpClient = HttpAsyncClientHelper.getHttpAsyncClient(); final CountDownLatch latch = new CountDownLatch(openIdList.getCount()); httpClient.start(); for (List openIds : openIdList.getOpenIds()) { try { threadPoolExecutor.execute(() -> groupSendTemplateMessage(httpClient, openIds, sendMessageTemplate, completeThreadCount, latch)); } catch (Exception e) { log.error("主线程循环过程中出现错误:{}", e.getMessage()); } } log.info("---执行完成-线程未停止!----完成数:{}, 失败数{}", completeThreadCount.getCompleted(), completeThreadCount.getFailed() + "取消数:" + completeThreadCount.getCancelled()); latch.await(); httpClient.close(); List notStopRunnable = threadPoolExecutor.shutdownNow(); for (Runnable runnable : notStopRunnable) { log.error("未停止任务:{}", runnable.toString()); } log.info("线程是否结束:{}, 异步是否结束:{}", threadPoolExecutor.isShutdown(), httpClient.isRunning()); log.info("---执行完成-线程已停止!!---完成数:{}, 失败数:{}", completeThreadCount.getCompleted(), completeThreadCount.getFailed() + "取消数:" + completeThreadCount.getCancelled()); return completeThreadCount; } /** * 发送一组openid的模版消息(任务分包,将openid分散然后由异步发送) * * @param completeThreadCount * @param openIds * @throws Exception */ private static void groupSendTemplateMessage(CloseableHttpAsyncClient httpClient, List openIds, SendMessageTemplate sendMessageTemplate, CompleteThreadCount completeThreadCount, CountDownLatch latch) { //将一万条请求任务逐次加入线程 for (String openId : openIds) { try { SendMessageTemplate singleSendMessageTemplate = SendMessageTemplate.builder().build(); BeanUtils.copyProperties(sendMessageTemplate, singleSendMessageTemplate); singleSendMessageTemplate.setOpenId(openId); Thread.sleep(Integer.parseInt(IniConfig.getInstance("speed"))); asyncTemplateMessageSend(httpClient, completeThreadCount, RequestBuilder .post() .setHeader(jsonHeader) .setUri("https://api.weixin.qq.com/cgi-bin/message/template/send") .addParameter("access_token", API.accessToken(Objects.requireNonNull(WxCacheRequest.getAccessToken()))) .addParameter("time_stamp", String.valueOf(System.currentTimeMillis())) .setEntity(new StringEntity(TemplateSelectHelper.select(singleSendMessageTemplate), StandardCharsets.UTF_8)) .build(), latch); } catch (Exception e) { log.error("httpClient异步线程执行中出现错误:{}", e.toString()); } } } /** * 异步发送网络请求(实际调起http的地方,异步调用) * * @param httpclient * @param request */ private static void asyncTemplateMessageSend(final CloseableHttpAsyncClient httpclient, final CompleteThreadCount completeThreadCount, final HttpUriRequest request, final CountDownLatch latch) throws IOException { httpclient.execute(request, new FutureCallback() { @SneakyThrows @Override public void completed(HttpResponse response) { completeThreadCount.setCompleted(completeThreadCount.getCompleted() + 1); latch.countDown(); } @SneakyThrows @Override public void failed(Exception ex) { log.error("回调过程中出现错误:" + ex.toString()); completeThreadCount.setFailed(completeThreadCount.getFailed() + 1); latch.countDown(); } @SneakyThrows @Override public void cancelled() { completeThreadCount.setCancelled(completeThreadCount.getCancelled() + 1); latch.countDown(); } }); } /** * 获取所有用户列表的集合(1W个用户为一组) * * @return */ private static OpenIdList getAllUserList() { List> openIds = new ArrayList<>(); List nextOpenIds = new ArrayList<>(); AllUser userList = getAllUser(null); String nextOpenId = userList.getNext_openid(); nextOpenIds.add(nextOpenId); Integer count = ((List) userList.getData().get("openid")).size(); openIds.add((List) userList.getData().get("openid")); for (int i = 0; i < Math.floor(userList.getTotal() / EVERY_OPENID_COUNT); i++) { try { if (nextOpenId == null || nextOpenId.isEmpty()) { break; } userList = getAllUser(nextOpenId); if (userList.getData().get("openid") == null) { break; } nextOpenId = userList.getNext_openid(); nextOpenIds.add(nextOpenId); count += ((List) userList.getData().get("openid")).size(); openIds.add((List) userList.getData().get("openid")); } catch (Exception e) { log.error("请求列表过程中出现错误:next_openid:" + nextOpenId + "错误:" + e.toString()); } } return OpenIdList.builder().openIds(openIds).count(count).build(); } // // /** // * 将所有openId组装成SendMessageTemplate集合 // * @param openIdList // * @param sendMessageTemplate // * @return // */ // private List> GetAllSendMessageTemplate(OpenIdList openIdList, SendMessageTemplate sendMessageTemplate){ // List> allSendMessageTemplate = new ArrayList<>(); // for (List openList : openIdList.getOpenIds()) { // allSendMessageTemplate.add(GetGroupSendMessageTemplate(openList,sendMessageTemplate)); // } // return allSendMessageTemplate; // } // // /** // * 单组openId组装成SendMessageTemplate哈希集合,防止重复 // * @param groupOpenId // * @param sendMessageTemplate // * @return // */ // private List GetGroupSendMessageTemplate(List groupOpenId, SendMessageTemplate sendMessageTemplate){ // List groupSendMessageTemplate = new ArrayList<>(); // for(String openId: groupOpenId){ // SendMessageTemplate newSendMessageTemplate =SendMessageTemplate.builder().build(); // BeanUtils.copyProperties(sendMessageTemplate,newSendMessageTemplate); // newSendMessageTemplate.setOpenId(openId); // groupSendMessageTemplate.add(newSendMessageTemplate); // } // return groupSendMessageTemplate; // } }