YOHOBUY / yoho-message-sender · Commits

Go to a project

Authored by unknown
Commit c066c87811d6a915c01ae96b6c52daeb7b378415 2 parents a1ea375c b453601b

Merge branch 'dev-2683微信服务号客服消息'

Inline Side-by-side
Showing 10 changed files with 310 additions and 7 deletions
@@ -8,4 +8,6 @@ public class RedisCacheKeyConstant { @@ -8,4 +8,6 @@ public class RedisCacheKeyConstant {
8 8
9 //小程序发送数统计key 9 //小程序发送数统计key
10 public static final String MINI_SEND_NUM_KEY = "yh:messageSender:MINI_SEND_NUM_"; 10 public static final String MINI_SEND_NUM_KEY = "yh:messageSender:MINI_SEND_NUM_";
  11 +
  12 + public static final String WECAHT_CUSTOM_SEND_KEY = "yh:messageSender:WECAHT_CUSTOM_SEND_";
11 } 13 }
@@ -3,11 +3,15 @@ package com.yoho.yhmessage.redis; @@ -3,11 +3,15 @@ package com.yoho.yhmessage.redis;
3 import com.yoho.core.redis.cluster.annotation.Redis; 3 import com.yoho.core.redis.cluster.annotation.Redis;
4 import com.yoho.core.redis.cluster.operations.nosync.*; 4 import com.yoho.core.redis.cluster.operations.nosync.*;
5 import com.yoho.core.redis.cluster.operations.serializer.RedisKeyBuilder; 5 import com.yoho.core.redis.cluster.operations.serializer.RedisKeyBuilder;
  6 +import org.apache.commons.collections.CollectionUtils;
6 import org.springframework.data.redis.serializer.RedisSerializer; 7 import org.springframework.data.redis.serializer.RedisSerializer;
7 import org.springframework.data.redis.serializer.StringRedisSerializer; 8 import org.springframework.data.redis.serializer.StringRedisSerializer;
8 import org.springframework.stereotype.Service; 9 import org.springframework.stereotype.Service;
9 10
  11 +import java.util.ArrayList;
  12 +import java.util.HashSet;
10 import java.util.List; 13 import java.util.List;
  14 +import java.util.Set;
11 import java.util.concurrent.TimeUnit; 15 import java.util.concurrent.TimeUnit;
12 16
13 /** 17 /**
@@ -103,4 +107,42 @@ public class PushRedisService { @@ -103,4 +107,42 @@ public class PushRedisService {
103 return count; 107 return count;
104 } 108 }
105 109
  110 + public void getSetFilter(Set<String> orginSet, String redisKey){
  111 + if(CollectionUtils.isEmpty(orginSet)){
  112 + return;
  113 + }
  114 + List<String> orginList = new ArrayList<>(orginSet);
  115 + Set<String> invalidSet = new HashSet<>();
  116 + List<Object> filteredList = pushRedisTemplate.executePipelined(connection -> {
  117 + connection.openPipeline();
  118 + byte[] key = stringSerializer.serialize(redisKey);
  119 + for (String str : orginList) {
  120 + connection.sIsMember(key, stringSerializer.serialize(str));
  121 + }
  122 + return null;
  123 + });
  124 + for (int j = 0; j < orginSet.size(); j++) {
  125 + //如果存在 则过滤
  126 + if ("true".equals(filteredList.get(j).toString())) {
  127 + invalidSet.add(orginList.get(j));
  128 + }
  129 + }
  130 + orginSet.removeAll(invalidSet);
  131 + }
  132 +
  133 + public void setListValue(String redisKey,List<String> valueList,int hours){
  134 + //写入redis中的当天发送uid set集合
  135 + pushRedisTemplate.executePipelined(connection -> {
  136 + connection.openPipeline();
  137 + byte[] key = stringSerializer.serialize(redisKey);
  138 + for (String value : valueList) {
  139 + connection.sAdd(key, stringSerializer.serialize(value));
  140 + }
  141 + return null;
  142 + });
  143 + //设置超时时间
  144 + RedisKeyBuilder redisTodayKey = RedisKeyBuilder.newInstance().appendFixed(redisKey);
  145 + pushRedisTemplate.expire(redisTodayKey,hours,TimeUnit.HOURS);
  146 + }
  147 +
106 } 148 }
@@ -4,7 +4,7 @@ @@ -4,7 +4,7 @@
4 <parent> 4 <parent>
5 <groupId>com.yoho</groupId> 5 <groupId>com.yoho</groupId>
6 <artifactId>yoho-starter</artifactId> 6 <artifactId>yoho-starter</artifactId>
7 - <version>1.4.5-SNAPSHOT</version> 7 + <version>1.4.7-SNAPSHOT</version>
8 </parent> 8 </parent>
9 9
10 <groupId>com.yoho.dsf</groupId> 10 <groupId>com.yoho.dsf</groupId>
@@ -16,4 +16,19 @@ consumer: @@ -16,4 +16,19 @@ consumer:
16 topic: msgcenter.inner.wechat 16 topic: msgcenter.inner.wechat
17 17
18 - class: com.yoho.yhmessage.wechat.consumer.MiniMsgSendConsumer 18 - class: com.yoho.yhmessage.wechat.consumer.MiniMsgSendConsumer
19 - topic: msgcenter.inner.mini  
  19 + topic: msgcenter.inner.mini
  20 +
  21 + - class: com.yoho.yhmessage.wechat.consumer.WechatCustomMsgConsumer
  22 + topic: msgcenter.inner.wechatCustom
  23 +
  24 +
  25 +producer:
  26 + - address: 172.16.6.54:5672
  27 + username: yoho
  28 + password: yoho
  29 + producers:
  30 + - bean: yhProducer
  31 + async: true
  32 + trace: false
  33 + confirm: true
  34 + persistent: false
@@ -16,4 +16,19 @@ consumer: @@ -16,4 +16,19 @@ consumer:
16 topic: msgcenter.inner.wechat 16 topic: msgcenter.inner.wechat
17 17
18 - class: com.yoho.yhmessage.wechat.consumer.MiniMsgSendConsumer 18 - class: com.yoho.yhmessage.wechat.consumer.MiniMsgSendConsumer
19 - topic: msgcenter.inner.mini  
  19 + topic: msgcenter.inner.mini
  20 +
  21 + - class: com.yoho.yhmessage.wechat.consumer.WechatCustomMsgConsumer
  22 + topic: msgcenter.inner.wechatCustom
  23 +
  24 +
  25 +producer:
  26 + - address: ${rabbit_common}
  27 + username: ${rabbit_common_user}
  28 + password: ${rabbit_common_password}
  29 + producers:
  30 + - bean: yhProducer
  31 + async: true
  32 + trace: false
  33 + confirm: true
  34 + persistent: false
  1 +package com.test;
  2 +
  3 +import com.alibaba.fastjson.JSONArray;
  4 +import com.alibaba.fastjson.JSONObject;
  5 +import com.yoho.message.dal.MessageFilterInfoMapper;
  6 +import com.yoho.service.model.msgcenter.wechat.WechatCustomMsgBO;
  7 +import com.yoho.yhmessage.wechat.service.IWechatService;
  8 +import org.junit.Test;
  9 +import org.junit.runner.RunWith;
  10 +import org.springframework.beans.factory.annotation.Autowired;
  11 +import org.springframework.test.context.ContextConfiguration;
  12 +import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
  13 +
  14 +import java.util.ArrayList;
  15 +import java.util.List;
  16 +
  17 +/**
  18 + * Created by min.ling on 2018/10/8.
  19 + */
  20 +@RunWith(SpringJUnit4ClassRunner.class)
  21 +@ContextConfiguration(locations = {"classpath*:META-INF/spring/spring*.xml","classpath*:META-INF/spring/mybatis-datasource.xml",})
  22 +public class WechatSendTest {
  23 +
  24 + @Autowired
  25 + MessageFilterInfoMapper filterInfoMapper;
  26 +
  27 + @Autowired
  28 + private IWechatService wechatService;
  29 +
  30 + @Test
  31 + public void sendCustomTest(){
  32 +
  33 + List<String> openIdList = new ArrayList<>();
  34 + openIdList.add("oemqmjl0M38zwP-7wnmhNcQqVMEA");
  35 +
  36 + JSONArray contentArray = new JSONArray();
  37 + JSONObject part = new JSONObject();
  38 + part.put("title","Happy Day");
  39 + part.put("description","Is Really A Happy Day");
  40 + part.put("url","https://m.yohobuy.com/");
  41 + part.put("picurl","http://img11.static.yhbimg.com/yhb-img01/2018/10/09/11/01212cad454186ddd6e9bab47d76dde7f3.png");
  42 + contentArray.add(part);
  43 + JSONObject content = new JSONObject();
  44 + content.put("articles",contentArray);
  45 +
  46 + WechatCustomMsgBO customMsgBO = new WechatCustomMsgBO();
  47 + customMsgBO.setPublicNumCode(2);
  48 + customMsgBO.setOpenIdList(openIdList);
  49 + customMsgBO.setSendKey("123");
  50 + customMsgBO.setFilterDays(0);
  51 + customMsgBO.setMsgType("news");
  52 + customMsgBO.setSendContent(content.toJSONString());
  53 + wechatService.sendWechatCustomMsg(customMsgBO);
  54 + }
  55 +}
@@ -16,4 +16,8 @@ public class Consts { @@ -16,4 +16,8 @@ public class Consts {
16 public static final String WEIXIN_ACCESS_TOKEN_URL = "https://api.weixin.qq.com/cgi-bin/token?grant_type=client_credential&appid=APPID&secret=APPSECRET"; 16 public static final String WEIXIN_ACCESS_TOKEN_URL = "https://api.weixin.qq.com/cgi-bin/token?grant_type=client_credential&appid=APPID&secret=APPSECRET";
17 //微信查询用户信息地址 17 //微信查询用户信息地址
18 public static final String QUERY_USER_INFO_URL = "https://api.weixin.qq.com/cgi-bin/user/info?access_token=ACCESS_TOKEN&openid=OPENID&lang=zh_CN"; 18 public static final String QUERY_USER_INFO_URL = "https://api.weixin.qq.com/cgi-bin/user/info?access_token=ACCESS_TOKEN&openid=OPENID&lang=zh_CN";
  19 + //客服消息发送地址
  20 + public static final String CUSTOM_MSG_SEND_URL = "https://api.weixin.qq.com/cgi-bin/message/custom/send?access_token=";
  21 +
  22 + public static final String CUSTOM_MSG_SEND_NUM_TOPIC = "crm.customMsgSendNum";
19 } 23 }
  1 +package com.yoho.yhmessage.wechat.consumer;
  2 +
  3 +import com.alibaba.fastjson.JSONObject;
  4 +import com.yoho.core.rabbitmq.YhConsumer;
  5 +import com.yoho.service.model.msgcenter.wechat.WechatCustomMsgBO;
  6 +import com.yoho.yhmessage.wechat.service.IWechatService;
  7 +import org.slf4j.Logger;
  8 +import org.slf4j.LoggerFactory;
  9 +import org.springframework.beans.factory.annotation.Autowired;
  10 +import org.springframework.stereotype.Component;
  11 +
  12 +/**
  13 + * 客服消息发送
  14 + * Created by min.ling on 2018/9/29.
  15 + */
  16 +@Component
  17 +public class WechatCustomMsgConsumer implements YhConsumer {
  18 +
  19 + private static final Logger log = LoggerFactory.getLogger(WechatCustomMsgConsumer.class);
  20 +
  21 + @Autowired
  22 + IWechatService wechatService;
  23 +
  24 +
  25 + @Override
  26 + public void handleMessage(Object message) throws Exception {
  27 + log.info("WechatCustomMsgConsumer.handleMessage with message is {}", message);
  28 +
  29 + try {
  30 + WechatCustomMsgBO customMsgBO = JSONObject.parseObject(message.toString(), WechatCustomMsgBO.class);
  31 + // 默认值 有货服务号
  32 + customMsgBO.setPublicNumCode(customMsgBO.getPublicNumCode() == 0 ? 2 : customMsgBO.getPublicNumCode());
  33 + wechatService.sendWechatCustomMsg(customMsgBO);
  34 + } catch (Exception e) {
  35 + log.error("WechatCustomMsgConsumer.handleMessage error with e is {}", e);
  36 + }
  37 + }
  38 +}
@@ -2,6 +2,7 @@ package com.yoho.yhmessage.wechat.service; @@ -2,6 +2,7 @@ package com.yoho.yhmessage.wechat.service;
2 2
3 import com.yoho.error.exception.ServiceException; 3 import com.yoho.error.exception.ServiceException;
4 import com.yoho.service.model.msgcenter.wechat.McWechatBO; 4 import com.yoho.service.model.msgcenter.wechat.McWechatBO;
  5 +import com.yoho.service.model.msgcenter.wechat.WechatCustomMsgBO;
5 6
6 /** 7 /**
7 * 8 *
@@ -24,4 +25,11 @@ public interface IWechatService { @@ -24,4 +25,11 @@ public interface IWechatService {
24 * @throws ServiceException <br> 25 * @throws ServiceException <br>
25 */ 26 */
26 void sendWechatMsg(McWechatBO wechatBO) throws ServiceException; 27 void sendWechatMsg(McWechatBO wechatBO) throws ServiceException;
  28 +
  29 + /**
  30 + * 发送微信客服消息
  31 + * @param customMsgBO
  32 + * @throws ServiceException
  33 + */
  34 + void sendWechatCustomMsg(WechatCustomMsgBO customMsgBO) throws ServiceException;
27 } 35 }
1 package com.yoho.yhmessage.wechat.service.impl; 1 package com.yoho.yhmessage.wechat.service.impl;
2 2
3 import com.alibaba.fastjson.JSONObject; 3 import com.alibaba.fastjson.JSONObject;
  4 +import com.google.common.util.concurrent.RateLimiter;
  5 +import com.yoho.core.common.utils.DateUtil;
  6 +import com.yoho.core.rabbitmq.YhProducer;
4 import com.yoho.error.exception.ServiceException; 7 import com.yoho.error.exception.ServiceException;
5 import com.yoho.service.model.msgcenter.wechat.McWechatBO; 8 import com.yoho.service.model.msgcenter.wechat.McWechatBO;
  9 +
6 import com.yoho.yhmessage.filter.IFilterSensitiveWordService; 10 import com.yoho.yhmessage.filter.IFilterSensitiveWordService;
  11 +
  12 +import com.yoho.service.model.msgcenter.wechat.WechatCustomMsgBO;
  13 +import com.yoho.yhmessage.constants.RedisCacheKeyConstant;
  14 +import com.yoho.yhmessage.redis.PushRedisService;
  15 +
7 import com.yoho.yhmessage.wechat.common.Consts; 16 import com.yoho.yhmessage.wechat.common.Consts;
8 import com.yoho.yhmessage.wechat.service.IWechatService; 17 import com.yoho.yhmessage.wechat.service.IWechatService;
9 import com.yoho.yhmessage.wechat.service.IWeixinAccessTokenService; 18 import com.yoho.yhmessage.wechat.service.IWeixinAccessTokenService;
10 import com.yoho.yhmessage.wechat.service.model.AccessToken; 19 import com.yoho.yhmessage.wechat.service.model.AccessToken;
11 import com.yoho.yhmessage.wechat.util.SendWechatRequest; 20 import com.yoho.yhmessage.wechat.util.SendWechatRequest;
  21 +import org.apache.commons.collections.CollectionUtils;
12 import org.apache.commons.lang3.StringUtils; 22 import org.apache.commons.lang3.StringUtils;
13 import org.slf4j.Logger; 23 import org.slf4j.Logger;
14 import org.slf4j.LoggerFactory; 24 import org.slf4j.LoggerFactory;
15 import org.springframework.beans.factory.annotation.Autowired; 25 import org.springframework.beans.factory.annotation.Autowired;
16 import org.springframework.stereotype.Service; 26 import org.springframework.stereotype.Service;
17 27
  28 +import javax.annotation.Resource;
  29 +import java.text.SimpleDateFormat;
  30 +import java.util.*;
  31 +
18 /** 32 /**
19 * 33 *
20 * <Description> 微信<br> 34 * <Description> 微信<br>
@@ -40,8 +54,22 @@ public class WechatServiceImpl implements IWechatService { @@ -40,8 +54,22 @@ public class WechatServiceImpl implements IWechatService {
40 private SendWechatRequest sendWechatRequest; 54 private SendWechatRequest sendWechatRequest;
41 55
42 @Autowired 56 @Autowired
  57 +
43 private IFilterSensitiveWordService filterSensitiveWordService; 58 private IFilterSensitiveWordService filterSensitiveWordService;
44 59
  60 + private PushRedisService pushRedisService;
  61 +
  62 + private static final String DATE_ID_FORMAT = "yyyyMMdd";
  63 +
  64 + //1s放10个令牌 100ms频率
  65 + private static final RateLimiter rateLimiter = RateLimiter.create(10);
  66 +
  67 + private static final int SEVEN_DAY_HOURS = 168;
  68 +
  69 + @Resource(name = "yhProducer")
  70 + private YhProducer yhProducer;
  71 +
  72 +
45 @Override 73 @Override
46 public void sendWechatMsg(McWechatBO wechatBO) throws ServiceException { 74 public void sendWechatMsg(McWechatBO wechatBO) throws ServiceException {
47 logger.info("begin sendTemplateMsg param {}", wechatBO); 75 logger.info("begin sendTemplateMsg param {}", wechatBO);
@@ -67,7 +95,7 @@ public class WechatServiceImpl implements IWechatService { @@ -67,7 +95,7 @@ public class WechatServiceImpl implements IWechatService {
67 logger.warn("sendWechatMsg with accessToken is null, publicCode is {},wechatBO is {}",wechatBO.getPublicNumberCode(),wechatBO); 95 logger.warn("sendWechatMsg with accessToken is null, publicCode is {},wechatBO is {}",wechatBO.getPublicNumberCode(),wechatBO);
68 return; 96 return;
69 } 97 }
70 - String templateUrl = this.getTemplateUrl(accessToken); 98 + String templateUrl = this.getTemplateUrl(accessToken,Consts.SEND_TEMPLATE_URL);
71 //针对需要展示微信用户昵称的消息，在这里调用微信api查询一把并替换，避免在其他服务调用微信api的复杂行，微信用户昵称统一用${wechatNickname}表示 99 //针对需要展示微信用户昵称的消息，在这里调用微信api查询一把并替换，避免在其他服务调用微信api的复杂行，微信用户昵称统一用${wechatNickname}表示
72 String data = this.addUserNickname(wechatBO, accessToken); 100 String data = this.addUserNickname(wechatBO, accessToken);
73 // 获取请求参数 101 // 获取请求参数
@@ -82,7 +110,7 @@ public class WechatServiceImpl implements IWechatService { @@ -82,7 +110,7 @@ public class WechatServiceImpl implements IWechatService {
82 logger.warn("send wechat template message error, e is {}", JSONObject.toJSONString(result) ); 110 logger.warn("send wechat template message error, e is {}", JSONObject.toJSONString(result) );
83 // 重新调接口获取access token 111 // 重新调接口获取access token
84 AccessToken accessTokenNew = weixinAccessTokenService.getAccessTokenFromUrl(wechatBO.getPublicNumberCode()); 112 AccessToken accessTokenNew = weixinAccessTokenService.getAccessTokenFromUrl(wechatBO.getPublicNumberCode());
85 - templateUrl = this.getTemplateUrl(accessTokenNew); 113 + templateUrl = this.getTemplateUrl(accessTokenNew,Consts.SEND_TEMPLATE_URL);
86 result = sendWechatRequest.sendWechatRequest(wechatBO.getPublicNumberCode(), templateUrl, "POST", sendMsg.toJSONString()); 114 result = sendWechatRequest.sendWechatRequest(wechatBO.getPublicNumberCode(), templateUrl, "POST", sendMsg.toJSONString());
87 } 115 }
88 sendWechatMsgLog.info("send wechat template message {},result {}, templateUrl is {}", sendMsg.toJSONString(), result, templateUrl); 116 sendWechatMsgLog.info("send wechat template message {},result {}, templateUrl is {}", sendMsg.toJSONString(), result, templateUrl);
@@ -116,12 +144,12 @@ public class WechatServiceImpl implements IWechatService { @@ -116,12 +144,12 @@ public class WechatServiceImpl implements IWechatService {
116 * @taskId <br> 144 * @taskId <br>
117 * @return <br> 145 * @return <br>
118 */ 146 */
119 - private String getTemplateUrl(AccessToken accessToken) { 147 + private String getTemplateUrl(AccessToken accessToken,String url) {
120 if (null == accessToken) { 148 if (null == accessToken) {
121 return null; 149 return null;
122 } 150 }
123 logger.info("accessToken is {}", accessToken.getToken()); 151 logger.info("accessToken is {}", accessToken.getToken());
124 - StringBuffer templateUrl = new StringBuffer(Consts.SEND_TEMPLATE_URL); 152 + StringBuffer templateUrl = new StringBuffer(url);
125 templateUrl.append(accessToken.getToken()); 153 templateUrl.append(accessToken.getToken());
126 return templateUrl.toString(); 154 return templateUrl.toString();
127 } 155 }
@@ -129,4 +157,100 @@ public class WechatServiceImpl implements IWechatService { @@ -129,4 +157,100 @@ public class WechatServiceImpl implements IWechatService {
129 private String getQueryUserUrl(String openId, AccessToken accessToken) { 157 private String getQueryUserUrl(String openId, AccessToken accessToken) {
130 return Consts.QUERY_USER_INFO_URL.replace("ACCESS_TOKEN", accessToken.getToken()).replace("OPENID", openId); 158 return Consts.QUERY_USER_INFO_URL.replace("ACCESS_TOKEN", accessToken.getToken()).replace("OPENID", openId);
131 } 159 }
  160 +
  161 +
  162 + @Override
  163 + public void sendWechatCustomMsg(WechatCustomMsgBO customMsgBO) throws ServiceException{
  164 + logger.info("enter sendWechatCustomMsg with sceneKey is {},msgBO is {}",customMsgBO.getSendKey(),customMsgBO.getSendContent());
  165 +
  166 + if(CollectionUtils.isEmpty(customMsgBO.getOpenIdList())){
  167 + logger.warn("sendWechatCustomMsg with sendOpenIds null, sceneKey is {}", customMsgBO.getSendKey());
  168 + return;
  169 + }
  170 + String sceneKey = customMsgBO.getSendKey();
  171 + Set<String> openIdSet = new HashSet<>(customMsgBO.getOpenIdList());
  172 + logger.info("sendWechatCustomMsg with sceneKey is {},send origin size is {}",sceneKey,openIdSet.size());
  173 + //根据过滤条件进行过滤
  174 + if(customMsgBO.getFilterDays() != 0){
  175 + getDaysFilter(customMsgBO, openIdSet);
  176 + }
  177 + //组装发送
  178 + List<String> sendSuccessList = customMsgSend(customMsgBO, openIdSet);
  179 + if(CollectionUtils.isEmpty(sendSuccessList)){
  180 + logger.warn("sendWechatCustomMsg with sendSuccess size is 0,sceneKey is {}",customMsgBO.getSendKey());
  181 + return;
  182 + }
  183 + logger.info("end sendWechatCustomMsg with sceneKey is {},successSize is {}",customMsgBO.getSendKey(),sendSuccessList.size());
  184 + //成功记录存redis
  185 + String nowDate = DateUtil.getToday(DATE_ID_FORMAT);
  186 + String redisKeyStr = RedisCacheKeyConstant.WECAHT_CUSTOM_SEND_KEY + customMsgBO.getPublicNumCode() + '_' + nowDate;
  187 + //成功记录存7天
  188 + pushRedisService.setListValue(redisKeyStr,sendSuccessList,SEVEN_DAY_HOURS);
  189 + logger.info("sendWechatCustomMsg with setRedis end, sceneKey is {},successSize is {}",customMsgBO.getSendKey(),sendSuccessList.size());
  190 + //发送成功数给到crm
  191 + sendMqToCrm(customMsgBO, sendSuccessList);
  192 + }
  193 +
  194 + private void sendMqToCrm(WechatCustomMsgBO customMsgBO, List<String> sendSuccessList) {
  195 + JSONObject object = new JSONObject();
  196 + object.put("activityId",customMsgBO.getSendKey());
  197 + object.put("sendNum",sendSuccessList.size());
  198 + yhProducer.send(Consts.CUSTOM_MSG_SEND_NUM_TOPIC,object);
  199 + }
  200 +
  201 + private void getDaysFilter(WechatCustomMsgBO customMsgBO, Set<String> openIdSet) {
  202 + int filterDayReal = customMsgBO.getFilterDays() - 1;
  203 + //根据间隔天数计算出需要过滤到的最早的一天
  204 + Calendar calendar = Calendar.getInstance();
  205 + calendar.add(Calendar.DATE, -filterDayReal);
  206 + Calendar c = Calendar.getInstance();
  207 + c.setTime(calendar.getTime());
  208 + int day = c.get(Calendar.DATE);
  209 + for (int i = 0; i <= filterDayReal; i++){
  210 + //从最早日期开始往后累加到每个日期，每天过滤
  211 + c.set(Calendar.DATE, day + i);
  212 + String nowDate = new SimpleDateFormat(DATE_ID_FORMAT).format(c.getTime());
  213 + //根据日期获取当天发送客服消息的记录
  214 + String redisKeyStr = RedisCacheKeyConstant.WECAHT_CUSTOM_SEND_KEY + customMsgBO.getPublicNumCode() + '_' + nowDate;
  215 + pushRedisService.getSetFilter(openIdSet,redisKeyStr);
  216 + logger.info("sendWechatCustomMsg with day filter,sceneKey is {},redisKeyStr is {},nowSize is {}",customMsgBO.getSendKey(),redisKeyStr,openIdSet.size());
  217 + if(openIdSet.size() == 0){
  218 + break;
  219 + }
  220 + }
  221 + }
  222 +
  223 +
  224 + private List<String> customMsgSend(WechatCustomMsgBO customMsgBO,Set<String> openIdSet){
  225 + logger.info("begin customMsgSend with sceneKey is {},send openId size is {}",customMsgBO.getSendKey(),openIdSet.size());
  226 + if(CollectionUtils.isEmpty(openIdSet)){
  227 + logger.warn("customMsgSend with openIdSet is null,sceneKey is {}",customMsgBO.getSendKey());
  228 + return new ArrayList<>();
  229 + }
  230 + // 获取token 和接口url
  231 + AccessToken accessToken = weixinAccessTokenService.queryWeixinAccessToken(customMsgBO.getPublicNumCode());
  232 + //若仍未获取到 token 则返回
  233 + if(accessToken == null){
  234 + logger.warn("customMsgSend with accessToken is null,sendKey is {}",customMsgBO.getSendKey());
  235 + return new ArrayList<>();
  236 + }
  237 + List<String> successList = new ArrayList<>();
  238 + String templateUrl = this.getTemplateUrl(accessToken, Consts.CUSTOM_MSG_SEND_URL);
  239 + //组装消息内容
  240 + JSONObject sendDataObject = new JSONObject();
  241 + sendDataObject.put("msgtype",customMsgBO.getMsgType());
  242 + sendDataObject.put(customMsgBO.getMsgType(),JSONObject.parseObject(customMsgBO.getSendContent()));
  243 + openIdSet.forEach(openId -> {
  244 + //限制100ms调一次
  245 + rateLimiter.acquire();
  246 + sendDataObject.put("touser", openId);
  247 + JSONObject result = sendWechatRequest.sendWechatRequest(customMsgBO.getPublicNumCode(), templateUrl, "POST", sendDataObject.toJSONString());
  248 + if (result != null && result.getIntValue("errcode") == 0) {
  249 + successList.add(openId);
  250 + }
  251 + sendWechatMsgLog.info("customMsgSend with sceneKey {},openId is {},result {}, content is {}", customMsgBO.getSendKey(), openId, result, sendDataObject.toJSONString());
  252 + });
  253 + return successList;
  254 + }
  255 +
132 } 256 }