From 7923a3e0c42fe8c84bb2461d93a1805158d7f419 Mon Sep 17 00:00:00 2001 From: heavyrain2012 Date: Sat, 23 Jan 2021 18:01:28 +0800 Subject: [PATCH] =?UTF-8?q?db=20=E8=87=AA=E5=8A=A8=E6=B8=85=E9=99=A4?= =?UTF-8?q?=E8=BF=87=E6=9C=9F=E5=8E=86=E5=8F=B2=E6=B6=88=E6=81=AF=E6=95=B0?= =?UTF-8?q?=E6=8D=AE?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- broker/config/wildfirechat.conf | 5 + .../java/io/moquette/BrokerConstants.java | 1 + .../main/java/io/moquette/server/Server.java | 1 + .../src/main/java/win/liyufan/im/DBUtil.java | 473 ++++++++++++------ .../win/liyufan/im/MessageShardingUtil.java | 20 + .../src/main/resources/wildfirechat.conf | 5 + 6 files changed, 365 insertions(+), 140 deletions(-) diff --git a/broker/config/wildfirechat.conf b/broker/config/wildfirechat.conf index 42da038e..494bb21e 100755 --- a/broker/config/wildfirechat.conf +++ b/broker/config/wildfirechat.conf @@ -35,6 +35,11 @@ local_port 80 ##是否使用内置DB。0使用mysql；1使用h2db；2使用mysql+mongodb。社区版只支持0和1，专业版支持2 embed.db 1 +## 是否自动清理历史消息记录，当为true时，IM服务会定时扫描t_messages_x和t_user_messages_x表，清除掉3年前的历史消息。 +## 如果不开启自动清理，需要手动清理t_messages_x和t_user_messages_x表内的历史数据，保存数据不能超过3年。 +## 使用mongodb时次开关无效（因为mongodb设置了消息过期时间，会自动清理）。 +db.auto_clean_history_messages true + ##服务器管理接口密钥 http.admin.secret_key 123456 diff --git a/broker/src/main/java/io/moquette/BrokerConstants.java b/broker/src/main/java/io/moquette/BrokerConstants.java index 9149840e..3260167c 100755 --- a/broker/src/main/java/io/moquette/BrokerConstants.java +++ b/broker/src/main/java/io/moquette/BrokerConstants.java @@ -117,6 +117,7 @@ public final class BrokerConstants { public static final String TOKEN_EXPIRE_TIME = "token.expire_time"; public static final String EMBED_DB_PROPERTY_NAME = "embed.db"; + public static final String DB_AUTO_CLEAN_HISTORY_MESSAGES = "db.auto_clean_history_messages"; public static final String SENSITIVE_Filter_Type = "sensitive.filter.type"; diff --git a/broker/src/main/java/io/moquette/server/Server.java b/broker/src/main/java/io/moquette/server/Server.java index 0fca413d..d7f79a64 100755 --- a/broker/src/main/java/io/moquette/server/Server.java +++ b/broker/src/main/java/io/moquette/server/Server.java @@ -436,6 +436,7 @@ public class Server { System.out.println("Server will flush data to db before shutting down, please wait 5 seconds!"); LOG.info("Unbinding server from the configured ports"); m_shutdowning = true; + DBUtil.SystemExiting = true; m_acceptor.close(); LOG.trace("Stopping MQTT protocol processor"); diff --git a/broker/src/main/java/win/liyufan/im/DBUtil.java b/broker/src/main/java/win/liyufan/im/DBUtil.java index 09ad81fd..563cfae0 100755 --- a/broker/src/main/java/win/liyufan/im/DBUtil.java +++ b/broker/src/main/java/win/liyufan/im/DBUtil.java @@ -14,6 +14,8 @@ import java.io.BufferedReader; import java.io.FileReader; import java.sql.*; import java.util.ArrayList; +import java.util.Calendar; +import java.util.Date; import java.util.List; import com.hazelcast.util.StringUtil; @@ -27,6 +29,8 @@ import org.slf4j.LoggerFactory; import java.util.concurrent.ConcurrentHashMap; +import static cn.wildfirechat.common.IMExceptionEvent.EventType.RDBS_Exception; + public class DBUtil { private static final Logger LOG = LoggerFactory.getLogger(DBUtil.class); private static ComboPooledDataSource comboPooledDataSource = null; @@ -39,118 +43,307 @@ public class DBUtil { } }; - public static boolean IsEmbedDB = false; + public static boolean IsEmbedDB = false; + public static boolean SystemExiting = false; + public static boolean ClearDBDebugMode = false; - public static void init(IConfig config) { - String embedDB = config.getProperty(BrokerConstants.EMBED_DB_PROPERTY_NAME); - if (embedDB != null && embedDB.equals("1")) { - IsEmbedDB = true; - LOG.info("Use h2 database"); + public static void init(IConfig config) { + String embedDB = config.getProperty(BrokerConstants.EMBED_DB_PROPERTY_NAME); + boolean autoCleanMsgs = "true".equals(config.getProperty(BrokerConstants.DB_AUTO_CLEAN_HISTORY_MESSAGES)); + if (embedDB != null && embedDB.equals("1")) { + IsEmbedDB = true; + LOG.info("Use h2 database"); + } else { + IsEmbedDB = false; + LOG.info("Use mysql database"); + } + + if (comboPooledDataSource == null) { + String migrateLocation; + if (IsEmbedDB) { + migrateLocation = "filesystem:./migrate/h2"; + comboPooledDataSource = new ComboPooledDataSource(); + + comboPooledDataSource.setJdbcUrl( "jdbc:h2:./h2db/wfchat;AUTO_SERVER=TRUE;MODE=MySQL" ); + comboPooledDataSource.setUser("SA"); + comboPooledDataSource.setPassword("SA"); + comboPooledDataSource.setMinPoolSize(5); + comboPooledDataSource.setAcquireIncrement(5); + comboPooledDataSource.setMaxPoolSize(20); + + comboPooledDataSource.setIdleConnectionTestPeriod(60 * 5); + comboPooledDataSource.setMinPoolSize(3); + comboPooledDataSource.setInitialPoolSize(3); + + try { + comboPooledDataSource.setDriverClass( "org.h2.Driver" ); //loads the jdbc driver + } catch (PropertyVetoException e) { + e.printStackTrace(); + Utility.printExecption(LOG, e); + System.exit(-1); + } } else { - IsEmbedDB = false; - LOG.info("Use mysql database"); + migrateLocation = "filesystem:./migrate/mysql"; + comboPooledDataSource = new ComboPooledDataSource("mysql"); + try { + String url01 = comboPooledDataSource.getJdbcUrl().substring(0,comboPooledDataSource.getJdbcUrl().indexOf("?")); + String datasourceName = url01.substring(url01.lastIndexOf("/")+1); + // 连接已经存在的数据库，如：mysql + String jdbc = comboPooledDataSource.getJdbcUrl().replace(datasourceName, ""); + Connection connection = DriverManager.getConnection(jdbc, comboPooledDataSource.getUser(), comboPooledDataSource.getPassword()); + Statement statement = connection.createStatement(); + + // 创建数据库 + statement.executeUpdate("CREATE DATABASE IF NOT EXISTS `" + datasourceName + "` CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci;"); + + statement.close(); + connection.close(); + } catch (Exception e) { + e.printStackTrace(); + System.exit(-1); + } + + } + Flyway flyway = Flyway.configure().dataSource(comboPooledDataSource).locations(migrateLocation).baselineOnMigrate(true).load(); + flyway.migrate(); + + if(autoCleanMsgs) { + cleanHistoryMsg(); + } + } + } + + private static void cleanHistoryMsg() { + new Thread(()->{ + long usedTime = 0; + while (true) { + if(SystemExiting) { + break; + } + try { + long sleepTime; + if(ClearDBDebugMode) { + sleepTime = 60 * 1000 - usedTime; + } else { + sleepTime = 60 * 60 * 1000 - usedTime; + } + if(sleepTime < 0) { + sleepTime = 5 * 1000; + } + + Thread.sleep(sleepTime); + LOG.info("Start clean history messages"); + + if(SystemExiting) { + break; + } + long start = System.currentTimeMillis(); + if(IsEmbedDB) { + clearH2DB(); + } else { + clearMySQL(); + } + usedTime = System.currentTimeMillis() - start; + } catch (InterruptedException e) { + e.printStackTrace(); + } + } + }).start(); + } + private static boolean sleep() { + if(SystemExiting) + return true; + + try { + if(ClearDBDebugMode) { + Thread.sleep(500); + } else { + Thread.sleep(10000); + } + } catch (InterruptedException e) { + e.printStackTrace(); + } + return SystemExiting; + } + + private static void clearH2DB() { + boolean hasMore = true; + while (hasMore) { + hasMore = clearOneTable("t_user_messages"); + if(sleep()) { + return; + } + } + hasMore = true; + + while (hasMore) { + hasMore = clearOneTable("t_messages"); + if(sleep()) { + return; + } + } + } + private static void clearMySQL() { + int i = 0; + do { + if(!clearOneTable("t_user_messages_" + i)) { + i++; } - if (comboPooledDataSource == null) { - String migrateLocation; - if (IsEmbedDB) { - migrateLocation = "filesystem:./migrate/h2"; - comboPooledDataSource = new ComboPooledDataSource(); + if(sleep()) { + return; + } + } while (i < 128); - comboPooledDataSource.setJdbcUrl( "jdbc:h2:./h2db/wfchat;AUTO_SERVER=TRUE;MODE=MySQL" ); - comboPooledDataSource.setUser("SA"); - comboPooledDataSource.setPassword("SA"); - comboPooledDataSource.setMinPoolSize(5); - comboPooledDataSource.setAcquireIncrement(5); - comboPooledDataSource.setMaxPoolSize(20); + boolean hasMore = true; + String msgTableName; + if(ClearDBDebugMode) { + msgTableName = MessageShardingUtil.getMessageTable(MessageShardingUtil.getMsgIdFromTimestamp(System.currentTimeMillis())); + } else { + msgTableName = MessageShardingUtil.getMessageTable(MessageShardingUtil.getMsgIdFromTimestamp(System.currentTimeMillis()), 1); + } + while (hasMore) { + hasMore = clearOneTable(msgTableName); + if(sleep()) { + return; + } + } - comboPooledDataSource.setIdleConnectionTestPeriod(60 * 5); - comboPooledDataSource.setMinPoolSize(3); - comboPooledDataSource.setInitialPoolSize(3); + Calendar cal = Calendar.getInstance(); + cal.setTime(new Date()); + if(cal.get(Calendar.DAY_OF_MONTH) > 25 && !ClearDBDebugMode) { + hasMore = true; + msgTableName = MessageShardingUtil.getMessageTable(MessageShardingUtil.getMsgIdFromTimestamp(System.currentTimeMillis()), 2); + while (hasMore) { + hasMore = clearOneTable(msgTableName); + if(sleep()) { + return; + } + } + } + } - try { - comboPooledDataSource.setDriverClass( "org.h2.Driver" ); //loads the jdbc driver - } catch (PropertyVetoException e) { - e.printStackTrace(); - Utility.printExecption(LOG, e); - System.exit(-1); - } + private static boolean clearOneTable(String tableName) { + Connection connection = null; + PreparedStatement statement = null; + ResultSet rs = null; + int clearCount = 999; + if(ClearDBDebugMode) { + clearCount = 19; + } + boolean hasMore = true; + try { + connection = DBUtil.getConnection(); + String sql = "select id, _mid from " + tableName + " order by id limit 1 offset " + clearCount; + statement = connection.prepareStatement(sql); + + rs = statement.executeQuery(); + int id = 0; + long mid = 0; + if (rs.next()) { + id = rs.getInt(1); + mid = rs.getLong(2); + } + if(id == 0) { + hasMore = false; + rs.close(); + rs = null; + statement.close(); + + sql = "select id, _mid from " + tableName + " order by id desc limit 1"; + statement = connection.prepareStatement(sql); + + rs = statement.executeQuery(); + if (rs.next()) { + id = rs.getInt(1); + mid = rs.getLong(2); + } + } + + rs.close(); + rs = null; + LOG.info("Check history message in table {}, id {}, mid {}, hasMore {}", tableName, id, mid, hasMore); + if(id != 0) { + Calendar tm = MessageShardingUtil.getCalendarFromMessageId(mid); + if(ClearDBDebugMode) { + tm.add(Calendar.HOUR, 1); } else { - migrateLocation = "filesystem:./migrate/mysql"; - comboPooledDataSource = new ComboPooledDataSource("mysql"); - try { - String url01 = comboPooledDataSource.getJdbcUrl().substring(0,comboPooledDataSource.getJdbcUrl().indexOf("?")); - String datasourceName = url01.substring(url01.lastIndexOf("/")+1); - // 连接已经存在的数据库，如：mysql - String jdbc = comboPooledDataSource.getJdbcUrl().replace(datasourceName, ""); - Connection connection = DriverManager.getConnection(jdbc, comboPooledDataSource.getUser(), comboPooledDataSource.getPassword()); - Statement statement = connection.createStatement(); - - // 创建数据库 - statement.executeUpdate("CREATE DATABASE IF NOT EXISTS `" + datasourceName + "` CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci;"); - - statement.close(); - connection.close(); - } catch (Exception e) { - e.printStackTrace(); - System.exit(-1); - } - + tm.add(Calendar.YEAR, 3); + tm.add(Calendar.MONTH, -1); + tm.add(Calendar.DATE, -4); } - Flyway flyway = Flyway.configure().dataSource(comboPooledDataSource).locations(migrateLocation).baselineOnMigrate(true).load(); - flyway.migrate(); - } - } - - private static List getCreateSql() { - List out = new ArrayList<>(); - try{ - BufferedReader br = new BufferedReader(new FileReader("h2/create_table.sql"));//构造一个BufferedReader类来读取文件 - String s = null; - StringBuilder result = new StringBuilder(); - while((s = br.readLine())!=null) { - result.append(s); - if (s.contains(";")) { - out.add(result.toString()); - result = new StringBuilder(); - } + if(tm.getTimeInMillis() < System.currentTimeMillis()) { + statement.close(); + sql = "delete from " + tableName + " where id <= " + id; + statement = connection.prepareStatement(sql); + int count = statement.executeUpdate(); + LOG.info("Clear {} rows", count); + } else { + LOG.info("No history messages need clean"); + hasMore = false; } - br.close(); - }catch(Exception e){ - e.printStackTrace(); - Utility.printExecption(LOG, e); + } else { + LOG.info("No messages"); } - return out; + } catch (Exception e) { + hasMore = false; + e.printStackTrace(); + Utility.printExecption(LOG, e, RDBS_Exception); + } finally { + DBUtil.closeDB(connection, statement, rs); } - //从数据源中获取数据库的连接 - public static Connection getConnection() throws SQLException { - long threadId = Thread.currentThread().getId(); + return hasMore; + } - if (map.get(threadId) != null) { - LOG.error("error here!!!! DB connection not close correctly"); - } - map.put(threadId, Thread.currentThread().getStackTrace().toString()); - Connection connection = transactionConnection.get(); - if (connection != null) { - LOG.debug("Thread {} get db connection {}", threadId, connection); - return connection; + private static List getCreateSql() { + List out = new ArrayList<>(); + try{ + BufferedReader br = new BufferedReader(new FileReader("h2/create_table.sql"));//构造一个BufferedReader类来读取文件 + String s = null; + StringBuilder result = new StringBuilder(); + while((s = br.readLine())!=null) { + result.append(s); + if (s.contains(";")) { + out.add(result.toString()); + result = new StringBuilder(); + } } + br.close(); + }catch(Exception e){ + e.printStackTrace(); + Utility.printExecption(LOG, e); + } + return out; + } + //从数据源中获取数据库的连接 + public static Connection getConnection() throws SQLException { + long threadId = Thread.currentThread().getId(); - connection = comboPooledDataSource.getConnection(); + if (map.get(threadId) != null) { + LOG.error("error here!!!! DB connection not close correctly"); + } + map.put(threadId, Thread.currentThread().getStackTrace().toString()); + Connection connection = transactionConnection.get(); + if (connection != null) { LOG.debug("Thread {} get db connection {}", threadId, connection); return connection; } - public static void beginTransaction() { - try { - Connection connection = getConnection(); - connection.setAutoCommit(false); - transactionConnection.set(connection); - } catch (SQLException e) { - e.printStackTrace(); - Utility.printExecption(LOG, e); - } + connection = comboPooledDataSource.getConnection(); + LOG.debug("Thread {} get db connection {}", threadId, connection); + return connection; + } + + public static void beginTransaction() { + try { + Connection connection = getConnection(); + connection.setAutoCommit(false); + transactionConnection.set(connection); + } catch (SQLException e) { + e.printStackTrace(); + Utility.printExecption(LOG, e); } + } public static void commit() { try { @@ -180,57 +373,57 @@ public class DBUtil { } } - //释放资源，将数据库连接还给数据库连接池 - public static void closeDB(Connection conn,PreparedStatement ps,ResultSet rs) { - LOG.debug("Thread {} release db connection {}", Thread.currentThread().getId(), conn); - try { - if (rs!=null) { - rs.close(); - } - } catch (Exception e) { - e.printStackTrace(); - Utility.printExecption(LOG, e); - } - - try { - if (ps!=null) { - ps.close(); - } - } catch (Exception e) { - e.printStackTrace(); - Utility.printExecption(LOG, e); - } - - try { - if (conn!=null && transactionConnection.get() != conn) { - conn.close(); - map.remove(Thread.currentThread().getId()); - } - } catch (Exception e) { - e.printStackTrace(); - Utility.printExecption(LOG, e); + //释放资源，将数据库连接还给数据库连接池 + public static void closeDB(Connection conn,PreparedStatement ps,ResultSet rs) { + LOG.debug("Thread {} release db connection {}", Thread.currentThread().getId(), conn); + try { + if (rs!=null) { + rs.close(); } + } catch (Exception e) { + e.printStackTrace(); + Utility.printExecption(LOG, e); } - //释放资源，将数据库连接还给数据库连接池 - public static void closeDB(Connection conn, PreparedStatement ps) { - LOG.debug("Thread {} release db connection {}", Thread.currentThread().getId(), conn); - try { - if (ps!=null) { - ps.close(); - } - } catch (SQLException e) { - e.printStackTrace(); - Utility.printExecption(LOG, e); - } - try { - if (conn!=null && transactionConnection.get() != conn) { - conn.close(); - map.remove(Thread.currentThread().getId()); - } - } catch (SQLException e) { - e.printStackTrace(); - Utility.printExecption(LOG, e); + try { + if (ps!=null) { + ps.close(); } + } catch (Exception e) { + e.printStackTrace(); + Utility.printExecption(LOG, e); } + + try { + if (conn!=null && transactionConnection.get() != conn) { + conn.close(); + map.remove(Thread.currentThread().getId()); + } + } catch (Exception e) { + e.printStackTrace(); + Utility.printExecption(LOG, e); + } + } + //释放资源，将数据库连接还给数据库连接池 + public static void closeDB(Connection conn, PreparedStatement ps) { + LOG.debug("Thread {} release db connection {}", Thread.currentThread().getId(), conn); + try { + if (ps!=null) { + ps.close(); + } + } catch (SQLException e) { + e.printStackTrace(); + Utility.printExecption(LOG, e); + } + + try { + if (conn!=null && transactionConnection.get() != conn) { + conn.close(); + map.remove(Thread.currentThread().getId()); + } + } catch (SQLException e) { + e.printStackTrace(); + Utility.printExecption(LOG, e); + } + } } diff --git a/broker/src/main/java/win/liyufan/im/MessageShardingUtil.java b/broker/src/main/java/win/liyufan/im/MessageShardingUtil.java index 3ed89528..32b1c817 100644 --- a/broker/src/main/java/win/liyufan/im/MessageShardingUtil.java +++ b/broker/src/main/java/win/liyufan/im/MessageShardingUtil.java @@ -101,6 +101,26 @@ public class MessageShardingUtil { year %= 3; return "t_messages_" + (year * 12 + month); } + static Calendar getCalendarFromMessageId(long mid) { + Calendar calendar = Calendar.getInstance(); + if (mid != Long.MAX_VALUE && mid != 0) { + mid >>= (nodeIdWidth + rotateIdWidth); + Date date = new Date(mid + T201801010000); + calendar.setTime(date); + } else { + Date date = new Date(System.currentTimeMillis()); + calendar.setTime(date); + } + return calendar; + } + public static long getMsgIdFromTimestamp(long timestamp) { + long id = timestamp - T201801010000; + + id <<= rotateIdWidth; + id <<= nodeIdWidth; + + return id; + } public static String getPreviousMessageTable(long mid) { return getMessageTable(mid, -1); diff --git a/distribution/src/main/resources/wildfirechat.conf b/distribution/src/main/resources/wildfirechat.conf index b1d298f6..20493bb1 100755 --- a/distribution/src/main/resources/wildfirechat.conf +++ b/distribution/src/main/resources/wildfirechat.conf @@ -31,6 +31,11 @@ local_port 80 ##是否使用内置DB。0使用mysql；1使用h2db；2使用mysql+mongodb。社区版只支持0和1，专业版支持2 embed.db 1 +## 是否自动清理历史消息记录，当为true时，IM服务会定时扫描t_messages_x和t_user_messages_x表，清除掉3年前的历史消息。 +## 如果不开启自动清理，需要手动清理t_messages_x和t_user_messages_x表内的历史数据，保存数据不能超过3年。 +## 使用mongodb时次开关无效（因为mongodb设置了消息过期时间，会自动清理）。 +db.auto_clean_history_messages true + ##服务器管理接口密钥 http.admin.secret_key 123456