anna/IM-server-new
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

优化批量添加频道订阅功能

Browse Source
This commit is contained in:
heavyrian2012
2026-05-10 14:55:55 +08:00
parent 42281c3ad3
commit 44b18a66c0
2 changed files with 46 additions and 25 deletions
Download Patch File Download Diff File Expand all files Collapse all files
broker/src/main/java/io/moquette/persistence/DatabaseStore.java
+35 -16
View File
@@ -4220,50 +4220,69 @@ public class DatabaseStore {
}
public List<String> findMissingMids(String cid, List<String> midList, boolean contain) {
String tempTable = "temp_check_mids_" + System.currentTimeMillis();
if (midList == null || midList.isEmpty()) {
return new ArrayList<>();
}
try (Connection conn = DBUtil.getConnection()) {
// Step 1: 创建临时表（仅当前连接可见，自动清理）
String createSql = "CREATE TEMPORARY TABLE " + tempTable + " (_mid VARCHAR(64) NOT NULL PRIMARY KEY)";
String tempTable = "temp_check_mids_" + System.currentTimeMillis();
Connection conn = null;
try {
conn = DBUtil.getConnection();
// Step 1: 创建临时表
try (Statement st = conn.createStatement()) {
st.execute(createSql);
st.execute("CREATE TEMPORARY TABLE " + tempTable +
" (_mid VARCHAR(64) NOT NULL PRIMARY KEY)");
}
// Step 2: 批量插入待检查的 mid（去重用 INSERT IGNORE）
// Step 2: 批量插入
String insertSql = "INSERT IGNORE INTO " + tempTable + " (_mid) VALUES (?)";
int count = 0;
try (PreparedStatement ps = conn.prepareStatement(insertSql)) {
int count = 0;
for (String mid : midList) {
ps.setString(1, mid);
ps.addBatch();
if (++count % 1000 == 0) {
ps.executeBatch(); // 每1000条刷盘
if (++count % 1000 == 0) {
ps.executeBatch();
ps.clearBatch();
}
}
ps.executeBatch();
if (count % 1000 != 0) {
ps.executeBatch();
}
}
// Step 3: 查询差集（NOT EXISTS 利用复合索引）
// Step 3: 查询
String querySql = "SELECT t._mid FROM " + tempTable + " t " +
"WHERE " + (contain?"":" NOT ") + " EXISTS (" +
"WHERE " + (contain ? "" : " NOT ") + " EXISTS (" +
" SELECT 1 FROM t_channel_listener c " +
" WHERE c._cid = ? AND c._mid = t._mid" +
")";
List<String> missing = new ArrayList<>();
List<String> result = new ArrayList<>();
try (PreparedStatement ps = conn.prepareStatement(querySql)) {
ps.setString(1, cid);
try (ResultSet rs = ps.executeQuery()) {
while (rs.next()) {
missing.add(rs.getString(1));
result.add(rs.getString(1));
}
}
}
return missing;
return result;
} catch (SQLException e) {
throw new RuntimeException(e);
throw new RuntimeException("查询缺失mid失败", e);
} finally {
// 显式清理临时表
if (conn != null) {
try (Statement st = conn.createStatement()) {
st.execute("DROP TEMPORARY TABLE IF EXISTS " + tempTable);
} catch (SQLException ignored) {
// 清理失败不影响主流程
} finally {
try { conn.close(); } catch (SQLException ignored) {}
}
}
}
}
sdk/src/main/java/cn/wildfirechat/sdk/Main.java
+11 -9
View File
@@ -1694,7 +1694,7 @@ public class Main {
//测试用户设置功能
IMResult<Void> setUserSettingResult = GeneralAdmin.setUserSetting("user1", 1, "test_key", "test_value");
IMResult<Void> setUserSettingResult = GeneralAdmin.setUserSetting("user1", 3003, "test_key", "test_value");
if (setUserSettingResult != null && setUserSettingResult.getErrorCode() == ErrorCode.ERROR_CODE_SUCCESS) {
System.out.println("set user setting success");
} else {
@@ -1702,7 +1702,7 @@ public class Main {
System.exit(-1);
}
IMResult<UserSettingPojo> getUserSettingResult = GeneralAdmin.getUserSetting("user1", 1, "test_key");
IMResult<UserSettingPojo> getUserSettingResult = GeneralAdmin.getUserSetting("user1", 3003, "test_key");
if (getUserSettingResult != null && getUserSettingResult.getErrorCode() == ErrorCode.ERROR_CODE_SUCCESS && "test_value".equals(getUserSettingResult.getResult().getValue())) {
System.out.println("get user setting success");
} else {
@@ -2726,13 +2726,15 @@ public class Main {
System.exit(-1);
}
//测试重新发布消息
IMResult<Void> republishResult = channelServiceApi.republishMessage(resultSendMessage.getResult().getMessageUid(), Arrays.asList("userId2", "userId3"));
if (republishResult != null && republishResult.getErrorCode() == ErrorCode.ERROR_CODE_SUCCESS) {
System.out.println("republish message success");
} else {
System.out.println("republish message failure");
System.exit(-1);
if(commercialServer) {
//测试重新发布消息
IMResult<Void> republishResult = channelServiceApi.republishMessage(resultSendMessage.getResult().getMessageUid(), Arrays.asList("userId2", "userId3"));
if (republishResult != null && republishResult.getErrorCode() == ErrorCode.ERROR_CODE_SUCCESS) {
System.out.println("republish message success");
} else {
System.out.println("republish message failure");
System.exit(-1);
}
}
ArticleContent articleContent = new ArticleContent("article1", "https://media.wfcoss.cn/channel-assets/20220816/2dd76540daa9444dae44e942aa1c2bbc.png", "这是一个测试文章", "测试一下文章的功能", "https://mp.weixin.qq.com/s/W6tanLbALd3qqZM8r3MTgA", true);