GoldenGate for Java adapter介绍二(代码篇)
本示例主要介绍通过实现OGG的接口函数,实现自定义处理增量数据,将数据实时写入到mariadb (OGG官方不支持此数据库,所以只能采用自定义方式实现)。以下是本次示例的4个类:
ConnectionFactory
package sample.handler.jdbc;
import java.sql.Connection; public interface ConnectionFactory { public Connection getConnection() throws Exception; } |
DriverClassConnectionFactory
package sample.handler.jdbc; import java.sql.Connection; import java.sql.Driver; import java.sql.SQLException; import java.util.Properties; import org.slf4j.Logger; import org.slf4j.LoggerFactory; public class DriverClassConnectionFactory implements ConnectionFactory { private static final Logger logger = LoggerFactory.getLogger(DriverClassConnectionFactory.class); protected String driver; protected String url; protected Properties arguments; @Override public Connection getConnection() throws InstantiationException, IllegalAccessException, ClassNotFoundException, SQLException { logger.info(new StringBuilder("Connect to [").append(url).append("] via [").append(driver).append("] with arguments: ").append(arguments).toString()); Driver driv = (Driver)Class.forName(driver).newInstance(); return driv.connect(url, arguments); } /** * @return the driver */ public String getDriver() { return driver; } /** * @param driver the driver to set */ public void setDriver(String driver) { this.driver = driver; } /** * @return the url */ public String getUrl() { return url; } /** * @param url the url to set */ public void setUrl(String url) { this.url = url; } /** * @return the arguments */ public Properties getArguments() { return arguments; } /** * @param arguments the arguments to set */ public void setArguments(Properties arguments) { this.arguments = arguments; } /** * @param user the user to set */ public void setUser(String user) { if (arguments==null) { arguments = new Properties(); } arguments.setProperty("user", user); } /** * @param password the password to set */ public void setPassword(String password) { if (arguments==null) { arguments = new Properties(); } arguments.setProperty("password", password); } } |
OGG接口实现类SimpleJDBCHandler
package sample.handler.jdbc; import java.math.BigDecimal; import java.sql.Connection; import java.sql.Driver; import java.sql.PreparedStatement; import java.sql.SQLException; import java.sql.Timestamp; import java.sql.Types; import java.text.ParseException; import java.text.SimpleDateFormat; import java.util.Arrays; import java.util.Properties; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import oracle.goldengate.datasource.*; import oracle.goldengate.datasource.GGDataSource.Status; import oracle.goldengate.datasource.meta.*; public class SimpleJDBCHandler extends AbstractHandler { private static final Logger logger = LoggerFactory.getLogger(SimpleJDBCHandler.class); protected ConnectionFactory connFactory; protected Connection conn; protected String driver; protected String url; protected Properties arguments; protected PreparedStatement pstmt; protected String lastOp; protected int keyIndex; protected int batchSize = 0; protected int maxBatchSize = 10000; protected long reportCount = 0; protected long lastReportTime; protected long opCount; protected SimpleDateFormat dateFormat; /** * @param connectionFactory the Class name of connection factory to set * @throws ClassNotFoundException * @throws IllegalAccessException * @throws InstantiationException */ public void setConnectionFactory(String connectionFactory) throws InstantiationException, IllegalAccessException, ClassNotFoundException { connFactory = (ConnectionFactory)Class.forName(connectionFactory).newInstance(); } // 获取目标端的DB连接 public Connection getConnection() throws Exception { if (conn==null) { if (connFactory==null) { logger.info(new StringBuilder("Connect to [").append(url).append("] via [").append(driver).append("] with arguments: ").append(arguments).toString()); Driver driv = (Driver)Class.forName(driver).newInstance(); conn = driv.connect(url, arguments); } else { conn = connFactory.getConnection(); } conn.setAutoCommit(false); } return conn; } /* (non-Javadoc) * 一个事务的开始 */ @Override public Status transactionBegin(DsEvent e, DsTransaction tx) { if (logger.isDebugEnabled()) logger.debug("Method transactionBegin invoked. TxID="+tx.getTranID()); //batchSize = 0; return super.transactionBegin(e, tx); } /* (non-Javadoc) *一个新的记录操作 */ @Override public Status operationAdded(DsEvent e, DsTransaction tx, DsOperation op) { if (logger.isDebugEnabled()) logger.debug(new StringBuilder("Method operationAdded invoked. TxID=").append(tx.getTranID()).append(" OpNum=").append(tx.getTotalOps()).toString()); Status status = GGDataSource.Status.OK; if (op.getOperationType().isInsert()) { status = insertAdded(e, tx, op); } else if (op.getOperationType().isUpdate()) { status = updateAdded(e, tx, op); } else if (op.getOperationType().isDelete()) { status = deleteAdded(e, tx, op); } if (reportCount>=100) { if (++opCount == reportCount) { long duration = System.currentTimeMillis() - lastReportTime; logger.info(new StringBuffer().append(opCount).append(" operations have been processed in last ").append(duration/1000).append(" seconds. Rate=").append(opCount*1000/duration).toString()); opCount = 0; lastReportTime = System.currentTimeMillis(); } } return status; } // insert操作 protected Status insertAdded(DsEvent e, DsTransaction tx, DsOperation op) { try { String currentOp = op.getTableName()+".INSERT"; if (!currentOp.equals(lastOp)) { if (batchSize>0 && pstmt!=null) { executeBatch(); } String sql = prepareInsertSql(e, tx, op); System.out.println("insert-sql:"+ sql); if (logger.isDebugEnabled()) logger.debug("Prepare insert. SQL=["+sql+"]"); pstmt = getConnection().prepareStatement(sql); lastOp = currentOp; } TableMetaData tmeta = e.getMetaData().getTableMetaData(op.getTableName()); for (int i=0; iColumnMetaData cmeta = tmeta.getColumnMetaData(i); DsColumn col = op.getColumn(i); bindValue(pstmt, i+1, cmeta, col.getAfter()); } pstmt.addBatch(); if (++batchSize == maxBatchSize) { executeBatch(); } return GGDataSource.Status.OK; } catch (Exception ex) { logger.error("Method insertAdded failed.", ex); return GGDataSource.Status.ABEND; } } // update操作 protected Status updateAdded(DsEvent e, DsTransaction tx, DsOperation op) { try { String sql = prepareUpdateSql(e, tx, op); if (!sql.equals(lastOp)) { if (batchSize>0 && pstmt!=null) { executeBatch(); } if (logger.isDebugEnabled()) logger.debug("Prepare update. SQL=["+sql+"]"); pstmt = getConnection().prepareStatement(sql); lastOp = sql; } System.out.println("update-sql:"+ sql); TableMetaData tmeta = e.getMetaData().getTableMetaData(op.getTableName()); for (int i=0, j=0, k=keyIndex; iColumnMetaData cmeta = tmeta.getColumnMetaData(i); DsColumn col = op.getColumn(i); if (!col.isMissing()) { bindValue(pstmt, ++j, cmeta, col.getAfter()); } if (cmeta.isKeyCol()) { if (col.hasBeforeValue()) { if (logger.isDebugEnabled()) logger.debug(new StringBuilder(op.getTableName().toString()).append(": Key ").append(cmeta.getColumnName()).append(" is updated. before=[").append(col.getBefore().getValue()).append("], after=[").append(col.getAfter().getValue()).append("]").toString()); bindValue(pstmt, ++k, cmeta, col.getBefore()); } else { bindValue(pstmt, ++k, cmeta, col.getAfter()); } } } pstmt.addBatch(); if (++batchSize == maxBatchSize) { executeBatch(); } return GGDataSource.Status.OK; } catch (Exception ex) { logger.error("Method updateAdded failed.", ex); return GGDataSource.Status.ABEND; } } // delete操作 protected Status deleteAdded(DsEvent e, DsTransaction tx, DsOperation op) { try { String currentOp = op.getTableName()+".DELETE"; if (!currentOp.equals(lastOp)) { if (batchSize>0 && pstmt!=null) { executeBatch(); } String sql = prepareDeleteSql(e, tx, op); System.out.println("delete-sql:"+ sql); if (logger.isDebugEnabled()) logger.debug("Prepare delete. SQL=["+sql+"]"); pstmt = getConnection().prepareStatement(sql); lastOp = currentOp; } TableMetaData tmeta = e.getMetaData().getTableMetaData(op.getTableName()); for (int i=0, j=0; iColumnMetaData cmeta = tmeta.getColumnMetaData(i); if (cmeta.isKeyCol()) { DsColumn col = op.getColumn(i); bindValue(pstmt, ++j, cmeta, col.getBefore()); } } pstmt.addBatch(); if (++batchSize == maxBatchSize) { executeBatch(); } return GGDataSource.Status.OK; } catch (Exception ex) { logger.error("Method deleteAdded failed.", ex); return GGDataSource.Status.ABEND; } } // 多个操作批量执行,提升效率 protected int[] executeBatch() throws SQLException { try { int[] result = pstmt.executeBatch(); if (logger.isDebugEnabled()) logger.debug(new StringBuilder("Batch executed. Op=[").append(lastOp).append("] size=").append(batchSize).append(" result=").append(Arrays.toString(result)).toString()); batchSize = 0; return result; } catch(Exception ex) { System.out.println(ex.getMessage()); ex.printStackTrace(); return null; } } protected String prepareInsertSql(DsEvent e, DsTransaction tx, DsOperation op) { String tbName = op.getTableName().toString().toLowerCase().replace(" scetc.", " public."); StringBuilder sql = new StringBuilder("insert into ").append(tbName).append("("); StringBuilder valueClause = new StringBuilder(") values ("); for (ColumnMetaData cmeta : e.getMetaData().getTableMetaData(op.getTableName()).getColumnMetaData()) { sql.append(cmeta.getColumnName()).append(","); valueClause.append("?,"); } sql.deleteCharAt(sql.length()-1).append(valueClause.deleteCharAt(valueClause.length()-1)).append(")"); return sql.toString(); } protected String prepareUpdateSql(DsEvent e, DsTransaction tx, DsOperation op) { String tbName = op.getTableName().toString().toLowerCase().replace(" scetc.", " public."); StringBuilder sql = new StringBuilder("update ").append(tbName).append(" set "); StringBuilder whereClause = new StringBuilder(" where 0=0"); TableMetaData tmeta = e.getMetaData().getTableMetaData(op.getTableName()); keyIndex = 0; for (int i=0; iColumnMetaData cmeta = tmeta.getColumnMetaData(i); DsColumn col = op.getColumn(i); if (!col.isMissing()) { sql.append(cmeta.getColumnName()).append("=?,"); keyIndex++; } if (cmeta.isKeyCol()) { whereClause.append(" and ").append(cmeta.getColumnName()).append("=?"); } } sql.deleteCharAt(sql.length()-1).append(whereClause); return sql.toString(); } protected String prepareDeleteSql(DsEvent e, DsTransaction tx, DsOperation op) { String tbName = op.getTableName().toString().toLowerCase().replace(" scetc.", " public."); StringBuilder sql = new StringBuilder("delete from ").append(tbName).append(" where 0=0"); for (ColumnMetaData cmeta : e.getMetaData().getTableMetaData(op.getTableName()).getKeyColumns()) { sql.append(" and ").append(cmeta.getColumnName()).append("=?"); } return sql.toString(); } protected void bindValue(PreparedStatement pstmt, int i, ColumnMetaData cmeta, DsColumn col) throws SQLException, ParseException { int jdbcType = cmeta.getDataType().getJDBCType(); if (logger.isDebugEnabled()) logger.debug(new StringBuilder("Bind value for column [").append(cmeta.getColumnName()).append("] Type=").append(cmeta.getDataType()).append(" Value=[").append(col.getValue()).append("]").toString()); if (col.isValueNull()) { pstmt.setNull(i, jdbcType); } else if (jdbcType==Types.CHAR || jdbcType==Types.VARCHAR) { pstmt.setString(i, col.getValue()); } else if (jdbcType==Types.NUMERIC || jdbcType==Types.DECIMAL) { pstmt.setBigDecimal(i, new BigDecimal(col.getValue())); } else if (jdbcType==Types.DATE || jdbcType==Types.TIMESTAMP) { //pstmt.setTimestamp(i, new Timestamp(dateFormat.parse(col.getValue()).getTime())); pstmt.setString(i, col.getValue()); } else { pstmt.setObject(i, col.getValue()); } } /* (non-Javadoc) * @see com.goldengate.atg.datasource.AbstractHandler#transactionCommit(com.goldengate.atg.datasource.DsEvent, com.goldengate.atg.datasource.DsTransaction) */ @Override public Status transactionCommit(DsEvent e, DsTransaction tx) { if (logger.isDebugEnabled()) logger.debug("Method transactionCommit invoked. TxID="+tx.getTranID()); if (conn!=null) try { if (batchSize>0 && pstmt!=null) { executeBatch(); } conn.commit(); } catch (SQLException ex) { logger.error("Transaction commit failed. TxID="+tx.getTranID(), ex); } return super.transactionCommit(e, tx); } /* (non-Javadoc) * @see com.goldengate.atg.datasource.AbstractHandler#transactionRollback(com.goldengate.atg.datasource.DsEvent, com.goldengate.atg.datasource.DsTransaction) */ @Override public Status transactionRollback(DsEvent e, DsTransaction tx) { if (logger.isDebugEnabled()) logger.debug("Method transactionRollback invoked. TxID="+tx.getTranID()); if (conn!=null) try { conn.rollback(); } catch (SQLException ex) { logger.error("Transaction rollback failed. TxID="+tx.getTranID(), ex); } return super.transactionRollback(e, tx); } /* (non-Javadoc) * @see com.goldengate.atg.datasource.AbstractHandler#reportStatus() */ @Override public String reportStatus() { logger.debug("Method reportStatus invoked."); return "SimpleJDBCHandler running..."; } /* (non-Javadoc) * @see com.goldengate.atg.datasource.AbstractHandler#destroy() */ @Override public void destroy() { logger.debug("Handler destroy..."); if (conn!=null) try { if (!conn.isClosed()) conn.close(); } catch (SQLException ex) { logger.error("Connection close failed.", ex); } super.destroy(); } /** * @return the driver */ public String getDriver() { return driver; } /** * @param driver the driver to set */ public void setDriver(String driver) { this.driver = driver; } /** * @return the url */ public String getUrl() { return url; } /** * @param url the url to set */ public void setUrl(String url) { this.url = url; } /** * @return the arguments */ public Properties getArguments() { return arguments; } /** * @param arguments the arguments to set */ public void setArguments(Properties arguments) { this.arguments = arguments; } /** * @param user the user to set */ public void setUser(String user) { if (arguments==null) { arguments = new Properties(); } arguments.setProperty("user", user); } /** * @param password the password to set */ public void setPassword(String password) { if (arguments==null) { arguments = new Properties(); } arguments.setProperty("password", password); } public void setDateFormat(String pattern) { dateFormat = new SimpleDateFormat(pattern); } /** * @return the maxBatchSize */ public int getMaxBatchSize() { return maxBatchSize; } /** * @param maxBatchSize the maxBatchSize to set */ public void setMaxBatchSize(int maxBatchSize) { this.maxBatchSize = maxBatchSize; } /** * @return the reportCount */ public long getReportCount() { return reportCount; } /** * @param reportCount the reportCount to set */ public void setReportCount(long reportCount) { if (reportCount<100) { logger.info("A reportCount less than 100 will turn off report."); this.reportCount = 0; } else if (this.reportCount != reportCount) { this.reportCount = reportCount; opCount = 0; lastReportTime = System.currentTimeMillis(); } } } |
从以上代码可以看出,可以通过重载不同的接口函数,获得事务、记录的相关信息,包括操作的表、时间、字段、操作类型及变化的数据等。
最后是基于此Handler创建一个测试类。
TestMariadb
package sample.handler.jdbc; public class TestMariadb { public static void main(String[] args) { SimpleJDBCHandler mariadb = new SimpleJDBCHandler(); gbase.setUrl("jdbc:mariadb://192.168.26.44:5258/mydb? "); gbase.setDateFormat("com.mariadb.jdbc.Driver"); gbase.setUser("root"); gbase.setPassword("admin123"); } } |
通过本示例提供的代码,可以实现将增量数据,从OGG的源端,以完全自定义的方式写入到任何需要的目标端。
转载于:https://www.cnblogs.com/margiex/p/8688604.html
GoldenGate for Java adapter介绍二(代码篇)相关推荐
- goldengate java_GoldenGate for Java Adapter介绍一(原理篇)
前言 Oracle Goldengate在很早前就推出了一个for java的版本,主要目的是方便把关系型数据实时写入到不支持的目标端,如JMS或Redis等key value数据库.在Hadoop刚 ...
- java coin介绍_代码示例中的Java 7:Project Coin
java coin介绍 该博客通过代码示例介绍了一些新的Java 7功能,这些项目在Project Coin一词下进行了概述. Project Coin的目标是向JDK 7添加一组小的语言更改.这些更 ...
- Java实现区块链 --- 代码篇
首先弄清楚需要哪些信息 自身的hash值,上一个区块的hash值,保存的数据信息,时间戳,随机数. 创建一个Block类 package blockChain;import java.util.Dat ...
- java抽象类介绍及代码
抽象类 抽象类本质就是一个类,抽象类经常作为父类使用,子类调用父类的方法或变量 1.凡是有abstract修饰的类叫做抽象类 public abstract class TsetAbstruct {} ...
- java自我介绍_口语化java自我介绍
口语化java自我介绍 自我介绍是对个人特点的总结与归纳.一个好的自我介绍可以让HR很快地了解自己,欣赏自己.下面小编为大家整理了口语化java自我介绍,希望大家喜欢. 口语化java自我介绍(一) ...
- java学习 类变量 类方法_这篇文章主要介绍了JAVA类变量及类方法代码实例详解,文中通过示例代码介绍的非常详细,对大家的学习或者工作具有一定的参考学习价值,需要的朋友可以参考下类变量(...
这篇文章主要介绍了JAVA类变量及类方法代码实例详解,文中通过示例代码介绍的非常详细,对大家的学习或者工作具有一定的参考学习价值,需要的朋友可以参考下 类变量(static) 类变量是该类的所有对象共 ...
- protobuf 3.5 java使用介绍(二)
protobuf 3.5 java使用介绍(二) 上一篇遗留了两个问题: 1,数据模型中有可能会出现数组格式,而数组里面是一个其他的模型,这个怎么来做? 2,构建数据消息的时候,通常会有一个头,一个体 ...
- Java开发规范之代码格式篇(上)
在程序员的世界里有两件最讨厌的事情,第一件事情是讨厌写代码注释,第二件事情是讨厌看别人的代码不写注释.虽然这只是个段子,但也反映了当下很多程序员的心声.下面简单介绍下代码规范的重要性,第一,规范的代码 ...
- 【SemiDrive源码分析】【X9芯片启动流程】21 - MailBox 核间通信机制介绍(代码分析篇)之 Mailbox for Linux 篇
[SemiDrive源码分析][X9芯片启动流程]21 - MailBox 核间通信机制介绍(代码分析篇)之 Mailbox for Linux 篇 一.Mailbox for Linux 驱动框架分 ...
最新文章
- php mssql_result 255 string,PHP连接MSSQL时nvarchar字段长度被截断为255的解决方法
- Confluence 6 生产环境备份策略
- VB.NET 在项目中添加App.config的配置
- 【学习笔记】JDBC:java提供的专门操纵数据库的API JDBC驱动程序的类型 JDBC常用的类与接口
- img打 webpack_webpack打包html里面img后src为“[object Module]”问题
- linux 桌面显示视频播放器,Ubuntu 13.10开启媒体播放器VLC桌面通知的步骤
- ps海报合成教程_如何利用PS制作海报?详细教程来了!
- oracle中如何分页,Oracle中操作分页
- linq distinct 不够用了!
- 第二篇:操纵MySQL数据库(2) - 基于ORM思想的SQLAlchemy库
- mysql中递增列_如何使用JDBC在MySQL中创建带有自动递增列的表?
- [转载] python中集合的表示_Python No.18_集合以及表示符号
- JAVA桌面小程序视频_「小程序JAVA实战」小程序视频组件与api介绍(51)
- OpenVINO DL workbench的docker安装
- linux勒索病毒如何恢复数据,勒索病毒和相应的解密工具
- Ajax传参中文问题
- java 比较excel文件,如何在Excel中使用宏比较两个Excel文件
- 用java给pdf压缩并加密_Java实现多文件压缩加密并重命名压缩文件对象的方法
- Flink从1.7到1.14版本升级汇总
- 英语的计算机软件如何拼写,软的英文单词
热门文章
- 计算机用户域怎么删除,如何查找并删除AD域中多余的计算机帐号?
- C语言中兴面试编程题,中兴一套笔试题及部分答案
- 循序渐进linux——基础知识、服务器搭建、系统管理、性能调优、集群应用_第四讲,Proxmox部署与应用...
- 【若依(ruoyi)】启动时,卡死在:Using existing EHCache named [loginRecordCache]
- graphpad如何做x轴在上方的图_数据分析最有用的25个 Matplotlib图
- android radiobutton_时隔一年,用新知识重构一个Android控件老库
- python 树结构三方包_python第三方库---BeautifulSoup库(搬运)
- dev 居中_div居中 div水平居中代码 | 帮助信息-动天数据
- 雷蛇灯光配置文件_雷蛇猎魂光蛛竞技版机械键盘评测
- axios vue 加载效果动画_在vue中通过axios异步使用echarts