Hibernate 多服務器數據同步方案：攔截器 + 本地隊列 + 定時重試

項目從單機遷移至集群，要求多數據庫最終一致性 (允許短暫不一致)，網絡不穩定 (鐵路內網)。方案：正常走分佈式事務/同步調用；斷網時 Hibernate Interceptor 攔截持久化操作，序列化到本地表 z_jcyy_tb；網絡恢復由後台線程讀取本地隊列重放同步。

架構圖解

[應用服務] --正常網絡--> [主庫] + [從庫1] + [從庫2]  (同步寫入)
       |
       |--斷網--> [Hibernate Interceptor] --> [本地隊列表 z_jcyy_tb]
                            (id, inputdate, object[Blob], action)
                            |
                            |--定時線程 (網絡恢復)--> 反序列化 --> 重放同步到遠程庫

1. 遠程 SessionFactory 建立 (動態解析 hibernate.cfg.xml)

private static String url_b, user_b, pass_b, driver_b;
static {
    try {
        // 反射讀取 SessionManager 私有常量 CONFIG_FILE_LOCATION
        Field field = SessionManager.class.getDeclaredField("CONFIG_FILE_LOCATION");
        field.setAccessible(true);
        String path = (String) field.get(SessionManager.class);
        // dom4j 解析 hibernate.cfg.xml
        Document doc = new SAXReader().read(TBDao.class.getResourceAsStream(path));
        // XPath + 正則提取 *_b 屬性
        String xpath = "/hibernate-configuration/session-factory/property[@name='XPATH_I']";
        Pattern p = Pattern.compile("(XPATH_I)");
        Matcher m = p.matcher(xpath);
        url_b = DocumentHelper.createXPath(m.replaceAll("connection.url_b")).selectSingleNode(doc).getText();
        user_b = DocumentHelper.createXPath(m.replaceAll("connection.username_b")).selectSingleNode(doc).getText();
        pass_b = DocumentHelper.createXPath(m.replaceAll("connection.password_b")).selectSingleNode(doc).getText();
        driver_b = DocumentHelper.createXPath(m.replaceAll("connection.driver_class")).selectSingleNode(doc).getText();
    } catch (Exception e) { e.printStackTrace(); }
}

public Session openSessionb() {
    try {
        Class.forName(driver_b);
        Connection conn = DriverManager.getConnection(url_b, user_b, pass_b);
        // 反射獲取已構建的 SessionFactory
        Field sfField = Arrays.stream(SessionManager.class.getDeclaredFields())
            .filter(f -> SessionFactory.class.equals(f.getType())).findFirst().orElseThrow();
        sfField.setAccessible(true);
        SessionFactory sf = (SessionFactory) sfField.get(SessionManager.class);
        return sf.openSession(conn); // 關鍵：使用外部 Connection 打開 Session
    } catch (Exception e) { return null; }
}

2. 斷網時攔截持久化操作 → 本地隊列

public void save(Object obj, String action) { // action: INSERT/UPDATE/DELETE
    Session session = null;
    try {
        session = SessionManager.currentSession(null, null);
        Transaction tr = session.beginTransaction();
        ZJcyyTb zj = new ZJcyyTb();
        zj.setAction(action);
        zj.setInputdate(new Date());
        session.save(zj);
        session.flush();
        session.refresh(zj, LockMode.UPGRADE); // 獲取自增 ID
        // 序列化物件到 Blob
        zj.setObject(new ObjectConvert().ObjectToBlob(obj));
        tr.commit();
    } catch (Exception e) { e.printStackTrace(); }
    finally { if (session != null && session.isOpen()) session.close(); }
}

ZJcyyTb 表結構：id (PK), inputdate, object (BLOB), action (VARCHAR)。

3. 定時線程重放同步 (TBDao.action())

public boolean action() {
    ObjectConvert oc = new ObjectConvert();
    Session local = null, remote = null;
    try {
        if (isSql() > 0) { // 本地有待同步記錄
            local = SessionManager.currentSession(null, null);
            remote = openSessionb(); // 嘗試連接遠程
            if (remote != null) {
                Transaction trL = local.beginTransaction();
                Transaction trR = remote.beginTransaction();
                Query q = local.createQuery("FROM ZJcyyTb ORDER BY inputdate");
                // 分頁處理，避免一次性載入過多
                for (int page = 0; ; page++) {
                    q.setFirstResult(page * 10).setMaxResults(10);
                    List list = q.list();
                    if (list.isEmpty()) break;
                    for (Object o : list) {
                        ZJcyyTb tb = (ZJcyyTb) o;
                        Object entity = oc.BlobToObject(tb.getObject(), null);
                        if (entity != null) {
                            switch (tb.getAction()) {
                                case DELETE: remote.delete(entity); break;
                                case INSERT: remote.save(entity); break;
                                case UPDATE: remote.update(entity); break;
                            }
                        }
                        local.delete(tb); // 同步成功刪除本地記錄
                    }
                    trL.commit(); trR.commit();
                }
            }
        }
        return true;
    } catch (Exception e) { return false; }
    finally { /* 關閉 Session */ }
}

private int isSql() { /* 查詢 z_jcyy_tb count(*) */ }

4. Hibernate Interceptor 接入點

在 hibernate.cfg.xml 註冊自定義 Interceptor：

com.example.sync.SyncInterceptor

SyncInterceptor 實現 onSave/onUpdate/onDelete，判斷遠程連接可用性：
- 可用 → 直接在遠程 Session 同步操作
- 不可用 → 調用 TBDao.save(entity, action) 入本地隊列

關鍵技術點總結

遺留優化方向

• 引入 JMS/消息隊列 替代表輪詢，解耦生產消費。
• 採用 Hibernate 二級緩存 (Ehcache/Infinispan) 減少重複查詢。
• 衝突檢測：同步前比較 version (樂觀鎖) 或 timestamp，發現衝突記錄人工介入。
• 監控大盤：本地隊列積壓量、同步延遲、失敗率，接入 Prometheus/Grafana。