posts - 165, comments - 198, trackbacks - 0, articles - 1
  BlogJava :: 首页 :: 新随笔 :: 联系 :: 聚合  :: 管理
本文是本人实际开发中遇到,特留文记录。在次我花了3天的时间解决,这个问题还是点难度的。

所用到知识点:
一般jdk
|-私有属性反射
|-序列化
|-正则表达使用
|-多线程使用
|-dom4j的xml读取
|+hibernate
   |-拦截器
   |-一个Session工厂同时连接不同数据库(本文关键
   |-oracle Blob 存取

等........

需求功能介绍:
 为性能考虑,单一服务器改成集群(每太服务器数据允许在一定时间内保持相步),给出的修改时间短,不过代码持久层比较统一(hibernate 感谢天还好是她! )。网络连接不稳定(铁路内网!)。

完成后效果:
 
当网络连接成功时,多数据库的同步。
   当
网络连接失败时,本地应用程序运用hibernate拦截器拦截正操作对象并记录下操作动作,序列化到本地时局库 z_jcyy_tb 表中。表数据属性为:id,inputdate(记录时间),object(序列对象),action(操作动作)。并安一定时间测试连接。如果成功,读取 z_jcyy_tb 表中数据 反序列化 再同步到 其他数据库中。

代码说明:
1.新Session 建立
  
hibernate.cfg.xml 在文件<session-factory>中添加
    <property name="connection.url_b">jdbc:oracle:thin:@192.168.1.114:1521:JCYY</property>

    
<property name="connection.username_b">jcyy</property>
    
<property name="connection.password_b">jcyy</property>
   TBDao -> OpenSession()
    private static String url_b = null ;
    
private static String use_b = null ;
    
private static String pass_b = null ;
    
private static String dirver_b = null ;
    
static {try {
        //取得hibernate.cfg.xml逻辑路径,和原来程序关联上 
        Field field   
=   SessionManager.class.getDeclaredField("CONFIG_FILE_LOCATION");
        field.setAccessible( 
true );
        String path  
=  (String) field.get(SessionManager. class );

        //通过 dom4j 加载 配置文件   

        Document docT 
= new SAXReader().read( TBDao.class.getResourceAsStream(path) );

        //正则+xpath读取 在hbn文件中加入的
<property name="..._b"> 的属性
        String xpath 
= "/hibernate-configuration/session-factory/property[@name='XPATH_I']" ;
        Pattern p 
= Pattern.compile("(XPATH_I)");
        Matcher ma 
= p.matcher(xpath);
        url_b  
= DocumentHelper.createXPath( ma.replaceAll("connection.url_b") ).selectSingleNode(docT).getText();
        use_b 
= DocumentHelper.createXPath( ma.replaceAll("connection.username_b")).selectSingleNode(docT).getText();
        pass_b 
= DocumentHelper.createXPath( ma.replaceAll("connection.password_b")).selectSingleNode(docT).getText();
        dirver_b 
= DocumentHelper.createXPath( ma.replaceAll("connection.driver_class")).selectSingleNode(docT).getText();
    } 
catch (Exception e) {e.printStackTrace();}}
   
    //利用hbn的SessionFactory得到 openSession(Connection); 打开异地数据库连接。
    //利用私有反射得到 加载完成的SessionFactory
    public Session openSessionb(){
        
try {
            Class.forName(dirver_b);
            Connection conn 
=  DriverManager.getConnection(url_b,use_b,pass_b);
            
            Field[] fields   
=   SessionManager.class.getDeclaredFields();
            Field field 
= null ;
            
for(int i=0;i<fields.length;i++){
                
if( SessionFactory.class.equals(  fields[i].getType() )  )
                    field 
= fields[i];
            }
            field.setAccessible(
true);
            SessionFactory sessionFactory  
=  (SessionFactory) field.get(SessionManager.class );
            
return sessionFactory.openSession(conn);
        } 
catch (Exception e) {
            System.out.println(
"--没有连接到总服务(openSessionb)--");
            
return null ;
        }
    }
 
2.异地数据同步失败后动作  TBDao->save() 冻结状态到数据库
    public void save(Object obj,String action) {
        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);
            //oracle Blob数据持久 请参考-->
序列化和反序列化对象到 数据库
            zj.setObject( new ObjectConvert().ObjectToBlob(obj) );
            tr.commit();
        } 
catch (Exception e) {
            e.printStackTrace();
        }
finally{
            
if(session!=null&& session.isOpen() )session.close();
        }

    }

3,失败后又成功连接后 (线程实现) TBDao->action()
    public int isSql(){
        
int is_count = 0 ;
        Session session 
= null ;
        
try {
            //得到本地Session 查看是否有连接失败后序列动作被保存
            session 
= SessionManager.currentSession(null,null);
            Transaction tr 
=  session.beginTransaction();
            Connection conn 
= session.connection();
            Statement stat 
= conn.createStatement();
            ResultSet rs 
= stat.executeQuery("select count(*) from z_jcyy_tb");
            rs.next();
            is_count 
= rs.getInt(1);
            tr.commit();
        } 
catch (Exception e) {
            e.printStackTrace();
        }
finally{
            
if(session!=null&& session.isOpen() )session.close();
        }
        
return is_count ;
    }

    
public boolean action(){
        
int isSql = 0 ;
        ObjectConvert oc 
= new ObjectConvert();
        Session session 
= null ;
        Session session_b 
= null ;
        
        
try {
            //有失败连接动作后尝试 远程数据库 
            
if( (isSql=isSql())>0 ){
                session 
= SessionManager.currentSession(null,null);
                //远程数据库连接 
                //如果成功连接:z_jcyy_tb表中数据同步到其他数据库中

                session_b 
= openSessionb();
                
if(session_b!=null){
                    Transaction tr_b 
= session_b.beginTransaction();
                    Transaction tr 
= session.beginTransaction();
                    
                    Query qu 
= session.createQuery(" from ZJcyyTb t order by t.inputdate");
                    
for(int i=0;i<=isSql/10;i++){
                        qu.setFirstResult(i
*10);
                        qu.setMaxResults(
10);
                        List list 
= qu.list();                
                        
for(Iterator it=list.iterator();it.hasNext();){
                            ZJcyyTb tb 
= (ZJcyyTb)it.next();
                            Object obj 
= null ;
                            obj 
= oc.BlobToObject(tb.getObject(),obj);
                            
if(obj!=null){
                                String action 
= tb.getAction();
                                
if(action.equals( TBDao.DELETE )){
                                    session_b.delete(obj);
                                }
                                
if(action.equals( TBDao.INSERT )){
                                    session_b.save(obj);
                                }
                                
if(action.equals( TBDao.UPDATE )){
                                    session_b.update(obj);
                                }
                            }
                            session.delete(tb);
                            tr.commit();
                        }
                    }
                    tr_b.commit();
                }
            }    
return true ;
        } 
catch (Exception e) {
            System.out.println(
"--没有连接到总服务(action)--");
        }
finally{
            
if(session_b!=null&&session_b.isOpen())session_b.close();
            
if(session!=null&& session.isOpen() )session.close();
            SessionManager.closeSession();    
        }
        
return false ;
    }

4.hbn 拦截器 ->Interceptor
package com.jjm.hibernate;

import java.io.File;
import java.io.FileInputStream;
import java.io.InputStream;
import java.io.Serializable;
import java.lang.reflect.Field;
import java.sql.Connection;
import java.sql.DriverManager;
import java.util.HashSet;
import java.util.Iterator;
import java.util.Set;

import org.dom4j.Document;
import org.dom4j.DocumentHelper;
import org.dom4j.XPath;
import org.dom4j.io.SAXReader;
import org.hibernate.CallbackException;
import org.hibernate.EntityMode;
import org.hibernate.Hibernate;
import org.hibernate.HibernateException;
import org.hibernate.Interceptor;
import org.hibernate.Session;
import org.hibernate.SessionFactory;
import org.hibernate.Transaction;
import org.hibernate.cfg.Configuration;
import org.hibernate.type.Type;

import com.jjm.rlzy.dao.TBDao;

public class TestInterceptor implements Interceptor,Serializable{
    
    
static private boolean isConn_b = false ;
    
static private TBDao tb = new TBDao();
    //线程 一分钟 检测连接失败 同步
    
static{
        
new Thread(new Runnable(){
            
public void run() {
                
while(true){
                    isConn_b 
= tb.action();
                    
try {
                        Thread.sleep(
60*1000);
                    } 
catch (InterruptedException e) {e.printStackTrace();}
                }
            }
        }).start();
    }

    
public boolean onFlushDirty(Object entity, Serializable id, Object[] currentState, Object[] previousState, String[] propertyNames, Type[] types) throws CallbackException {
        Session session 
= null ;
        
try {
            
if(isConn_b){
                session 
=  tb.openSessionb();
                Transaction tr 
= session.beginTransaction();
                session.update(entity);
                tr.commit();
            }
else{
                tb.save(entity,TBDao.UPDATE);
            }
        } 
catch (Exception e) {
            e.printStackTrace() ;
            tb.save(entity,TBDao.UPDATE);
            isConn_b 
= false ;
        }
finally{
            
if(session!=null)session.close();
// 拦截器中 绝对不能有这句 ->  SessionManager.closeSession();
        }
        
return false;
    }

    
public boolean onSave(Object entity, Serializable id, Object[] state, String[] propertyNames, Type[] types) throws CallbackException {
        Session session 
= null ;
        
try {
            
if(isConn_b){
                session 
=  tb.openSessionb();
                Transaction tr 
= session.beginTransaction();
                session.save(entity);
                tr.commit();
            }
else{
                tb.save(entity,TBDao.INSERT);
            }
        } 
catch (Exception e) {
            e.printStackTrace() ;
            tb.save(entity,TBDao.INSERT);
            isConn_b 
= false ;
        }
finally{
            
if(session!=null)session.close();
        }
        
return false;
    }

    
public void onDelete(Object entity, Serializable id, Object[] state, String[] propertyNames, Type[] types) throws CallbackException {
        Session session 
= null ;
        
try {
            
if(isConn_b){
                session 
=  tb.openSessionb();
                Transaction tr 
= session.beginTransaction();
                session.delete(entity);
                tr.commit();
            }
else{
                tb.save(entity,TBDao.DELETE);
            }
        } 
catch (Exception e) {
            e.printStackTrace() ;
            tb.save(entity,TBDao.DELETE);
            isConn_b 
= false ;
        }
finally{
            
if(session!=null)session.close();
        }
    }

  ................................


}


后记:
由于一些原因代码写的有些简陋,但功能实现,表达的的意思也还可以(自己有点 ‘买瓜了’哈哈!)。我写出来的目的希望大家能共同进步,这等大家拍砖了 :)



        



评论

# re: hibernate 多服务器数据同步问题(支持延迟同步)  回复  更多评论   

2007-12-03 20:15 by fengzl
帅哥整理个源代码来看看~~

# re: hibernate 多服务器数据同步问题(支持延迟同步)  回复  更多评论   

2007-12-03 21:19 by BeanSoft
是呀... 呵呵 这么好的文章

# re: hibernate 多服务器数据同步问题(支持延迟同步)  回复  更多评论   

2007-12-03 22:17 by G_G
代码 我一定在明天 给出
这就 献丑了

# re: hibernate 多服务器数据同步问题(支持延迟同步)  回复  更多评论   

2007-12-04 10:05 by G_G
涉及到的源代码
http://www.blogjava.net/Files/Good-Game/hbnSrc.rar

# re: hibernate 多服务器数据同步问题(支持延迟同步)  回复  更多评论   

2007-12-04 19:59 by TonySoft
朋友,不知道你的本意是解决什么样的问题.是数据库集群,还是应用服务器集群?

如果是数据库的集群建议你使用开源数据同步框架来做.

# re: hibernate 多服务器数据同步问题(支持延迟同步)  回复  更多评论   

2007-12-05 10:27 by G_G
@TonySoft
哦 是数据同步
开源数据同步框架!能说的具体点吗?
我先在知道的就只有EJB但好想不是开元等级的 !

# re: hibernate 多服务器数据同步问题(支持延迟同步)  回复  更多评论   

2007-12-22 20:57 by 张久强
通用ODBC数据源同步软件:同步专家网络版。

# re: hibernate 多服务器数据同步问题(支持延迟同步)  回复  更多评论   

2007-12-22 20:57 by 张久强
网址:http://www.ahjoe.com/

# re: hibernate 多服务器数据同步问题(支持延迟同步)  回复  更多评论   

2008-04-15 17:45 by wynn
我想请教一下, 你的Interceptor 如果要配置到Spring里,应该怎么做?
谢谢

# re: hibernate 多服务器数据同步问题(支持延迟同步)[未登录]  回复  更多评论   

2008-04-17 09:51 by G_G
在此我还没用 spring
不 好意思

# re: hibernate 多服务器数据同步问题(支持延迟同步)  回复  更多评论   

2008-10-29 09:54 by 游客
好,有价值。

# re: hibernate 多服务器数据同步问题(支持延迟同步)  回复  更多评论   

2010-04-07 12:42 by 张久强
不错.很好.谢谢了!

http://www.51mypc.com

只有注册用户登录后才能发表评论。


网站导航: