服务号分组群发支持配置发送全部粉丝

long-article-server/src/main/java/com/tzld/piaoquan/longarticle/dao/mapper/crawler/ArticleUseGroupMapper.java

@@ -34,6 +34,10 @@ public interface ArticleUseGroupMapper {
 
     List<String> selectOpenIds(@Param("ghId") String ghId, @Param("count") Integer count);
 
+    List<ArticleUseGroup> selectAllOpenIdsByPage(@Param("ghId") String ghId,
+                                                 @Param("lastId") Long lastId,
+                                                 @Param("pageSize") Integer pageSize);
+
     List<String> selectOpenIdsByRemainingCount(@Param("ghId") String ghId, @Param("remainingCount") Integer remainingCount);
 
     Integer selectFansBeforePublishCount(@Param("ghId") String ghId);
@@ -46,4 +50,4 @@ public interface ArticleUseGroupMapper {
      */
     int batchDecreaseRemainingCount(@Param("gzhId") String gzhId, @Param("openIdList") List<String> openIdList);
 
-}
+}

long-article-server/src/main/java/com/tzld/piaoquan/longarticle/service/local/impl/CoreServiceImpl.java

@@ -20,6 +20,7 @@ import com.tzld.piaoquan.longarticle.model.dto.RecallVideoScoreParam;
 import com.tzld.piaoquan.longarticle.model.dto.RecallVideoScoreVO;
 import com.tzld.piaoquan.longarticle.model.dto.VideoArticleMatchQueryParam;
 import com.tzld.piaoquan.longarticle.model.dto.VideoArticleMatchResult;
+import com.tzld.piaoquan.longarticle.model.po.crawler.ArticleUseGroup;
 import com.tzld.piaoquan.longarticle.model.po.crawler.FwhDailyPublishDetail;
 import com.tzld.piaoquan.longarticle.model.po.longarticle.*;
 import com.tzld.piaoquan.longarticle.model.vo.*;
@@ -132,6 +133,16 @@ public class CoreServiceImpl implements CoreService {
     @ApolloJsonValue("${recall.video.config.codes:[\"VIDEO_TOPIC\",\"VIDEO_TITLE\"]}")
     private List<String> recallVideoConfigCodes;
 
+    /**
+     * 服务号分组群发时发送全部粉丝的账号列表。
+     * 支持配置 accountId、accountName、ghId；配置 all 时对所有服务号生效。
+     */
+    @ApolloJsonValue("${batch.group.publish.all.fans.accounts:[]}")
+    private List<String> batchGroupPublishAllFansAccounts;
+
+    @Value("${batch.group.publish.all.fans.page-size:10000}")
+    private Integer batchGroupPublishAllFansPageSize;
+
     @Value("${recall.replace.title.cover.switch:true}")
     private Boolean replaceRecallTitleCoverSwitch;
 
@@ -1302,9 +1313,29 @@ public class CoreServiceImpl implements CoreService {
                 return;
             }
         }
+        if (isBatchGroupPublishAllFansAccount(planAccount)) {
+            // 游标分页查询并逐批发送，避免百万级 openId 一次性加载到内存
+            pushAllFansByPage(planAccount, gzhPushParam, pushContentList, sendIds);
+            return;
+        }
+
+        List<String> sendOpenIds = getLimitedSendOpenIds(planAccount);
+        if (CollectionUtils.isEmpty(sendOpenIds)) {
+            return;
+        }
+        saveFwhDailyPublishDetail(planAccount);
+        List<PublishContent> publishContentList = publicContentService.getPublishContentById(sendIds);
+        int group = 1;
+        for (List<String> batch : Lists.partition(sendOpenIds, 10000)) {
+            pushBatchGroup(planAccount, gzhPushParam, pushContentList, publishContentList, batch, group);
+            group++;
+        }
+    }
+
+    private List<String> getLimitedSendOpenIds(PlanAccount planAccount) {
+        List<String> sendOpenIds = new ArrayList<>();
         int remainingDaysInCurrentMonth = DateUtil.getRemainingDaysInCurrentMonth();
         Integer remainingCount;
-        List<String> sendOpenIds = new ArrayList<>();
         if (remainingDaysInCurrentMonth <= 4) {
             remainingCount = articleUseGroupMapper.selectRemainingCountByGzhId(planAccount.getGhId(), remainingDaysInCurrentMonth);
             List<String> remainingCountOpenIds = articleUseGroupMapper.selectOpenIdsByRemainingCount(planAccount.getGhId(), remainingDaysInCurrentMonth);
@@ -1316,9 +1347,47 @@ public class CoreServiceImpl implements CoreService {
             List<String> needSendOpenIds = articleUseGroupMapper.selectOpenIds(planAccount.getGhId(), remainingCount / remainingDaysInCurrentMonth);
             sendOpenIds.addAll(needSendOpenIds);
         }
-        if (CollectionUtils.isEmpty(sendOpenIds)) {
-            return;
+        return sendOpenIds;
+    }
+
+    private void pushAllFansByPage(PlanAccount planAccount, CreateBatchGroupPushTaskParam gzhPushParam,
+                                   List<PushContentParam> pushContentList, List<Long> sendIds) {
+        int pageSize = Math.min(10000, Math.max(1,
+                Optional.ofNullable(batchGroupPublishAllFansPageSize).orElse(10000)));
+        long lastId = 0L;
+        int group = 1;
+        boolean detailSaved = false;
+        List<PublishContent> publishContentList = publicContentService.getPublishContentById(sendIds);
+        while (!Thread.currentThread().isInterrupted()) {
+            List<ArticleUseGroup> page;
+            try {
+                page = articleUseGroupMapper.selectAllOpenIdsByPage(planAccount.getGhId(), lastId, pageSize);
+            } catch (RuntimeException e) {
+                // SQL 配置了单页 30 秒超时；超时后结束当前账号，避免长期占用 core 工作线程
+                log.error("query all fans timeout or error, accountId={}, ghId={}, lastId={}, pageSize={}",
+                        planAccount.getAccountId(), planAccount.getGhId(), lastId, pageSize, e);
+                return;
+            }
+            if (CollectionUtils.isEmpty(page)) {
+                return;
+            }
+            if (!detailSaved) {
+                saveFwhDailyPublishDetail(planAccount);
+                detailSaved = true;
+            }
+            List<String> batch = page.stream().map(ArticleUseGroup::getOpenId).collect(Collectors.toList());
+            pushBatchGroup(planAccount, gzhPushParam, pushContentList, publishContentList, batch, group);
+            lastId = page.get(page.size() - 1).getId();
+            group++;
+            if (page.size() < pageSize) {
+                return;
+            }
         }
+        log.warn("query all fans interrupted, accountId={}, ghId={}, lastId={}",
+                planAccount.getAccountId(), planAccount.getGhId(), lastId);
+    }
+
+    private void saveFwhDailyPublishDetail(PlanAccount planAccount) {
         Integer count = articleUseGroupMapper.selectFansBeforePublishCount(planAccount.getGhId());
         FwhDailyPublishDetail fwhDailyPublishDetail = new FwhDailyPublishDetail();
         fwhDailyPublishDetail.setAccountName(planAccount.getAccountName());
@@ -1327,70 +1396,81 @@ public class CoreServiceImpl implements CoreService {
         fwhDailyPublishDetail.setFansBeforePublish(count);
         fwhDailyPublishDetail.setPublishTimesBeforePublish(articleUseGroupMapper.selectRemainingCountByGzhId(planAccount.getGhId(), null));
         fwhDailyPublishDetailMapper.insertSelective(fwhDailyPublishDetail);
-        int batchSize = 10000; // 每组大小
-        int totalSize = sendOpenIds.size();
-        List<PublishContent> publishContentList = publicContentService.getPublishContentById(sendIds);
-        int group = 1;
-        for (int i = 0; i < totalSize; i += batchSize) {
-            int toIndex = Math.min(i + batchSize, totalSize);
-            // 获取当前批次的数据
-            int batchInsertSize = 1000;
-            List<GroupSendOpenId> batchInsertList = new ArrayList<>(batchInsertSize);
-            List<String> batch = sendOpenIds.subList(i, toIndex);
-            for (String openId : batch) {
-                GroupSendOpenId groupSendOpenId = new GroupSendOpenId();
-                groupSendOpenId.setGhId(planAccount.getGhId());
-                groupSendOpenId.setPublishDate(DateUtil.getThatDayDateString());
-                groupSendOpenId.setUserGroupId(group);
-                groupSendOpenId.setOpenId(openId);
-                batchInsertList.add(groupSendOpenId);
-                // 当批次大小达到1000时执行插入，并清空列表
-                if (batchInsertList.size() >= batchInsertSize) {
-                    groupSendOpenIdMapper.batchInsert(batchInsertList);
-                    batchInsertList.clear();
-                }
-            }
-            // 处理剩余的记录（如果有）
-            if (!batchInsertList.isEmpty()) {
-                groupSendOpenIdMapper.batchInsert(batchInsertList);
-            }
+    }
 
+    private void pushBatchGroup(PlanAccount planAccount, CreateBatchGroupPushTaskParam gzhPushParam,
+                                List<PushContentParam> pushContentList, List<PublishContent> publishContentList,
+                                List<String> batch, int group) {
+        int batchInsertSize = 1000;
+        List<GroupSendOpenId> batchInsertList = new ArrayList<>(batchInsertSize);
+        for (String openId : batch) {
+            GroupSendOpenId groupSendOpenId = new GroupSendOpenId();
+            groupSendOpenId.setGhId(planAccount.getGhId());
+            groupSendOpenId.setPublishDate(DateUtil.getThatDayDateString());
+            groupSendOpenId.setUserGroupId(group);
+            groupSendOpenId.setOpenId(openId);
+            batchInsertList.add(groupSendOpenId);
+            if (batchInsertList.size() >= batchInsertSize) {
+                groupSendOpenIdMapper.batchInsert(batchInsertList);
+                batchInsertList.clear();
+            }
+        }
+        if (!batchInsertList.isEmpty()) {
+            groupSendOpenIdMapper.batchInsert(batchInsertList);
+        }
+
+        CreateBatchGroupItemParam createBatchGroupItemParam = new CreateBatchGroupItemParam();
+        createBatchGroupItemParam.setPushContentList(pushContentList);
+        createBatchGroupItemParam.setUserGroupId(String.valueOf(group));
+        createBatchGroupItemParam.setOpenIds(batch);
+        gzhPushParam.setBatchGroupItems(Collections.singletonList(createBatchGroupItemParam));
+        // 百万粉丝场景避免将全部 openId 写入日志，防止日志序列化拖慢 core 工作线程
+        log.info("batch group push planId={}, accountId={}, group={}, openIdCount={}, contentCount={}",
+                planAccount.getPlanId(), planAccount.getAccountId(), group, batch.size(), pushContentList.size());
+        String pushId = aigcService.createBatchGroupPushTask(gzhPushParam);
+        log.info("pushId = {}", pushId);
+        for (PublishContent publishContent : publishContentList) {
+            PublishContent publishContent1 = new PublishContent();
+            BeanUtils.copyProperties(publishContent, publishContent1);
+            publishContent1.setId(null);
+            publishContent1.setPushId(pushId);
+            publishContent1.setStatus(1);
+            publishContent1.setUserGroupId(group);
+            publishContent1.setRootId(publishContent.getId());
+            publishContent1.setCreateTime(null);
+            publishContent1.setUpdateTime(null);
+            publishContentMapper.insertSelective(publishContent1);
+            GroupSendResult groupSendResult = new GroupSendResult();
+            groupSendResult.setAccountName(planAccount.getAccountName());
+            groupSendResult.setGhId(planAccount.getGhId());
+            groupSendResult.setUserGroupId(group);
+            groupSendResult.setContentId(publishContent.getSourceId());
+            groupSendResult.setPushId(pushId);
+            groupSendResult.setTraceId(publishContent.getTraceId());
+            groupSendResult.setPublishContentId(publishContent.getPublishContentId());
+            groupSendResult.setPublishDate(DateUtil.getThatDayDateString());
+            groupSendResult.setSentCount(batch.size());
+            groupSendResultMapper.insertSelective(groupSendResult);
+        }
+    }
 
-            List<CreateBatchGroupItemParam> createBatchGroupItemParams = new ArrayList<>();
-            CreateBatchGroupItemParam createBatchGroupItemParam = new CreateBatchGroupItemParam();
-            createBatchGroupItemParam.setPushContentList(pushContentList);
-            createBatchGroupItemParam.setUserGroupId(String.valueOf(group));
-            createBatchGroupItemParam.setOpenIds(batch);
-            createBatchGroupItemParams.add(createBatchGroupItemParam);
-            gzhPushParam.setBatchGroupItems(createBatchGroupItemParams);
-            log.info("gzhPushParam={}", gzhPushParam);
-            String pushId = aigcService.createBatchGroupPushTask(gzhPushParam);
-            log.info("pushId = {}", pushId);
-            for (PublishContent publishContent : publishContentList) {
-                PublishContent publishContent1 = new PublishContent();
-                BeanUtils.copyProperties(publishContent, publishContent1);
-                publishContent1.setId(null);
-                publishContent1.setPushId(pushId);
-                publishContent1.setStatus(1);
-                publishContent1.setUserGroupId(group);
-                publishContent1.setRootId(publishContent.getId());
-                publishContent1.setCreateTime(null);
-                publishContent1.setUpdateTime(null);
-                publishContentMapper.insertSelective(publishContent1);
-                GroupSendResult groupSendResult = new GroupSendResult();
-                groupSendResult.setAccountName(planAccount.getAccountName());
-                groupSendResult.setGhId(planAccount.getGhId());
-                groupSendResult.setUserGroupId(group);
-                groupSendResult.setContentId(publishContent.getSourceId());
-                groupSendResult.setPushId(pushId);
-                groupSendResult.setTraceId(publishContent.getTraceId());
-                groupSendResult.setPublishContentId(publishContent.getPublishContentId());
-                groupSendResult.setPublishDate(DateUtil.getThatDayDateString());
-                groupSendResult.setSentCount(batch.size());
-                groupSendResultMapper.insertSelective(groupSendResult);
+    private boolean isBatchGroupPublishAllFansAccount(PlanAccount planAccount) {
+        if (CollectionUtils.isEmpty(batchGroupPublishAllFansAccounts) || planAccount == null) {
+            return false;
+        }
+        for (String configuredAccount : batchGroupPublishAllFansAccounts) {
+            if (StringUtils.isBlank(configuredAccount)) {
+                continue;
+            }
+            String account = configuredAccount.trim();
+            if ("all".equalsIgnoreCase(account)
+                    || account.equals(planAccount.getAccountId())
+                    || account.equals(planAccount.getAccountName())
+                    || account.equals(planAccount.getGhId())) {
+                return true;
             }
-            group++;
         }
+        return false;
     }
 
     private List<PublishMiniprogramParam> getPublishCardList(List<PublishMiniprogram> publishMiniprogramList) {

long-article-server/src/main/resources/mapper/crawler/ArticleUseGroupMapper.xml

@@ -246,6 +246,15 @@
     limit #{count,jdbcType=INTEGER}
   </select>
 
+  <select id="selectAllOpenIdsByPage" resultMap="BaseResultMap" timeout="30">
+    select id, open_id
+    from article_user_group
+    where is_delete = 0 and gzh_id = #{ghId,jdbcType=VARCHAR}
+      and id &gt; #{lastId,jdbcType=BIGINT}
+    order by id asc
+    limit #{pageSize,jdbcType=INTEGER}
+  </select>
+
   <select id="selectOpenIdsByRemainingCount" resultType="java.lang.String">
     select open_id
     from article_user_group
@@ -271,4 +280,4 @@
       and is_delete = 0
       and remaining_count > 0
   </update>
-</mapper>
+</mapper>