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:
heavyrain2012
2021-02-12 21:20:44 +08:00
parent 5857a3fbd1
commit 90449e08c6
2 changed files with 37 additions and 70 deletions
Download Patch File Download Diff File Expand all files Collapse all files
broker/src/main/java/io/moquette/persistence/DatabaseStore.java
+27 -43
View File
@@ -1136,7 +1136,6 @@ public class DatabaseStore {
"`_type` = ?," +
"`_extra` = ?," +
"`_dt` = ?," +
"`_member_count` = ?," +
"`_mute` = ?" +
", `_join_type` = ?" +
", `_private_chat` = ?" +
@@ -1167,7 +1166,6 @@ public class DatabaseStore {
statement.setInt(index++, groupInfo.getType());
statement.setString(index++, groupInfo.getExtra());
statement.setLong(index++, groupInfo.getUpdateDt() == 0 ? System.currentTimeMillis() : groupInfo.getUpdateDt());
statement.setInt(index++, groupInfo.getMemberCount());
statement.setInt(index++, groupInfo.getMute());
statement.setInt(index++, groupInfo.getJoinType());
statement.setInt(index++, groupInfo.getPrivateChat());
@@ -1605,12 +1603,10 @@ public class DatabaseStore {
ResultSet rs = null;
try {
connection = DBUtil.getConnection();
String sql = "select `_gid` from t_group_member where `_mid` = ? and `_type` <> ?";
String sql = "select `_gid` from t_group_member where `_mid` = ? and `_type` <> 4";
statement = connection.prepareStatement(sql);
statement.setString(1, userId);
statement.setInt(2, ProtoConstants.GroupMemberType.GroupMemberType_Removed);
rs = statement.executeQuery();
Set<String> out = new HashSet<>();
@@ -1628,25 +1624,17 @@ public class DatabaseStore {
}
return null;
}
void updateGroupMemberCountDt(final String groupId, final int count, final long dt) {
void updateGroupMemberDt(final String groupId, final long dt) {
Connection connection = null;
PreparedStatement statement = null;
try {
connection = DBUtil.getConnection();
String sql ;
if (count >= 0) {
sql = "update t_group set `_member_count` = ?, `_member_dt` = ? , `_dt` = ? where `_gid` = ?";
} else {
sql = "update t_group set `_member_dt` = ?, `_dt` = ? where `_gid` = ?";
}
String sql = "update t_group set `_member_dt` = ?, `_dt` = ? where `_gid` = ?";
statement = connection.prepareStatement(sql);
int index = 1;
if (count >=0) {
statement.setInt(index++, count);
}
statement.setLong(index++, dt);
statement.setLong(index++, dt);
statement.setString(index++, groupId);
@@ -1662,6 +1650,29 @@ public class DatabaseStore {
}
}
void updateGroupMemberCountDt(final String groupId) {
Connection connection = null;
PreparedStatement statement = null;
try {
connection = DBUtil.getConnection();
String sql = "update t_group set `_member_count` = (select count(*) from t_group_member where `_gid` = ? and `_type` <> 4 limit 1), `_dt` = ?, `_member_dt` = `_member_dt` + 1 where `_gid` = ?";
statement = connection.prepareStatement(sql);
statement.setString(1, groupId);
statement.setLong(2, System.currentTimeMillis());
statement.setString(3, groupId);
int c = statement.executeUpdate();
LOG.info("Update rows {}", c);
} catch (SQLException e) {
// TODO Auto-generated catch block
e.printStackTrace();
Utility.printExecption(LOG, e, RDBS_Exception);
} finally {
DBUtil.closeDB(connection, statement);
}
}
void persistGroupMember(final String groupId, final List<WFCMessage.GroupMember> memberList) {
Connection connection = null;
PreparedStatement statement = null;
@@ -1724,11 +1735,8 @@ public class DatabaseStore {
int removeGroupMember(String groupId, List<String> groupMembers) {
Connection connection = null;
PreparedStatement statement = null;
PreparedStatement statement2 = null;
try {
connection = DBUtil.getConnection();
connection.setAutoCommit(false);
StringBuilder sqlBuilder = new StringBuilder("update t_group_member set `_type` = ?, `_dt` = ? where `_mid` in (");
for (int i = 0; i < groupMembers.size(); i++) {
sqlBuilder.append("?");
@@ -1737,9 +1745,7 @@ public class DatabaseStore {
}
}
sqlBuilder.append(")");
sqlBuilder.append(" and _gid = ?");
statement = connection.prepareStatement(sqlBuilder.toString());
int index = 1;
@@ -1751,36 +1757,14 @@ public class DatabaseStore {
statement.setString(index++, memberId);
}
statement.setString(index++, groupId);
int count = statement.executeUpdate();
LOG.info("Update rows {}", count);
String sql = "update t_group set `_member_count` = `_member_count` - ?, `_member_dt` = ? , `_dt` = ? where `_gid` = ?";
statement2 = connection.prepareStatement(sql);
index = 1;
statement2.setInt(index++, count);
statement2.setLong(index++, current);
statement2.setLong(index++, current);
statement2.setString(index++, groupId);
statement2.executeUpdate();
connection.commit();
connection.setAutoCommit(true);
} catch (SQLException e) {
// TODO Auto-generated catch block
if (connection != null) {
try {
connection.commit();
connection.setAutoCommit(true);
} catch (SQLException e1) {
e1.printStackTrace();
}
}
e.printStackTrace();
Utility.printExecption(LOG, e, RDBS_Exception);
} finally {
DBUtil.closeDB(connection, statement);
DBUtil.closeDB(connection, statement2);
}
return 0;
}
broker/src/main/java/io/moquette/persistence/MemoryMessagesStore.java
+10 -27
View File
@@ -1146,26 +1146,11 @@ public class MemoryMessagesStore implements IMessagesStore {
}
}
for (WFCMessage.GroupMember member : members) {
if (newInviteUsers.contains(member.getMemberId())) {
groupMembers.remove(groupId, member);
}
}
for (WFCMessage.GroupMember member : memberList) {
groupMembers.put(groupId, member);
}
int count = 0;
for (WFCMessage.GroupMember member : groupMembers.get(groupId)) {
if (member.getType() != GroupMemberType_Removed) {
count++;
}
}
mIMap.put(groupId, groupInfo.toBuilder().setMemberUpdateDt(updateDt).setUpdateDt(updateDt).setMemberCount(count).build());
databaseStore.persistGroupMember(groupId, memberList);
databaseStore.updateGroupMemberCountDt(groupId, count, updateDt);
databaseStore.updateGroupMemberCountDt(groupId);
groupMembers.remove(groupId);
mIMap.evict(groupId);
List<String> memberIds = new ArrayList<>();
for (WFCMessage.GroupMember member : memberList) {
@@ -1191,12 +1176,12 @@ public class MemoryMessagesStore implements IMessagesStore {
void removeGroupMember(String groupId, List<String> memberIds) {
HazelcastInstance hzInstance = m_Server.getHazelcastInstance();
databaseStore.removeGroupMember(groupId, memberIds);
databaseStore.updateGroupMemberCountDt(groupId);
MultiMap<String, WFCMessage.GroupMember> groupMembers = hzInstance.getMultiMap(GROUP_MEMBERS);
groupMembers.remove(groupId);
IMap<String, WFCMessage.GroupInfo> mIMap = hzInstance.getMap(GROUPS_MAP);
mIMap.evict(groupId);
}
void removeFavGroup(String groupId, List<String> memberIds) {
@@ -1384,8 +1369,6 @@ public class MemoryMessagesStore implements IMessagesStore {
}
}
WFCMessage.GroupInfo.Builder newInfoBuilder = oldInfo.toBuilder();
if (modifyType == Modify_Group_Name)
@@ -1407,7 +1390,8 @@ public class MemoryMessagesStore implements IMessagesStore {
newInfoBuilder.setUpdateDt(System.currentTimeMillis());
mIMap.put(groupId, newInfoBuilder.build());
databaseStore.persistGroupInfo(newInfoBuilder.build());
mIMap.evict(groupId);
if (modifyType == Modify_Group_Mute) {
if (newInfoBuilder.getMute() > 0) {
@@ -1488,7 +1472,7 @@ public class MemoryMessagesStore implements IMessagesStore {
groupMembers.remove(groupId, member);
member = member.toBuilder().setAlias(alias).setUpdateDt(updateDt).build();
databaseStore.persistGroupMember(groupId, Arrays.asList(member));
databaseStore.updateGroupMemberCountDt(groupId, -1, updateDt);
databaseStore.updateGroupMemberDt(groupId, updateDt);
groupMembers.put(groupId, member);
mIMap.set(groupId, groupInfo.toBuilder().setUpdateDt(updateDt).setMemberUpdateDt(updateDt).build());
@@ -1692,11 +1676,10 @@ public class MemoryMessagesStore implements IMessagesStore {
member = member.toBuilder().setType(type == 0 ? ProtoConstants.GroupMemberType.GroupMemberType_Normal : ProtoConstants.GroupMemberType.GroupMemberType_Manager).setUpdateDt(updateDt).build();
databaseStore.persistGroupMember(groupId, Arrays.asList(member));
groupMembers.put(groupId, member);
// userList.remove(member.getMemberId());
}
}
mIMap.put(groupId, groupInfo.toBuilder().setUpdateDt(updateDt).setMemberUpdateDt(updateDt).build());
databaseStore.persistGroupInfo(groupInfo.toBuilder().setUpdateDt(updateDt).setMemberUpdateDt(updateDt).build());
mIMap.evict(groupId);
callbackGroupMemberEvent(operator, groupId, userList, ProtoConstants.GroupMemberUpdateEventType.Group_Member_Event_Type_Update, (type == 0 ? ProtoConstants.GroupMemberType.GroupMemberType_Normal : ProtoConstants.GroupMemberType.GroupMemberType_Manager) + "");
return ErrorCode.ERROR_CODE_SUCCESS;