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
2023-04-13 09:18:43 +08:00
parent 62b36cd2b4
commit afee0466e7
1 changed files with 78 additions and 84 deletions
Download Patch File Download Diff File Expand all files Collapse all files
broker/src/main/java/io/moquette/persistence/DatabaseStore.java
+78 -84
View File
@@ -2411,59 +2411,55 @@ public class DatabaseStore {
} }
void deleteUserStatus(String userId) { void deleteUserStatus(String userId) {
mScheduler.execute(()->{ Connection connection = null;
Connection connection = null; PreparedStatement statement = null;
PreparedStatement statement = null; try {
try { connection = DBUtil.getConnection();
connection = DBUtil.getConnection(); String sql = "delete from t_user_status where _uid = ?";
statement = connection.prepareStatement(sql);
statement.setString(1, userId);
int count = statement.executeUpdate();
LOG.info("Update rows {}", count);
} catch (SQLException e) {
// TODO Auto-generated catch block
e.printStackTrace();
Utility.printExecption(LOG, e, RDBS_Exception);
} finally {
DBUtil.closeDB(connection, statement);
}
}
void updateUserStatus(String userId, int status) {
Connection connection = null;
PreparedStatement statement = null;
try {
connection = DBUtil.getConnection();
if (status == 0) {
String sql = "delete from t_user_status where _uid = ?"; String sql = "delete from t_user_status where _uid = ?";
statement = connection.prepareStatement(sql); statement = connection.prepareStatement(sql);
statement.setString(1, userId); statement.setString(1, userId);
int count = statement.executeUpdate(); int count = statement.executeUpdate();
LOG.info("Update rows {}", count); LOG.info("Update rows {}", count);
} catch (SQLException e) { } else {
// TODO Auto-generated catch block String sql = "insert into t_user_status (`_uid`, `_status`, `_dt`) values(?,?,?) ON DUPLICATE KEY UPDATE `_status` = ?";
e.printStackTrace();
Utility.printExecption(LOG, e, RDBS_Exception); statement = connection.prepareStatement(sql);
} finally { int index = 1;
DBUtil.closeDB(connection, statement); statement.setString(index++, userId);
statement.setInt(index++, status);
statement.setLong(index++, System.currentTimeMillis());
statement.setInt(index++, status);
int count = statement.executeUpdate();
LOG.info("Update rows {}", count);
} }
}); } catch (SQLException e) {
} // TODO Auto-generated catch block
e.printStackTrace();
void updateUserStatus(String userId, int status) { Utility.printExecption(LOG, e, RDBS_Exception);
mScheduler.execute(()->{ } finally {
Connection connection = null; DBUtil.closeDB(connection, statement);
PreparedStatement statement = null; }
try {
connection = DBUtil.getConnection();
if (status == 0) {
String sql = "delete from t_user_status where _uid = ?";
statement = connection.prepareStatement(sql);
statement.setString(1, userId);
int count = statement.executeUpdate();
LOG.info("Update rows {}", count);
} else {
String sql = "insert into t_user_status (`_uid`, `_status`, `_dt`) values(?,?,?) ON DUPLICATE KEY UPDATE `_status` = ?";
statement = connection.prepareStatement(sql);
int index = 1;
statement.setString(index++, userId);
statement.setInt(index++, status);
statement.setLong(index++, System.currentTimeMillis());
statement.setInt(index++, status);
int count = statement.executeUpdate();
LOG.info("Update rows {}", count);
}
} catch (SQLException e) {
// TODO Auto-generated catch block
e.printStackTrace();
Utility.printExecption(LOG, e, RDBS_Exception);
} finally {
DBUtil.closeDB(connection, statement);
}
});
} }
int getGeneratedId() { int getGeneratedId() {
@@ -3342,47 +3338,45 @@ public class DatabaseStore {
} }
void updateChannelListener(final String channelId, final String listener, final boolean listen) { void updateChannelListener(final String channelId, final String listener, final boolean listen) {
LOG.info("Database remove channel {}", channelId); LOG.info("updateChannelListener channel {}, listener {}, listen {}", channelId, listener, listen);
mScheduler.execute(()->{ Connection connection = null;
Connection connection = null; PreparedStatement statement = null;
PreparedStatement statement = null; try {
try { connection = DBUtil.getConnection();
connection = DBUtil.getConnection(); String sql;
String sql; if (listen) {
if (listen) { sql = "insert into t_channel_listener (`_cid`" +
sql = "insert into t_channel_listener (`_cid`" + ", `_mid`" +
", `_mid`" + ", `_dt`) values(?, ?, ?)" +
", `_dt`) values(?, ?, ?)" + " ON DUPLICATE KEY UPDATE " +
" ON DUPLICATE KEY UPDATE " + "`_dt` = ?";
"`_dt` = ?"; statement = connection.prepareStatement(sql);
statement = connection.prepareStatement(sql); int index = 1;
int index = 1; statement.setString(index++, channelId);
statement.setString(index++, channelId); statement.setString(index++, listener);
statement.setString(index++, listener); statement.setLong(index++, System.currentTimeMillis());
statement.setLong(index++, System.currentTimeMillis()); statement.setLong(index++, System.currentTimeMillis());
statement.setLong(index++, System.currentTimeMillis());
int count = statement.executeUpdate(); int count = statement.executeUpdate();
LOG.info("Update rows {}", count); LOG.info("Update rows {}", count);
} else { } else {
sql = "delete from t_channel_listener where `_cid`=? and `_mid`=?"; sql = "delete from t_channel_listener where `_cid`=? and `_mid`=?";
statement = connection.prepareStatement(sql); statement = connection.prepareStatement(sql);
int index = 1; int index = 1;
statement.setString(index++, channelId); statement.setString(index++, channelId);
statement.setString(index++, listener); statement.setString(index++, listener);
int count = statement.executeUpdate(); int count = statement.executeUpdate();
LOG.info("Update rows {}", count); LOG.info("Update rows {}", count);
}
} catch (SQLException e) {
// TODO Auto-generated catch block
e.printStackTrace();
Utility.printExecption(LOG, e, RDBS_Exception);
} finally {
DBUtil.closeDB(connection, statement);
} }
}); } catch (SQLException e) {
// TODO Auto-generated catch block
e.printStackTrace();
Utility.printExecption(LOG, e, RDBS_Exception);
} finally {
DBUtil.closeDB(connection, statement);
}
} }
List<String> getUserChannels(String userId) { List<String> getUserChannels(String userId) {