|
@@ -1,6 +1,5 @@
|
|
|
package com.fs.feishu.service;
|
|
package com.fs.feishu.service;
|
|
|
|
|
|
|
|
-
|
|
|
|
|
import com.fs.common.exception.CustomException;
|
|
import com.fs.common.exception.CustomException;
|
|
|
import com.fs.feishu.domain.FeishuAccount;
|
|
import com.fs.feishu.domain.FeishuAccount;
|
|
|
import com.fs.feishu.mapper.FeishuAccountMapper;
|
|
import com.fs.feishu.mapper.FeishuAccountMapper;
|
|
@@ -13,12 +12,15 @@ import org.springframework.stereotype.Component;
|
|
|
import javax.annotation.PostConstruct;
|
|
import javax.annotation.PostConstruct;
|
|
|
import java.util.ArrayList;
|
|
import java.util.ArrayList;
|
|
|
import java.util.List;
|
|
import java.util.List;
|
|
|
|
|
+import java.util.stream.Collectors;
|
|
|
|
|
|
|
|
/**
|
|
/**
|
|
|
* 飞书账号池管理:
|
|
* 飞书账号池管理:
|
|
|
* 1. 启动时从数据库加载所有启用公共账号,构建 Client 列表
|
|
* 1. 启动时从数据库加载所有启用公共账号,构建 Client 列表
|
|
|
* 2. 使用 Redis 循环队列实现跨实例的轮询负载均衡
|
|
* 2. 使用 Redis 循环队列实现跨实例的轮询负载均衡
|
|
|
- * 3. 提供禁用异常账号的能力,同步更新数据库、内存和 Redis 索引
|
|
|
|
|
|
|
+ * 3. 提供 Client + FeishuAccount 的包装对象,便于获取账号信息
|
|
|
|
|
+ * 4. 支持禁用异常账号,同步更新数据库、内存和 Redis 索引
|
|
|
|
|
+ * 5. 支持手动刷新账号池
|
|
|
*/
|
|
*/
|
|
|
@Component
|
|
@Component
|
|
|
@Slf4j
|
|
@Slf4j
|
|
@@ -27,114 +29,204 @@ public class FeishuClientPool {
|
|
|
@Autowired
|
|
@Autowired
|
|
|
private FeishuAccountMapper feishuAccountMapper;
|
|
private FeishuAccountMapper feishuAccountMapper;
|
|
|
|
|
|
|
|
-
|
|
|
|
|
@Autowired
|
|
@Autowired
|
|
|
private StringRedisTemplate stringRedisTemplate;
|
|
private StringRedisTemplate stringRedisTemplate;
|
|
|
|
|
|
|
|
- private volatile List<Client> clients;
|
|
|
|
|
- private volatile List<FeishuAccount> accountList;
|
|
|
|
|
|
|
+ /**
|
|
|
|
|
+ * 内存中的 Client 列表(volatile 保证可见性)
|
|
|
|
|
+ */
|
|
|
|
|
+ private volatile List<Client> clients = new ArrayList<>();
|
|
|
|
|
+
|
|
|
|
|
+ /**
|
|
|
|
|
+ * 内存中的 FeishuAccount 列表(与 clients 顺序一致,同索引对应)
|
|
|
|
|
+ */
|
|
|
|
|
+ private volatile List<FeishuAccount> accounts = new ArrayList<>();
|
|
|
|
|
+
|
|
|
|
|
+ /**
|
|
|
|
|
+ * Redis 中存储索引列表的 key
|
|
|
|
|
+ */
|
|
|
private static final String POOL_KEY = "feishu:account_pool";
|
|
private static final String POOL_KEY = "feishu:account_pool";
|
|
|
|
|
|
|
|
|
|
+ /**
|
|
|
|
|
+ * 初始化时加载账号池
|
|
|
|
|
+ */
|
|
|
@PostConstruct
|
|
@PostConstruct
|
|
|
public void init() {
|
|
public void init() {
|
|
|
refresh();
|
|
refresh();
|
|
|
}
|
|
}
|
|
|
|
|
|
|
|
|
|
+ /**
|
|
|
|
|
+ * 刷新账号池:从数据库加载所有启用状态的公共账号,重建内存列表和 Redis 索引
|
|
|
|
|
+ */
|
|
|
public synchronized void refresh() {
|
|
public synchronized void refresh() {
|
|
|
- //公共账号
|
|
|
|
|
- List<FeishuAccount> accounts = feishuAccountMapper.selectAllPublicAccounts();
|
|
|
|
|
- if (accounts.isEmpty()) {
|
|
|
|
|
- log.error("没有可用的公共飞书应用账号");
|
|
|
|
|
|
|
+ // 从数据库加载 status=1 的公共账号(假设公共账号有标志 isPublic=1)
|
|
|
|
|
+ List<FeishuAccount> accountList = feishuAccountMapper.selectAllPublicAccounts();
|
|
|
|
|
+ if (accountList.isEmpty()) {
|
|
|
|
|
+ log.error("没有可用的公共飞书应用账号,账号池为空");
|
|
|
|
|
+ // 清空内存和 Redis
|
|
|
|
|
+ this.clients = new ArrayList<>();
|
|
|
|
|
+ this.accounts = new ArrayList<>();
|
|
|
|
|
+ resetPoolIndexes(0);
|
|
|
|
|
+ return;
|
|
|
}
|
|
}
|
|
|
|
|
|
|
|
- List<Client> newClients = new ArrayList<>(accounts.size());
|
|
|
|
|
- for (FeishuAccount acc : accounts) {
|
|
|
|
|
- Client client = Client.newBuilder(acc.getAppId(), acc.getAppSecret())
|
|
|
|
|
- .logReqAtDebug(true)
|
|
|
|
|
- .build();
|
|
|
|
|
- newClients.add(client);
|
|
|
|
|
- }
|
|
|
|
|
|
|
+ // 构建 Client 列表
|
|
|
|
|
+ List<Client> newClients = accountList.stream()
|
|
|
|
|
+ .map(acc -> Client.newBuilder(acc.getAppId(), acc.getAppSecret())
|
|
|
|
|
+ .logReqAtDebug(true)
|
|
|
|
|
+ .build())
|
|
|
|
|
+ .collect(Collectors.toList());
|
|
|
|
|
|
|
|
- this.accountList = accounts;
|
|
|
|
|
|
|
+ // 更新内存
|
|
|
this.clients = newClients;
|
|
this.clients = newClients;
|
|
|
|
|
+ this.accounts = accountList;
|
|
|
|
|
+
|
|
|
|
|
+ // 重置 Redis 索引
|
|
|
resetPoolIndexes(newClients.size());
|
|
resetPoolIndexes(newClients.size());
|
|
|
|
|
+
|
|
|
log.info("飞书账号池刷新完成,当前可用账号数: {}", newClients.size());
|
|
log.info("飞书账号池刷新完成,当前可用账号数: {}", newClients.size());
|
|
|
}
|
|
}
|
|
|
|
|
|
|
|
|
|
+ /**
|
|
|
|
|
+ * 重置 Redis 中的索引队列
|
|
|
|
|
+ */
|
|
|
private void resetPoolIndexes(int size) {
|
|
private void resetPoolIndexes(int size) {
|
|
|
stringRedisTemplate.delete(POOL_KEY);
|
|
stringRedisTemplate.delete(POOL_KEY);
|
|
|
- if (size == 0) {
|
|
|
|
|
|
|
+ if (size <= 0) {
|
|
|
return;
|
|
return;
|
|
|
}
|
|
}
|
|
|
List<String> indexes = new ArrayList<>(size);
|
|
List<String> indexes = new ArrayList<>(size);
|
|
|
for (int i = 0; i < size; i++) {
|
|
for (int i = 0; i < size; i++) {
|
|
|
indexes.add(String.valueOf(i));
|
|
indexes.add(String.valueOf(i));
|
|
|
}
|
|
}
|
|
|
- // 使用 StringRedisTemplate 的 rightPushAll,保证纯字符串存储
|
|
|
|
|
stringRedisTemplate.opsForList().rightPushAll(POOL_KEY, indexes);
|
|
stringRedisTemplate.opsForList().rightPushAll(POOL_KEY, indexes);
|
|
|
log.info("Redis 账号索引池已重置,数量: {}", size);
|
|
log.info("Redis 账号索引池已重置,数量: {}", size);
|
|
|
}
|
|
}
|
|
|
|
|
|
|
|
- public Client getClient() {
|
|
|
|
|
|
|
+ /**
|
|
|
|
|
+ * 获取一个可用账号的包装对象(包含 Client 和 FeishuAccount)
|
|
|
|
|
+ * 使用 Redis 循环队列实现轮询,保证多实例负载均衡
|
|
|
|
|
+ */
|
|
|
|
|
+ public ClientWrapper getClientWrapper() {
|
|
|
while (true) {
|
|
while (true) {
|
|
|
- // 1. 从 Redis 循环队列取出索引(原子操作)
|
|
|
|
|
|
|
+ // 1. 从 Redis 循环队列取出索引(原子操作 rightPopAndLeftPush)
|
|
|
String indexStr = stringRedisTemplate.opsForList()
|
|
String indexStr = stringRedisTemplate.opsForList()
|
|
|
.rightPopAndLeftPush(POOL_KEY, POOL_KEY);
|
|
.rightPopAndLeftPush(POOL_KEY, POOL_KEY);
|
|
|
if (indexStr == null) {
|
|
if (indexStr == null) {
|
|
|
- log.warn("账号池为空,尝试重新初始化");
|
|
|
|
|
- resetPoolIndexes(clients.size());
|
|
|
|
|
|
|
+ // 队列为空,可能是账号池被清空或尚未初始化,尝试刷新
|
|
|
|
|
+ log.warn("账号池索引为空,尝试重新初始化");
|
|
|
|
|
+ refresh();
|
|
|
|
|
+ // 如果刷新后仍为空,抛出异常
|
|
|
|
|
+ if (clients.isEmpty()) {
|
|
|
|
|
+ throw new CustomException("当前没有可用的飞书公共账号");
|
|
|
|
|
+ }
|
|
|
|
|
+ continue;
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ int index;
|
|
|
|
|
+ try {
|
|
|
|
|
+ index = Integer.parseInt(indexStr);
|
|
|
|
|
+ } catch (NumberFormatException e) {
|
|
|
|
|
+ log.warn("无效的索引值: {}, 丢弃并重试", indexStr);
|
|
|
continue;
|
|
continue;
|
|
|
}
|
|
}
|
|
|
- int index = Integer.parseInt(indexStr);
|
|
|
|
|
|
|
|
|
|
- // 2. 获取当前 clients 快照(volatile 读)
|
|
|
|
|
|
|
+ // 2. 获取当前内存快照
|
|
|
List<Client> currentClients = this.clients;
|
|
List<Client> currentClients = this.clients;
|
|
|
|
|
+ List<FeishuAccount> currentAccounts = this.accounts;
|
|
|
|
|
|
|
|
- // 3. 如果索引越界(说明列表已被其他线程缩小),丢弃该索引并重试
|
|
|
|
|
|
|
+ // 3. 检查索引有效性
|
|
|
if (index >= currentClients.size()) {
|
|
if (index >= currentClients.size()) {
|
|
|
- log.warn("索引 {} 已失效(当前账号数 {}),重试", index, currentClients.size());
|
|
|
|
|
|
|
+ log.warn("索引 {} 越界(当前账号数 {}),丢弃并重试", index, currentClients.size());
|
|
|
continue;
|
|
continue;
|
|
|
}
|
|
}
|
|
|
- return currentClients.get(index);
|
|
|
|
|
|
|
+
|
|
|
|
|
+ // 4. 返回包装对象
|
|
|
|
|
+ return new ClientWrapper(currentClients.get(index), currentAccounts.get(index));
|
|
|
}
|
|
}
|
|
|
}
|
|
}
|
|
|
|
|
|
|
|
/**
|
|
/**
|
|
|
- * 禁用指定的 Client,并移除出账号池
|
|
|
|
|
|
|
+ * 禁用指定账号(根据 FeishuAccount 对象),将其从池中移除
|
|
|
|
|
+ * 同步方法保证内存和 Redis 的一致性
|
|
|
*
|
|
*
|
|
|
- * @param failedClient 需要禁用的 Client
|
|
|
|
|
- * @param errorMsg 错误原因,会写入数据库 error_msg 字段
|
|
|
|
|
|
|
+ * @param account 要禁用的账号(必须包含 ID)
|
|
|
|
|
+ * @param errorMsg 禁用原因,会写入数据库
|
|
|
*/
|
|
*/
|
|
|
- public synchronized void disableClient(Client failedClient, String errorMsg) {
|
|
|
|
|
- for (int i = 0; i < clients.size(); i++) {
|
|
|
|
|
- if (clients.get(i) == failedClient) {
|
|
|
|
|
- FeishuAccount account = accountList.get(i);
|
|
|
|
|
- // 更新数据库状态为禁用(status=0),并记录错误信息
|
|
|
|
|
- feishuAccountMapper.updateStatusAndErrorMsg(account.getId(), 0, errorMsg);
|
|
|
|
|
- // 从内存中移除
|
|
|
|
|
- clients.remove(i);
|
|
|
|
|
- accountList.remove(i);
|
|
|
|
|
- // 重建索引池,保证索引仍然连续
|
|
|
|
|
- resetPoolIndexes(clients.size());
|
|
|
|
|
- log.warn("飞书账号[{}]已被自动禁用,原因: {}", account.getAppId(), errorMsg);
|
|
|
|
|
- throw new CustomException(String.format("飞书账号[%s]不可用: %s", account.getAppId(), errorMsg));
|
|
|
|
|
|
|
+ public synchronized void disableAccount(FeishuAccount account, String errorMsg) {
|
|
|
|
|
+ if (account == null || account.getId() == null) {
|
|
|
|
|
+ log.warn("禁用账号失败:账号或ID为空");
|
|
|
|
|
+ return;
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ Long accountId = account.getId();
|
|
|
|
|
+ // 在 accounts 列表中查找匹配的索引(使用 ID 比较,避免对象引用问题)
|
|
|
|
|
+ int indexToRemove = -1;
|
|
|
|
|
+ for (int i = 0; i < accounts.size(); i++) {
|
|
|
|
|
+ if (accounts.get(i).getId().equals(accountId)) {
|
|
|
|
|
+ indexToRemove = i;
|
|
|
|
|
+ break;
|
|
|
}
|
|
}
|
|
|
}
|
|
}
|
|
|
- // 如果未找到,理论上不会执行到这里
|
|
|
|
|
- throw new CustomException("无法找到对应的飞书账号");
|
|
|
|
|
|
|
+
|
|
|
|
|
+ if (indexToRemove == -1) {
|
|
|
|
|
+ log.warn("账号 ID {} 不在当前池中,可能已被移除", accountId);
|
|
|
|
|
+ return;
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ // 更新数据库状态为禁用(status=0),并记录错误信息
|
|
|
|
|
+ feishuAccountMapper.updateStatusAndErrorMsg(accountId, 0, errorMsg);
|
|
|
|
|
+
|
|
|
|
|
+ // 从内存列表中移除(注意:不能直接修改原列表,要创建新列表避免并发问题)
|
|
|
|
|
+ List<Client> newClients = new ArrayList<>(clients);
|
|
|
|
|
+ List<FeishuAccount> newAccounts = new ArrayList<>(accounts);
|
|
|
|
|
+ newClients.remove(indexToRemove);
|
|
|
|
|
+ newAccounts.remove(indexToRemove);
|
|
|
|
|
+
|
|
|
|
|
+ // 更新 volatile 引用
|
|
|
|
|
+ this.clients = newClients;
|
|
|
|
|
+ this.accounts = newAccounts;
|
|
|
|
|
+
|
|
|
|
|
+ // 重置 Redis 索引
|
|
|
|
|
+ resetPoolIndexes(newClients.size());
|
|
|
|
|
+
|
|
|
|
|
+ log.warn("飞书账号[{}]已被自动禁用,原因: {}", account.getAppId(), errorMsg);
|
|
|
}
|
|
}
|
|
|
|
|
|
|
|
/**
|
|
/**
|
|
|
- * 获取当前可用账号数量的快照(用于重试上限)
|
|
|
|
|
|
|
+ * 获取当前可用账号数量的快照
|
|
|
*/
|
|
*/
|
|
|
public int getAvailableCount() {
|
|
public int getAvailableCount() {
|
|
|
return clients.size();
|
|
return clients.size();
|
|
|
}
|
|
}
|
|
|
|
|
|
|
|
/**
|
|
/**
|
|
|
- * 获取所有可用 Client 的快照(用于定时任务等遍历场景)
|
|
|
|
|
|
|
+ * 获取所有可用 Client 的快照(只读,用于遍历)
|
|
|
*/
|
|
*/
|
|
|
public List<Client> getAllClients() {
|
|
public List<Client> getAllClients() {
|
|
|
return new ArrayList<>(clients);
|
|
return new ArrayList<>(clients);
|
|
|
}
|
|
}
|
|
|
|
|
+
|
|
|
|
|
+ // ==================== 内部包装类 ====================
|
|
|
|
|
+
|
|
|
|
|
+ /**
|
|
|
|
|
+ * 客户端包装类,同时持有 Client 和对应的 FeishuAccount
|
|
|
|
|
+ */
|
|
|
|
|
+ public static class ClientWrapper {
|
|
|
|
|
+ private final Client client;
|
|
|
|
|
+ private final FeishuAccount account;
|
|
|
|
|
+
|
|
|
|
|
+ public ClientWrapper(Client client, FeishuAccount account) {
|
|
|
|
|
+ this.client = client;
|
|
|
|
|
+ this.account = account;
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ public Client getClient() {
|
|
|
|
|
+ return client;
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ public FeishuAccount getAccount() {
|
|
|
|
|
+ return account;
|
|
|
|
|
+ }
|
|
|
|
|
+ }
|
|
|
}
|
|
}
|