数据库同步

旧城等待， 2022-07-14 14:12 218阅读 0赞

上次发布订阅实现数据库同步，这次用触发器+代码实现了按表的同步。

思路是：1，需同步的A。B两个表都建立触发器，和同步表A-Trigger,B\_Trigger。在对表增删改的时候，用触发器记录存放在同步表中。

2，定时读取两个同步表，然后将更改记录更新到对方的表中。

实现：

1 触发器：建立一个和A结构一样的同步表，并新加一个字段，用来存放操作的类型，如：update，insert，delete。然后在A表上建立触发器，将对A的增删改记录存放到这个同步表中。

use DB 
    ---创建同步表
    select top 1 * into unit_Trigger from unit;
    ----删除临时表中的数据
    delete from unit_Trigger;
    select * from unit_Trigger;
    ---在临时表中增加记录 操作类型的字段 TYPE_TRIGGER
    alter table unit_Trigger add TYPE_TRIGGER varchar(20);
    go
    
    ----创建插入触发器
    if exists(select name from sysobjects where name='unit_insert_trigger' and type='tr')
    drop trigger unit_insert_trigger;
    go
    
    create trigger unit_insert_trigger 
    on unit
    after insert
    as 
    begin
    insert into ProbeDb.dbo.unit_Trigger(C1,C2,C3,TYPE_TRIGGER)
    				(select C1,C2,C3,'insert' from inserted);
    end
    
    go
    
    --创建更改触发器
    if exists(select name from sysobjects where name='unit_update_trigger' and type='tr')
    drop trigger unit_update_trigger;
    
    go
    
    create trigger unit_update_trigger 
    on unit
    after update
    as 
    begin
     insert into ProbeDb.dbo.unit_Trigger (C1,C2,C3,TYPE_TRIGGER)
     				(select C1,C2,C3,'update' from inserted);
    end
    
    go
    
    
    
    --创建删除触发器
    if exists(select name from sysobjects where name='unit_delete_trigger' and type='tr')
    drop trigger unit_delete_trigger;
    go
    
    create trigger unit_delete_trigger 
    on unit
    after delete
    as 
    begin
    insert into ProbeDb.dbo.unit_Trigger(C1,C2,C3,TYPE_TRIGGER)
    
     				(select C1,C2,C3,'delete' from deleted);
    end

2，提取数据并插入

1）编写实体类，bean

2）编写DAO

public class UnitDao {
    
    	/**
    	 * 读取触发器记录表
    	 * @param conn
    	 * @return
    	 */
    	public List readTrigger(Connection conn,String tableName){
    		List<Unit> unitList = new ArrayList<Unit>();
    		String sqlSelect = "select * from unit_Trigger";
    		
    		Statement statem = null;
    		ResultSet rs = null;
    		try {
    			statem = conn.createStatement();
    			rs = statem.executeQuery(sqlSelect);
    			while(rs.next()){
    				Unit unit = new Unit();
    				
    				unit.setC1(rs.getString("C1"));
    				unit.setC2(rs.getString("C2"));
    				unit.setC3(rs.getString("C3"));
    				unit.setTYPE_TRIGGER(rs.getString("TYPE_TRIGGER"));
    				
    				unitList.add(unit);
    			}
    			
    			conn.commit();
    		} catch (SQLException e) {
    			e.printStackTrace();
    		}finally{
    			try {
    				if(rs!= null){
    					rs.close();
    				}
    				if(statem!=null){
    					statem.close();
    				}
    			} catch (SQLException e) {
    				e.printStackTrace();
    			}
    		}
    		return unitList;
    	}
    	
    	/**
    	 * 清空触发器记录表中的数据
    	 * @param conn
    	 */
    	public void clearTrigger(Connection conn,String tbTriggerName){
    		String sqlDelete = "delete from "+tbTriggerName;
    		Statement statem = null;
    		try {
    			statem = conn.createStatement();
    		} catch (SQLException e) {
    			e.printStackTrace();
    		}
    		try {
    			statem.execute(sqlDelete);
    			conn.commit();
    		} catch (SQLException e) {
    			e.printStackTrace();
    		}
    	}
    	
    	/**
    	 * 向触发器表中插入数据
    	 */
    	public void insertTrigger(Connection conn,String tbTriggerName,List<Unit> unitTriggerList){
    		String sqlInsert = "insert into  "+tbTriggerName
    				+ "(C1,C2,C3,TYPE_TRIGGER) "
    				+ "values (?,?,?,?)";
    		for(Unit unitTigger : unitTriggerList){
    			PreparedStatement psInsert;
    			try {
    				psInsert = conn.prepareStatement(sqlInsert);
    				
    				psInsert.setString(1, unitTigger.getC1());
    				psInsert.setString(2, unitTigger.getC2());
    				psInsert.setString(3,unitTigger.getC3());
    				psInsert.setString(4, unitTigger.getTYPE_TRIGGER());
    				
    				
    				psInsert.execute();
    				conn.commit();
    			} catch (SQLException e) {
    				e.printStackTrace();
    			}
    			
    		}
    	}
    
    	/**
    	 * 按主键查询数据
    	 * @param conn
    	 * @return
    	 */
    	public Unit selectByKey(Connection conn,String tableName, String C1){
    		String sql = "select * from "+tableName+" where C1 = '"+C1+"'";
    		Unit unit = null;
    		Statement statem = null;
    		try {
    			statem = conn.createStatement();
    			ResultSet rs = statem.executeQuery(sql);
    			while(rs.next()){
    				unit = new Unit();
    			
    				unit.setC1(rs.getString("C1"));
    				unit.setC2(rs.getString("C2"));
    				unit.setC3(rs.getString("C3"));
    
    				
    				
    			}
    		} catch (SQLException e) {
    			e.printStackTrace();
    		}
    		return unit;
    	}
    	
    	/**
    	 * 插入一条记录
    	 * @param conn
    	 * @return
    	 */
    	public boolean insert(Connection conn,String tableName ,Unit unit){
    		boolean flag = false;
    		int insNum = 0;
    		String sqlInsertunit = "insert into "+tableName+" (C1,C2,C3) "
    				+ "values (?,?,?)";
    		PreparedStatement psInsert;
    		try {
    			psInsert = conn.prepareStatement(sqlInsertunit);
    			psInsert.setString(1,unit.getC1());
    			psInsert.setString(2, unit.getC2());
    			psInsert.setString(3, unit.getC3());
    
    			insNum = psInsert.executeUpdate();
    			conn.commit();
    			if (insNum == 1){
    				flag = true;
    			}
    		} catch (SQLException e) {
    			try {
    				conn.rollback();
    			} catch (SQLException e1) {
    				e1.printStackTrace();
    			}
    
    			e.printStackTrace();
    		}
    		return flag;
    	}
    	/**
    	 * 更新数据
    	 * @param conn
    	 * @return
    	 */
    	public boolean update(Connection conn,String tableName,Unit unit){
    		boolean flag = false;
    		
    	
    		String sqlUpdateunit = "update "+tableName+" set C1=?,C2=?,C3=?) "
    				+ "where C1=?";
    		int udNum = 0;
    		PreparedStatement psUpdate;
    		try {
    			psUpdate = conn.prepareStatement(sqlUpdateunit);
    
    			psUpdate.setString(1, unit.getC1());
    			psUpdate.setString(2, unit.getC2());
    			psUpdate.setString(3, unit.getC3());
    			
    			udNum = psUpdate.executeUpdate();
    			conn.commit();
    			if(udNum == 1){
    				flag = true;
    			}
    		} catch (SQLException e) {
    			try {
    				conn.rollback();
    			} catch (SQLException e1) {
    				e1.printStackTrace();
    			}
    
    			e.printStackTrace();
    		}
    		
    		return flag;
    	}
    	
    	/**
    	 * 删除一条数据
    	 * @param conn
    	 * @return
    	 */
    	public boolean deleteByKey(Connection conn,String tableName,Unit unit){
    		int delNum = 0;
    		boolean flag = false;
    		String sqlDeleteunit = "delete from "+tableName+" where C1=? ";
    		PreparedStatement psDelete;
    		try {
    			psDelete = conn.prepareStatement(sqlDeleteunit);
    			psDelete.setString(1, unit.getC1());
    			delNum = psDelete.executeUpdate();
    			conn.commit();
    			if(delNum == 1){
    				flag = true;
    			}
    		} catch (SQLException e) {
    			try {
    				conn.rollback();
    			} catch (SQLException e1) {
    				e1.printStackTrace();
    			}
    			e.printStackTrace();
    		}
    		return flag;
    	}
    	
    	public List readUnit(Connection conn,String tableName){
    		List unitList = new ArrayList();
    		String sqlSelect = "select * from "+tableName;
    		
    		Statement statem = null;
    		ResultSet rs = null;
    		try {
    			statem = conn.createStatement();
    			rs = statem.executeQuery(sqlSelect);
    			while(rs.next()){
    				Unit unit = new Unit();
    				unit.setC1(rs.getString("C1"));
    				unit.setC2(rs.getString("C2"));
    				unit.setC3(rs.getString("C3"));
    
    				unitList.add(unit);
    			}
    			
    			conn.commit();
    		} catch (SQLException e) {
    			e.printStackTrace();
    		}finally{
    			try {
    				if(rs!= null){
    					rs.close();
    				}
    				if(statem!=null){
    					statem.close();
    				}
    			} catch (SQLException e) {
    				e.printStackTrace();
    			}
    		}
    		
    		return unitList;
    	}
    	
    }

3） 提取数据并更新

public class UnitAccess  implements Runnable{
    
    	private final String tableNameA = "Unit";
    	private final String tableNameB = "TB";
    	private final String tbTriggerNameA = "unit_Trigger";
    	private final String tbTriggerNameB = "TB_Trigger";
        @Override  
        public void run() {  
        	boolean keyA = false;
        	boolean keyB = false;
        	Connection connA = null;
        	Connection connB = null;
        	connA = DBCPConn.getConnectionA();
        	connB = DBCPConn.getConnectionB();
        	UnitDao unitDao = new UnitDao();
        	
        	List unitListA = unitDao.readTrigger(connA,tbTriggerNameA);
        	unitDao.clearTrigger(connA,tbTriggerNameA);
        	keyA = true;
        	List unitListB = unitDao.readTrigger(connB,tbTriggerNameB);
        	unitDao.clearTrigger(connB,tbTriggerNameB);
        	keyB = true;
        	while(!keyA || !keyB){
        		System.out.println("等待全部取数完成");
        	}
        	writeUnit(connB,tableNameB, tbTriggerNameB, unitListA);
        	writeUnit(connA,tableNameA, tbTriggerNameA, unitListB);
        }	
        
         public void writeUnit(Connection conn,String tableName,String tbTriggerName,List<Unit> unitList){
    		/**
    		 * 1,将Trigger表中的数据保存起来
    		 */
    		List<Unit> unitTriggerList = new ArrayList<Unit>();
    		UnitDao unitDao = new UnitDao();
    		unitTriggerList = unitDao.readTrigger(conn,tbTriggerName);
    		unitDao.clearTrigger(conn,tbTriggerName);
    		/**
    		 *2. 将传过来的数据更新到数据库
    		 */
    		for(Unit unit: unitList){
    			switch(unit.getTYPE_TRIGGER()){
    			case "insert":
    				Unit unitinsert = unitDao.selectByKey(conn, tableName,unit.getF_DWBH());
    				if( unitinsert == null){
    					unitDao.insert(conn,tableName, unit);
    				}
    				else{
    					unitDao.update(conn,tableName, unit);
    				}
    				break;
    			case "update":
    				Unit unitUpdate = unitDao.selectByKey(conn ,tableName,unit.getF_DWBH());
    				if( unitUpdate == null){
    					unitDao.insert(conn,tableName, unit);
    				}
    				else{
    					unitDao.update(conn,tableName, unit);
    				}
    				break;
    			case "delete":
    				unitDao.deleteByKey(conn,tableName, unit);
    				break;
    			}
    		}
        			
    		/**
    		 *3 删除更新产生的Trigger数据
    		 */
    		unitDao.clearTrigger(conn,tbTriggerName);
    		
    		/**
    		 * 4.恢复Trigger 原先的数据
    		 */
    		if(!unitTriggerList.isEmpty()){
    			unitDao.insertTrigger(conn, tbTriggerName,unitTriggerList);
    		}
        }
    
    }