定时客服多线程

src/main/java/com/yaoyaozw/customer/controller/TestController.java

@@ -25,7 +25,7 @@ public class TestController {
 
         Calendar calendar = Calendar.getInstance();
         calendar.set(Calendar.AM_PM,0);
-        calendar.set(Calendar.HOUR,20);
+        calendar.set(Calendar.HOUR,16);
         calendar.set(Calendar.MINUTE,0);
         calendar.set(Calendar.SECOND, 0);
         calendar.set(Calendar.MILLISECOND, 0);
@@ -34,4 +34,6 @@ public class TestController {
         integrationRequestDTO.setRequestDate(calendar.getTime());
         customerGraphicsService.sendCustomerMessage(integrationRequestDTO);
     }
+
+
 }

src/main/java/com/yaoyaozw/customer/entity/CustomerGraphics.java

@@ -107,6 +107,13 @@ public class CustomerGraphics implements Serializable {
     @TableLogic
     private Integer isDeleted;
 
+    @TableField(exist = false)
+    private Integer sendCount;
+
+    public void updateCount(){
+       sendCount++;
+    }
+
     private static final long serialVersionUID = 1L;
 
     public static final String COL_ID = "id";

src/main/java/com/yaoyaozw/customer/mapper/AccountOrderMapper.java

@@ -12,7 +12,7 @@ import java.util.List;
 @Mapper
 public interface AccountOrderMapper extends BaseMapper<AccountOrder> {
 
-    List<AccountOrder> findNewestRuntimeAccountOrder();
+    List<AccountOrder> findNewestRuntimeAccountOrder(Date requestDate);
 
     List<UserAvgAmountVO> findMonthDailyAccountOrder();
 }
\ No newline at end of file

src/main/java/com/yaoyaozw/customer/mapper/AuthorizerTokenMapper.java

@@ -5,8 +5,13 @@ import com.yaoyaozw.customer.entity.AuthorizerToken;
 import org.apache.ibatis.annotations.Mapper;
 import org.apache.ibatis.annotations.Param;
 
+import java.util.List;
+import java.util.Set;
+
 @Mapper
 public interface AuthorizerTokenMapper extends BaseMapper<AuthorizerToken> {
 
     AuthorizerToken findByAppid(@Param("appid")String appid);
+
+    List<AuthorizerToken> findByAppidIn(@Param("appidList") Set<String> appidList);
 }
\ No newline at end of file

src/main/java/com/yaoyaozw/customer/scheduling/SchedulingTask.java

@@ -36,7 +36,7 @@ public class SchedulingTask {
 
     @Autowired
     private CustomerDelayPublishService customerDelayPublishService;
-    private RegisterUserEntityService registerUserEntityService;
+
     @Autowired
     private RedisTemplate redisTemplate;
     @Autowired

src/main/java/com/yaoyaozw/customer/service/AuthorizerTokenService.java

@@ -2,8 +2,13 @@ package com.yaoyaozw.customer.service;
 
 import com.yaoyaozw.customer.entity.AuthorizerToken;
 import com.baomidou.mybatisplus.extension.service.IService;
+
+import java.util.List;
+import java.util.Set;
+
 public interface AuthorizerTokenService extends IService<AuthorizerToken>{
 
     AuthorizerToken findTokenByAppid(String appid);
 
+    List<AuthorizerToken> findTokenByAppidList(Set<String> appidList);
 }

src/main/java/com/yaoyaozw/customer/service/CustomerDelayPublishService.java

@@ -9,6 +9,9 @@ import com.baomidou.mybatisplus.extension.service.IService;
 * @date 2022/9/28 18:51
 */
 public interface CustomerDelayPublishService extends IService<CustomerDelayPublish>{
-
+    /**
+     * 延时客服发送
+     * @param integrationRequestDTO
+     */
     public void sendCustomerDelayMessage(IntegrationRequestDTO integrationRequestDTO);
 }

src/main/java/com/yaoyaozw/customer/service/impl/AccountOrderServiceImpl.java

@@ -35,7 +35,7 @@ public class AccountOrderServiceImpl extends ServiceImpl<AccountOrderMapper, Acc
     @Override
     public void activeUserByOrder(IntegrationRequestDTO integrationRequestDTO) {
         //获取runtime订单最新一版数据
-        List<AccountOrder> runtimeOrders= baseMapper.findNewestRuntimeAccountOrder();
+        List<AccountOrder> runtimeOrders= baseMapper.findNewestRuntimeAccountOrder(integrationRequestDTO.getRequestDate());
 
         Set<String> openIdSet = runtimeOrders.stream().map(AccountOrder::getOpenId).collect(Collectors.toSet());
 
@@ -58,7 +58,7 @@ public class AccountOrderServiceImpl extends ServiceImpl<AccountOrderMapper, Acc
             Double sum=registerUser.getPayAmount();
 
             for (AccountOrder accountOrder : accountOrders) {
-                //最近时间
+                //最近活跃时间
                 lastFinishTime = (lastFinishTime != null && lastFinishTime.compareTo(accountOrder.getFinishTime()) > 0) ? lastFinishTime : accountOrder.getFinishTime();
                 //最大类型
                 type = Math.max(accountOrder.getType() + 1, type);
@@ -67,6 +67,7 @@ public class AccountOrderServiceImpl extends ServiceImpl<AccountOrderMapper, Acc
             //计算用户信息
             if (registerUser.getLastActive()==null){
                 //排期重置....
+                delayCustomerManage(registerUser,lastFinishTime,delayMessageMap);
             }
             //活跃时间重置
             registerUser.setLastActive(lastFinishTime);
@@ -76,10 +77,8 @@ public class AccountOrderServiceImpl extends ServiceImpl<AccountOrderMapper, Acc
             registerUser.setPayNum(registerUser.getPayNum()+accountOrders.size());
             //累计金额
             registerUser.setPayAmount(sum);
-
         }
         registerUserEntityService.updateBatchById(userList);
-
     }
 
 
@@ -136,7 +135,6 @@ public class AccountOrderServiceImpl extends ServiceImpl<AccountOrderMapper, Acc
                 registerUserPublish.setCustomerPublish(publishTime);
             }
 
-
         }
     }
 }

src/main/java/com/yaoyaozw/customer/service/impl/AuthorizerTokenServiceImpl.java

@@ -3,6 +3,8 @@ package com.yaoyaozw.customer.service.impl;
 import org.springframework.stereotype.Service;
 import javax.annotation.Resource;
 import java.util.List;
+import java.util.Set;
+
 import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
 import com.yaoyaozw.customer.entity.AuthorizerToken;
 import com.yaoyaozw.customer.mapper.AuthorizerTokenMapper;
@@ -16,4 +18,9 @@ public class AuthorizerTokenServiceImpl extends ServiceImpl<AuthorizerTokenMappe
         return baseMapper.findByAppid(appid);
 
     }
+
+    @Override
+    public List<AuthorizerToken> findTokenByAppidList(Set<String> appidList) {
+        return baseMapper.findByAppidIn(appidList);
+    }
 }

src/main/java/com/yaoyaozw/customer/service/impl/CustomerDelayPublishServiceImpl.java

@@ -42,10 +42,8 @@ public class CustomerDelayPublishServiceImpl extends ServiceImpl<CustomerDelayPu
     private WeChatService weChatService;
 
 
-
     @Override
     public void sendCustomerDelayMessage(IntegrationRequestDTO integrationRequestDTO) {
-
         //读取延时客服排期人
         List<CustomerDelayPublish> allPostUser = list(new QueryWrapper<CustomerDelayPublish>().eq(CustomerDelayPublish.COL_CUSTOMER_PUBLISH,integrationRequestDTO.getRequestDate()));
         //涉及的appid
@@ -54,12 +52,9 @@ public class CustomerDelayPublishServiceImpl extends ServiceImpl<CustomerDelayPu
         if (!allPostUser.isEmpty()){
             //号-用户
             Map<String, List<CustomerDelayPublish>> userMap = allPostUser.stream().collect(Collectors.groupingBy(CustomerDelayPublish::getAppId));
-
+            //获取所有延时客服
             List<CustomerDelayItemVO> allDelayCustomerMessage = customerGraphicsDelayService.findAllDelayCustomerMessage(appidSet,null);
-/*
-            List<AuthorizerToken> tokenList = authorizerTokenService.list();
-            Map<String, String> tokenMap = tokenList.stream().collect(Collectors.toMap(AuthorizerToken::getAuthorizerAppid, AuthorizerToken::getAuthorizerAccessToken));
-*/
+
             if (allDelayCustomerMessage!=null&&!allDelayCustomerMessage.isEmpty()){
 
                 List<Future<CustomerDelayPublish>>futureList =new ArrayList<>();

src/main/java/com/yaoyaozw/customer/service/impl/CustomerGraphicsServiceImpl.java

@@ -239,6 +239,7 @@ public class CustomerGraphicsServiceImpl extends ServiceImpl<CustomerGraphicsMap
         List<CustomerGraphics> customerGraphicsList = list(new QueryWrapper<CustomerGraphics>().eq(CustomerGraphics.COL_POST_TIME, requestDate));
 
         for (CustomerGraphics customerGraphics : customerGraphicsList) {
+            customerGraphics.setSendCount(0);
             //人群包id
             Long packId = customerGraphics.getPackId();
             //根据人群包找人,并按appId分组

src/main/java/com/yaoyaozw/customer/service/wechat/service/WeChatRestService.java

 package com.yaoyaozw.customer.service.wechat.service;
 
+import cn.hutool.json.JSONUtil;
+import com.alibaba.fastjson.JSONObject;
+import com.alibaba.nacos.client.utils.JSONUtils;
+import com.yaoyaozw.customer.service.wechat.entity.WeChatResponseEntity;
+import org.apache.http.HttpEntity;
+import org.apache.http.HttpResponse;
+import org.apache.http.client.HttpClient;
+import org.apache.http.client.methods.HttpPost;
+import org.apache.http.conn.ClientConnectionManager;
+import org.apache.http.entity.StringEntity;
+import org.apache.http.impl.client.HttpClients;
 import org.springframework.http.ResponseEntity;
 import org.springframework.stereotype.Service;
 import org.springframework.util.MultiValueMap;
@@ -8,6 +19,8 @@ import org.springframework.web.client.RestTemplate;
 import org.springframework.web.util.UriComponents;
 import org.springframework.web.util.UriComponentsBuilder;
 
+import java.io.BufferedReader;
+import java.io.InputStreamReader;
 import java.net.URI;
 import java.net.URISyntaxException;
 import java.util.Map;
@@ -55,4 +68,33 @@ public class WeChatRestService {
         return result;
 
     }
+
+    public <T>Object httpPostRequest(String fullPath, Object requestEntity,Class<T> responseEntity)throws Exception{
+
+        HttpClient defaultHttpClient = HttpClients.createDefault();
+        UriComponentsBuilder uriComponentsBuilder = UriComponentsBuilder.fromHttpUrl(fullPath);
+        HttpPost httpPost = new HttpPost(uriComponentsBuilder.toUriString());
+        httpPost.setHeader("Content-Type", "application/json;charset=utf-8");
+        if (requestEntity != null) {
+            //转换为json格式并打印
+            String json = JSONObject.toJSONString(requestEntity);
+            HttpEntity httpEntity = new StringEntity(json, "utf-8");
+            httpPost.setEntity(httpEntity);
+        }
+        HttpResponse httpResponse = defaultHttpClient.execute(httpPost);
+        if (httpResponse.getStatusLine().getStatusCode() != 200) {
+            String errorLog = "请求失败，errorCode:" + httpResponse.getStatusLine().getStatusCode();
+            throw new Exception(fullPath + errorLog);
+        }
+        //读取返回信息
+        String output;
+        BufferedReader bufferedReader = new BufferedReader(new InputStreamReader(httpResponse.getEntity().getContent(), "utf-8"));
+        StringBuilder stringBuilder = new StringBuilder();
+        while ((output = bufferedReader.readLine()) != null) {
+            stringBuilder.append(output);
+        }
+        String s = stringBuilder.toString();
+        return JSONUtils.deserializeObject(s, responseEntity);
+
+    }
 }

src/main/java/com/yaoyaozw/customer/service/wechat/service/WeChatServiceImpl.java

 package com.yaoyaozw.customer.service.wechat.service;
 
 
+import com.yaoyaozw.customer.constants.CustomerCommonConstant;
 import com.yaoyaozw.customer.entity.AuthorizerToken;
 import com.yaoyaozw.customer.entity.CustomerDelayPublish;
 import com.yaoyaozw.customer.entity.CustomerGraphics;
@@ -23,10 +24,7 @@ import java.util.ArrayList;
 import java.util.Date;
 import java.util.List;
 import java.util.Map;
-import java.util.concurrent.Future;
-import java.util.concurrent.LinkedBlockingQueue;
-import java.util.concurrent.ThreadPoolExecutor;
-import java.util.concurrent.TimeUnit;
+import java.util.concurrent.*;
 import java.util.stream.Collectors;
 
 @Service
@@ -42,12 +40,18 @@ public class WeChatServiceImpl implements WeChatService{
 
     private static final ThreadPoolExecutor EXECUTOR = new ThreadPoolExecutor(20, 100, 60L, TimeUnit.SECONDS, new LinkedBlockingQueue<>(2),new ThreadPoolExecutor.CallerRunsPolicy());
 
+    private final static Object LOCK=new Object();
+
     @Autowired
     private WeChatRestService weChatRestService;
 
     @Async("myExecutor")
     @Override
     public Future<CustomerDelayPublish> sendCustomerDelayMessage(String token, CustomerDelayPublish user, Map<Integer,CustomerDelayItemVO> delaySortMap) {
+
+
+
+
         //当前时间批次发文素材的用户关注时间一致
         Long subscribeTimestamp = user.getGmtCreate().getTime();
 
@@ -72,12 +76,15 @@ public class WeChatServiceImpl implements WeChatService{
         //构建请求参数（文本/图文）
         WeChatCustomerRequestEntity customerRequest = buildCustomerRequest(customerGraphics,referralEntityVo);
 
+        //计数器
+        CountDownLatch latch = new CountDownLatch(openidList.size());
+
         if (customerRequest!=null){
 
             for (int i = 0; i < openidList.size(); i+=THREAD_NUM) {
 
                 List<CrowdPackageUserVO> subOpenidList = openidList.subList(i, Math.min(i + THREAD_NUM, openidList.size()));
-                //TODO:配置多线程+测试
+                //TODO:配置多线程+测试+记录发送量
                 EXECUTOR.execute(()->{
 
                     for (CrowdPackageUserVO crowdPackageUserVO : subOpenidList) {
@@ -87,15 +94,18 @@ public class WeChatServiceImpl implements WeChatService{
                         try {
                             System.err.println(customerRequest.getTouser());
 
-                            ResponseEntity<WeChatResponseEntity> response = weChatRestService.sendPostRequest(fullPath, WeChatResponseEntity.class, customerRequest);
-
-                            WeChatResponseEntity body = response.getBody();
-
-                            System.err.println(body);
+                            WeChatResponseEntity response=(WeChatResponseEntity)weChatRestService.httpPostRequest(fullPath,  customerRequest,WeChatResponseEntity.class);
+                            //线程锁
+                            synchronized (LOCK){
+                                customerGraphics.updateCount();
+                            }
+                            System.err.println(response);
 
                         } catch (Exception e) {
 
                             e.printStackTrace();
+                        }finally {
+                            latch.countDown();
                         }
                     }
                 });
@@ -104,6 +114,7 @@ public class WeChatServiceImpl implements WeChatService{
     }
 
 
+
     private WeChatCustomerRequestEntity buildCustomerRequest(CustomerGraphics customerGraphics,List<ReferralEntityVo> urlList){
         //类型判断
         if (CUSTOMER_TEXT.equals(customerGraphics.getType())){
@@ -113,24 +124,24 @@ public class WeChatServiceImpl implements WeChatService{
             if (customerGraphics.getReferralSize()!=null&&customerGraphics.getReferralSize().equals(sortReferral.size())){
 
                 String content = customerGraphics.getContent();
-                //TODO:配置文本，多条链接替换进content
+
                 for (Map.Entry<Integer, String> replaceReferral : sortReferral.entrySet()) {
 
                     Integer sort = replaceReferral.getKey();
 
                     String url = replaceReferral.getValue();
                     //替换占位符
-                    content=content.replace("占位符常量" + sort, url);
+                    content=content.replace(CustomerCommonConstant.CUSTOMER_TEXT_URL_PLACEHOLDER + sort, url);
                 }
                 return new WeChatCustomerRequestEntity(CUSTOMER_TEXT,content);
+
             }else{
+
                 return null;
             }
-
         }else if (CUSTOMER_NEWS.equals(customerGraphics.getType())){
 
             return new WeChatCustomerRequestEntity(CUSTOMER_NEWS,customerGraphics.getExtendTitle(),customerGraphics.getContent(),urlList.get(0).getReferral(),customerGraphics.getCoverUrl());
-
         }
         return null;
     }

src/main/resources/mapper/AccountOrderMapper.xml

@@ -24,11 +24,12 @@
       ao.id, ao.setup_id, ao.profit, ao.finish_time, ao.`type`, ao.open_id, ao.user_subscribe_time
     FROM
       account_order ao
+<!--
         INNER JOIN ( SELECT  max( gmt_create) AS maxDate FROM account_order where data_frequency=0 and gmt_create>CURRENT_DATE  ) st ON
 
         ao.gmt_create = st.maxDate
-
-    where  ao.data_frequency=0 and ao.open_id LIKE 'o%'
+-->
+    where  ao.data_frequency=0 and ao.open_id LIKE 'o%' and ao.gmt_create=#{requestDate}
 
   </select>
 

src/main/resources/mapper/AuthorizerTokenMapper.xml

@@ -18,4 +18,16 @@
     select id,authorizer_access_token,authorizer_appid from authorizer_token where authorizer_appid=#{appid}
 
   </select>
+
+  <select id="findByAppidIn" resultMap="BaseResultMap">
+
+    select id,authorizer_access_token,authorizer_appid from authorizer_token
+
+    where authorizer_appid in
+
+    <foreach collection="appidList" item="appid" open="(" close=")" separator=",">
+    #{appid}
+  </foreach>
+
+  </select>
 </mapper>
\ No newline at end of file