chenjl/canal
1
0
Fork 0
mirror of https://github.com/alibaba/canal.git synced 2026-11-04 00:30:15 +08:00
Code Issues Packages Projects Releases Wiki Activity

统一Canal 依赖的Spring 版本，升级ibatis 至mybatis (#3182)

Browse Source 
* fix issue:#561 增加一个外部参数控制，允许位点不存在时自动重置到当前位点

* Add MariaDB GTID support

* 1. fix maven central repository url access error.(Update to https)
2. upgrade spring version to 5.0.5.RELEASE
3. upgrade ibatis to mybatis

* fix unit test run error
This commit is contained in:
文哥
2020-11-20 09:24:56 +08:00
committed by GitHub
parent a5c09c521e
commit 20add85ff6
27 changed files with 207 additions and 176 deletions
Download Patch File Download Diff File Expand all files Collapse all files

12
deployer/src/main/resources/spring/default-instance.xml
View File

@@ -15,22 +15,22 @@
<bean id="instance" class="com.alibaba.otter.canal.instance.spring.CanalInstanceWithSpring"> <bean id="instance" class="com.alibaba.otter.canal.instance.spring.CanalInstanceWithSpring">
<property name="destination" value="${canal.instance.destination}" /> <property name="destination" value="${canal.instance.destination}" />
<property name="eventParser"> <property name="eventParser">
<ref local="eventParser" /> <ref bean="eventParser" />
</property> </property>
<property name="eventSink"> <property name="eventSink">
<ref local="eventSink" /> <ref bean="eventSink" />
</property> </property>
<property name="eventStore"> <property name="eventStore">
<ref local="eventStore" /> <ref bean="eventStore" />
</property> </property>
<property name="metaManager"> <property name="metaManager">
<ref local="metaManager" /> <ref bean="metaManager" />
</property> </property>
<property name="alarmHandler"> <property name="alarmHandler">
<ref local="alarmHandler" /> <ref bean="alarmHandler" />
</property> </property>
<property name="mqConfig"> <property name="mqConfig">
<ref local="mqConfig" /> <ref bean="mqConfig" />
</property> </property>
</bean> </bean>

12
deployer/src/main/resources/spring/file-instance.xml
View File

@@ -15,22 +15,22 @@
<bean id="instance" class="com.alibaba.otter.canal.instance.spring.CanalInstanceWithSpring"> <bean id="instance" class="com.alibaba.otter.canal.instance.spring.CanalInstanceWithSpring">
<property name="destination" value="${canal.instance.destination}" /> <property name="destination" value="${canal.instance.destination}" />
<property name="eventParser"> <property name="eventParser">
<ref local="eventParser" /> <ref bean="eventParser" />
</property> </property>
<property name="eventSink"> <property name="eventSink">
<ref local="eventSink" /> <ref bean="eventSink" />
</property> </property>
<property name="eventStore"> <property name="eventStore">
<ref local="eventStore" /> <ref bean="eventStore" />
</property> </property>
<property name="metaManager"> <property name="metaManager">
<ref local="metaManager" /> <ref bean="metaManager" />
</property> </property>
<property name="alarmHandler"> <property name="alarmHandler">
<ref local="alarmHandler" /> <ref bean="alarmHandler" />
</property> </property>
<property name="mqConfig"> <property name="mqConfig">
<ref local="mqConfig" /> <ref bean="mqConfig" />
</property> </property>
</bean> </bean>

12
deployer/src/main/resources/spring/group-instance.xml
View File

@@ -15,22 +15,22 @@
<bean id="instance" class="com.alibaba.otter.canal.instance.spring.CanalInstanceWithSpring"> <bean id="instance" class="com.alibaba.otter.canal.instance.spring.CanalInstanceWithSpring">
<property name="destination" value="${canal.instance.destination}" /> <property name="destination" value="${canal.instance.destination}" />
<property name="eventParser"> <property name="eventParser">
<ref local="eventParser" /> <ref bean="eventParser" />
</property> </property>
<property name="eventSink"> <property name="eventSink">
<ref local="eventSink" /> <ref bean="eventSink" />
</property> </property>
<property name="eventStore"> <property name="eventStore">
<ref local="eventStore" /> <ref bean="eventStore" />
</property> </property>
<property name="metaManager"> <property name="metaManager">
<ref local="metaManager" /> <ref bean="metaManager" />
</property> </property>
<property name="alarmHandler"> <property name="alarmHandler">
<ref local="alarmHandler" /> <ref bean="alarmHandler" />
</property> </property>
<property name="mqConfig"> <property name="mqConfig">
<ref local="mqConfig" /> <ref bean="mqConfig" />
</property> </property>
</bean> </bean>

12
deployer/src/main/resources/spring/memory-instance.xml
View File

@@ -15,22 +15,22 @@
<bean id="instance" class="com.alibaba.otter.canal.instance.spring.CanalInstanceWithSpring"> <bean id="instance" class="com.alibaba.otter.canal.instance.spring.CanalInstanceWithSpring">
<property name="destination" value="${canal.instance.destination}" /> <property name="destination" value="${canal.instance.destination}" />
<property name="eventParser"> <property name="eventParser">
<ref local="eventParser" /> <ref bean="eventParser" />
</property> </property>
<property name="eventSink"> <property name="eventSink">
<ref local="eventSink" /> <ref bean="eventSink" />
</property> </property>
<property name="eventStore"> <property name="eventStore">
<ref local="eventStore" /> <ref bean="eventStore" />
</property> </property>
<property name="metaManager"> <property name="metaManager">
<ref local="metaManager" /> <ref bean="metaManager" />
</property> </property>
<property name="alarmHandler"> <property name="alarmHandler">
<ref local="alarmHandler" /> <ref bean="alarmHandler" />
</property> </property>
<property name="mqConfig"> <property name="mqConfig">
<ref local="mqConfig" /> <ref bean="mqConfig" />
</property> </property>
</bean> </bean>

6
deployer/src/main/resources/spring/tsdb/h2-tsdb.xml
View File

@@ -46,16 +46,16 @@
<property name="validationQuery" value="SELECT 1" /> <property name="validationQuery" value="SELECT 1" />
</bean> </bean>
<bean id="sqlMapClient" class="org.springframework.orm.ibatis.SqlMapClientFactoryBean"> <bean id="sqlSessionFactory" class="org.mybatis.spring.SqlSessionFactoryBean">
<property name="dataSource" ref="dataSource"/> <property name="dataSource" ref="dataSource"/>
<property name="configLocation" value="classpath:spring/tsdb/sql-map/sqlmap-config.xml"/> <property name="configLocation" value="classpath:spring/tsdb/sql-map/sqlmap-config.xml"/>
</bean> </bean>
<bean id="metaHistoryDAO" class="com.alibaba.otter.canal.parse.inbound.mysql.tsdb.dao.MetaHistoryDAO"> <bean id="metaHistoryDAO" class="com.alibaba.otter.canal.parse.inbound.mysql.tsdb.dao.MetaHistoryDAO">
<property name="sqlMapClient" ref="sqlMapClient"/> <property name="sqlSessionFactory" ref="sqlSessionFactory"/>
</bean> </bean>
<bean id="metaSnapshotDAO" class="com.alibaba.otter.canal.parse.inbound.mysql.tsdb.dao.MetaSnapshotDAO"> <bean id="metaSnapshotDAO" class="com.alibaba.otter.canal.parse.inbound.mysql.tsdb.dao.MetaSnapshotDAO">
<property name="sqlMapClient" ref="sqlMapClient"/> <property name="sqlSessionFactory" ref="sqlSessionFactory"/>
</bean> </bean>
</beans> </beans>

6
deployer/src/main/resources/spring/tsdb/mysql-tsdb.xml
View File

@@ -48,16 +48,16 @@
<property name="useUnfairLock" value="true" /> <property name="useUnfairLock" value="true" />
</bean> </bean>
<bean id="sqlMapClient" class="org.springframework.orm.ibatis.SqlMapClientFactoryBean"> <bean id="sqlSessionFactory" class="org.mybatis.spring.SqlSessionFactoryBean">
<property name="dataSource" ref="dataSource"/> <property name="dataSource" ref="dataSource"/>
<property name="configLocation" value="classpath:spring/tsdb/sql-map/sqlmap-config.xml"/> <property name="configLocation" value="classpath:spring/tsdb/sql-map/sqlmap-config.xml"/>
</bean> </bean>
<bean id="metaHistoryDAO" class="com.alibaba.otter.canal.parse.inbound.mysql.tsdb.dao.MetaHistoryDAO"> <bean id="metaHistoryDAO" class="com.alibaba.otter.canal.parse.inbound.mysql.tsdb.dao.MetaHistoryDAO">
<property name="sqlMapClient" ref="sqlMapClient"/> <property name="sqlSessionFactory" ref="sqlSessionFactory"/>
</bean> </bean>
<bean id="metaSnapshotDAO" class="com.alibaba.otter.canal.parse.inbound.mysql.tsdb.dao.MetaSnapshotDAO"> <bean id="metaSnapshotDAO" class="com.alibaba.otter.canal.parse.inbound.mysql.tsdb.dao.MetaSnapshotDAO">
<property name="sqlMapClient" ref="sqlMapClient"/> <property name="sqlSessionFactory" ref="sqlSessionFactory"/>
</bean> </bean>
</beans> </beans>

20
deployer/src/main/resources/spring/tsdb/sql-map/sqlmap-config.xml
View File

@@ -1,8 +1,14 @@
<?xml version="1.0" encoding="UTF-8" ?> <?xml version="1.0" encoding="UTF-8" ?>
<!DOCTYPE sqlMapConfig PUBLIC "-//iBATIS.com//DTD SQL Map Config 2.0//EN" <!DOCTYPE configuration PUBLIC "-//mybatis.org//DTD Config 3.0//EN"
"http://www.ibatis.com/dtd/sql-map-config-2.dtd"> "http://mybatis.org/dtd/mybatis-3-config.dtd">
<sqlMapConfig> <configuration>
<settings useStatementNamespaces="true"/> <typeAliases>
<sqlMap resource="spring/tsdb/sql-map/sqlmap_history.xml"/> <typeAlias alias="metaHistoryDO" type="com.alibaba.otter.canal.parse.inbound.mysql.tsdb.dao.MetaHistoryDO"/>
<sqlMap resource="spring/tsdb/sql-map/sqlmap_snapshot.xml"/> <typeAlias alias="metaSnapshotDO" type="com.alibaba.otter.canal.parse.inbound.mysql.tsdb.dao.MetaSnapshotDO"/>
</sqlMapConfig> </typeAliases>
<mappers>
<mapper resource="spring/tsdb/sql-map/sqlmap_history.xml"/>
<mapper resource="spring/tsdb/sql-map/sqlmap_snapshot.xml"/>
</mappers>
</configuration>

23
deployer/src/main/resources/spring/tsdb/sql-map/sqlmap_history.xml
View File

@@ -1,7 +1,6 @@
<?xml version="1.0" encoding="UTF-8" ?> <?xml version="1.0" encoding="UTF-8" ?>
<!DOCTYPE sqlMap PUBLIC "-//ibatis.apache.org//DTD SQL Map 2.0//EN" "http://ibatis.apache.org/dtd/sql-map-2.dtd" > <!DOCTYPE mapper PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN" "http://mybatis.org/dtd/mybatis-3-mapper.dtd">
<sqlMap namespace="meta_history"> <mapper namespace="com.alibaba.otter.canal.parse.inbound.mysql.tsdb.dao.MetaHistoryMapper">
<typeAlias alias="metaHistoryDO" type="com.alibaba.otter.canal.parse.inbound.mysql.tsdb.dao.MetaHistoryDO"/>
<sql id="allColumns"> <sql id="allColumns">
<![CDATA[ <![CDATA[
gmt_create,gmt_modified,destination,binlog_file,binlog_offest,binlog_master_id,binlog_timestamp,use_schema,sql_schema,sql_table,sql_text,sql_type,extra gmt_create,gmt_modified,destination,binlog_file,binlog_offest,binlog_master_id,binlog_timestamp,use_schema,sql_schema,sql_table,sql_text,sql_type,extra
@@ -15,31 +14,31 @@
]]> ]]>
</sql> </sql>
<select id="findByTimestamp" parameterClass="java.util.Map" resultClass="metaHistoryDO"> <select id="findByTimestamp" parameterType="java.util.Map" resultType="metaHistoryDO">
select select
<include refid="allVOColumns"/> <include refid="allVOColumns"/>
from meta_history a from meta_history a
<![CDATA[ <![CDATA[
where destination = #destination# and binlog_timestamp >= #snapshotTimestamp# and binlog_timestamp <= #timestamp# where destination = #{destination} and binlog_timestamp >= #{snapshotTimestamp} and binlog_timestamp <= #{timestamp}
order by binlog_timestamp asc,id asc order by binlog_timestamp asc,id asc
]]> ]]>
</select> </select>
<insert id="insert" parameterClass="metaHistoryDO"> <insert id="insert" parameterType="metaHistoryDO">
insert into meta_history (<include refid="allColumns"/>) insert into meta_history (<include refid="allColumns"/>)
values(CURRENT_TIMESTAMP,CURRENT_TIMESTAMP,#destination#,#binlogFile#,#binlogOffest#,#binlogMasterId#,#binlogTimestamp#,#useSchema#,#sqlSchema#,#sqlTable#,#sqlText#,#sqlType#,#extra#) values(CURRENT_TIMESTAMP,CURRENT_TIMESTAMP,#{destination},#{binlogFile},#{binlogOffest},#{binlogMasterId},#{binlogTimestamp},#{useSchema},#{sqlSchema},#{sqlTable},#{sqlText},#{sqlType},#{extra})
</insert> </insert>
<delete id="deleteByName" parameterClass="java.util.Map"> <delete id="deleteByName" parameterType="java.util.Map">
delete from meta_history delete from meta_history
where destination=#destination# where destination=#{destination}
</delete> </delete>
<delete id="deleteByTimestamp" parameterClass="java.util.Map"> <delete id="deleteByTimestamp" parameterType="java.util.Map">
<![CDATA[ <![CDATA[
delete from meta_history delete from meta_history
where destination=#destination# and binlog_timestamp < #timestamp# where destination=#{destination} and binlog_timestamp < #{timestamp}
]]> ]]>
</delete> </delete>
</sqlMap> </mapper>

31
deployer/src/main/resources/spring/tsdb/sql-map/sqlmap_snapshot.xml
View File

@@ -1,9 +1,6 @@
<?xml version="1.0" encoding="UTF-8" ?> <?xml version="1.0" encoding="UTF-8" ?>
<!DOCTYPE sqlMap PUBLIC "-//ibatis.apache.org//DTD SQL Map 2.0//EN" "http://ibatis.apache.org/dtd/sql-map-2.dtd" > <!DOCTYPE mapper PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN" "http://mybatis.org/dtd/mybatis-3-mapper.dtd">
<sqlMap namespace="meta_snapshot"> <mapper namespace="com.alibaba.otter.canal.parse.inbound.mysql.tsdb.dao.MetaSnapshotMapper">
<typeAlias alias="metaSnapshotDO" type="com.alibaba.otter.canal.parse.inbound.mysql.tsdb.dao.MetaSnapshotDO"/>
<typeAlias alias="tableMetaSnapshotDO"
type="com.alibaba.middleware.jingwei.biz.dataobject.CanalTableMetaSnapshotDO"/>
<sql id="allColumns"> <sql id="allColumns">
<![CDATA[ <![CDATA[
gmt_create,gmt_modified,destination,binlog_file,binlog_offest,binlog_master_id,binlog_timestamp,data,extra gmt_create,gmt_modified,destination,binlog_file,binlog_offest,binlog_master_id,binlog_timestamp,data,extra
@@ -16,36 +13,36 @@
]]> ]]>
</sql> </sql>
<select id="findByTimestamp" parameterClass="java.util.Map" resultClass="metaSnapshotDO"> <select id="findByTimestamp" parameterType="java.util.Map" resultType="metaSnapshotDO">
select <include refid="allVOColumns"/> select <include refid="allVOColumns"/>
<![CDATA[ <![CDATA[
from meta_snapshot a from meta_snapshot a
where destination = #destination# and binlog_timestamp < #timestamp# where destination = #{destination} and binlog_timestamp < #{timestamp}
order by binlog_timestamp desc,id desc order by binlog_timestamp desc,id desc
limit 1 limit 1
]]> ]]>
</select> </select>
<insert id="insert" parameterClass="metaSnapshotDO"> <insert id="insert" parameterType="metaSnapshotDO">
insert into meta_snapshot (<include refid="allColumns"/>) insert into meta_snapshot (<include refid="allColumns"/>)
values(CURRENT_TIMESTAMP,CURRENT_TIMESTAMP,#destination#,#binlogFile#,#binlogOffest#,#binlogMasterId#,#binlogTimestamp#,#data#,#extra#) values(CURRENT_TIMESTAMP,CURRENT_TIMESTAMP,#{destination},#{binlogFile},#{binlogOffest},#{binlogMasterId},#{binlogTimestamp},#{data},#{extra})
</insert> </insert>
<update id="update" parameterClass="metaSnapshotDO"> <update id="update" parameterType="metaSnapshotDO">
update meta_snapshot set gmt_modified=now(), update meta_snapshot set gmt_modified=now(),
binlog_file=#binlogFile#,binlog_offest=#binlogOffest#,binlog_master_id=#binlogMasterId#,binlog_timestamp=#binlogTimestamp#,data=#data#,extra=#extra# binlog_file=#{binlogFile},binlog_offest=#{binlogOffest},binlog_master_id=#{binlogMasterId},binlog_timestamp=#{binlogTimestamp},data=#{data},extra=#{extra}
where destination=#destination# and binlog_timestamp=0 where destination=#{destination} and binlog_timestamp=0
</update> </update>
<delete id="deleteByName" parameterClass="java.util.Map"> <delete id="deleteByName" parameterType="java.util.Map">
delete from meta_snapshot delete from meta_snapshot
where destination=#destination# where destination=#{destination}
</delete> </delete>
<delete id="deleteByTimestamp" parameterClass="java.util.Map"> <delete id="deleteByTimestamp" parameterType="java.util.Map">
<![CDATA[ <![CDATA[
delete from meta_snapshot delete from meta_snapshot
where destination=#destination# and binlog_timestamp < #timestamp# and binlog_timestamp > 0 where destination=#{destination} and binlog_timestamp < #{timestamp} and binlog_timestamp > 0
]]> ]]>
</delete> </delete>
</sqlMap> </mapper>

2
example/src/main/java/com/alibaba/otter/canal/example/SimpleCanalClientTest.java
View File

@@ -22,7 +22,7 @@ public class SimpleCanalClientTest extends AbstractCanalClientTest {
// 根据ip，直接创建链接，无HA的功能 // 根据ip，直接创建链接，无HA的功能
String destination = "example"; String destination = "example";
String ip = AddressUtils.getHostIp(); String ip = AddressUtils.getHostIp();
CanalConnector connector = CanalConnectors.newSingleConnector(new InetSocketAddress(ip, 11111), CanalConnector connector = CanalConnectors.newSingleConnector(new InetSocketAddress("127.0.0.1", 11111),
destination, destination,
"canal", "canal",
"canal"); "canal");

10
instance/spring/src/test/resources/spring/default-instance.xml
View File

@@ -34,19 +34,19 @@
<bean id="instance" class="com.alibaba.otter.canal.instance.spring.CanalInstanceWithSpring"> <bean id="instance" class="com.alibaba.otter.canal.instance.spring.CanalInstanceWithSpring">
<property name="destination" value="${canal.instance.destination}" /> <property name="destination" value="${canal.instance.destination}" />
<property name="eventParser"> <property name="eventParser">
<ref local="eventParser" /> <ref bean="eventParser" />
</property> </property>
<property name="eventSink"> <property name="eventSink">
<ref local="eventSink" /> <ref bean="eventSink" />
</property> </property>
<property name="eventStore"> <property name="eventStore">
<ref local="eventStore" /> <ref bean="eventStore" />
</property> </property>
<property name="metaManager"> <property name="metaManager">
<ref local="metaManager" /> <ref bean="metaManager" />
</property> </property>
<property name="alarmHandler"> <property name="alarmHandler">
<ref local="alarmHandler" /> <ref bean="alarmHandler" />
</property> </property>
</bean> </bean>

10
instance/spring/src/test/resources/spring/file-instance.xml
View File

@@ -34,19 +34,19 @@
<bean id="instance" class="com.alibaba.otter.canal.instance.spring.CanalInstanceWithSpring"> <bean id="instance" class="com.alibaba.otter.canal.instance.spring.CanalInstanceWithSpring">
<property name="destination" value="${canal.instance.destination}" /> <property name="destination" value="${canal.instance.destination}" />
<property name="eventParser"> <property name="eventParser">
<ref local="eventParser" /> <ref bean="eventParser" />
</property> </property>
<property name="eventSink"> <property name="eventSink">
<ref local="eventSink" /> <ref bean="eventSink" />
</property> </property>
<property name="eventStore"> <property name="eventStore">
<ref local="eventStore" /> <ref bean="eventStore" />
</property> </property>
<property name="metaManager"> <property name="metaManager">
<ref local="metaManager" /> <ref bean="metaManager" />
</property> </property>
<property name="alarmHandler"> <property name="alarmHandler">
<ref local="alarmHandler" /> <ref bean="alarmHandler" />
</property> </property>
</bean> </bean>

10
instance/spring/src/test/resources/spring/group-instance.xml
View File

@@ -34,19 +34,19 @@
<bean id="instance" class="com.alibaba.otter.canal.instance.spring.CanalInstanceWithSpring"> <bean id="instance" class="com.alibaba.otter.canal.instance.spring.CanalInstanceWithSpring">
<property name="destination" value="${canal.instance.destination}" /> <property name="destination" value="${canal.instance.destination}" />
<property name="eventParser"> <property name="eventParser">
<ref local="eventParser" /> <ref bean="eventParser" />
</property> </property>
<property name="eventSink"> <property name="eventSink">
<ref local="eventSink" /> <ref bean="eventSink" />
</property> </property>
<property name="eventStore"> <property name="eventStore">
<ref local="eventStore" /> <ref bean="eventStore" />
</property> </property>
<property name="metaManager"> <property name="metaManager">
<ref local="metaManager" /> <ref bean="metaManager" />
</property> </property>
<property name="alarmHandler"> <property name="alarmHandler">
<ref local="alarmHandler" /> <ref bean="alarmHandler" />
</property> </property>
</bean> </bean>

10
instance/spring/src/test/resources/spring/memory-instance.xml
View File

@@ -34,19 +34,19 @@
<bean id="instance" class="com.alibaba.otter.canal.instance.spring.CanalInstanceWithSpring"> <bean id="instance" class="com.alibaba.otter.canal.instance.spring.CanalInstanceWithSpring">
<property name="destination" value="${canal.instance.destination}" /> <property name="destination" value="${canal.instance.destination}" />
<property name="eventParser"> <property name="eventParser">
<ref local="eventParser" /> <ref bean="eventParser" />
</property> </property>
<property name="eventSink"> <property name="eventSink">
<ref local="eventSink" /> <ref bean="eventSink" />
</property> </property>
<property name="eventStore"> <property name="eventStore">
<ref local="eventStore" /> <ref bean="eventStore" />
</property> </property>
<property name="metaManager"> <property name="metaManager">
<ref local="metaManager" /> <ref bean="metaManager" />
</property> </property>
<property name="alarmHandler"> <property name="alarmHandler">
<ref local="alarmHandler" /> <ref bean="alarmHandler" />
</property> </property>
</bean> </bean>

8
parse/pom.xml
View File

@@ -54,8 +54,12 @@
<artifactId>mysql-connector-java</artifactId> <artifactId>mysql-connector-java</artifactId>
</dependency> </dependency>
<dependency> <dependency>
<groupId>org.apache.ibatis</groupId> <groupId>org.mybatis</groupId>
<artifactId>ibatis-sqlmap</artifactId> <artifactId>mybatis-spring</artifactId>
</dependency>
<dependency>
<groupId>org.mybatis</groupId>
<artifactId>mybatis</artifactId>
</dependency> </dependency>
<dependency> <dependency>
<groupId>com.h2database</groupId> <groupId>com.h2database</groupId>

6
parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/tsdb/dao/MetaBaseDAO.java
View File

@@ -9,14 +9,14 @@ import javax.sql.DataSource;
import org.apache.commons.io.IOUtils; import org.apache.commons.io.IOUtils;
import org.apache.commons.lang.StringUtils; import org.apache.commons.lang.StringUtils;
import org.springframework.orm.ibatis.support.SqlMapClientDaoSupport; import org.mybatis.spring.support.SqlSessionDaoSupport;
/** /**
* @author agapple 2017年10月14日 上午1:05:22 * @author agapple 2017年10月14日 上午1:05:22
* @since 1.0.25 * @since 1.0.25
*/ */
@SuppressWarnings("deprecation") @SuppressWarnings("deprecation")
public class MetaBaseDAO extends SqlMapClientDaoSupport { public class MetaBaseDAO extends SqlSessionDaoSupport {
protected boolean isH2 = false; protected boolean isH2 = false;
@@ -24,7 +24,7 @@ public class MetaBaseDAO extends SqlMapClientDaoSupport {
Connection conn = null; Connection conn = null;
InputStream input = null; InputStream input = null;
try { try {
DataSource dataSource = getDataSource(); DataSource dataSource = getSqlSessionFactory().getConfiguration().getEnvironment().getDataSource();
conn = dataSource.getConnection(); conn = dataSource.getConnection();
String name = "mysql"; String name = "mysql";
isH2 = isH2(conn); isH2 = isH2(conn);

8
parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/tsdb/dao/MetaHistoryDAO.java
View File

@@ -15,7 +15,7 @@ import com.google.common.collect.Maps;
public class MetaHistoryDAO extends MetaBaseDAO { public class MetaHistoryDAO extends MetaBaseDAO {
public Long insert(MetaHistoryDO metaDO) { public Long insert(MetaHistoryDO metaDO) {
return (Long) getSqlMapClientTemplate().insert("meta_history.insert", metaDO); return getSqlSession().getMapper(MetaHistoryMapper.class).insert(metaDO);
} }
public List<MetaHistoryDO> findByTimestamp(String destination, Long snapshotTimestamp, Long timestamp) { public List<MetaHistoryDO> findByTimestamp(String destination, Long snapshotTimestamp, Long timestamp) {
@@ -23,13 +23,13 @@ public class MetaHistoryDAO extends MetaBaseDAO {
params.put("destination", destination); params.put("destination", destination);
params.put("snapshotTimestamp", snapshotTimestamp == null ? 0L : snapshotTimestamp); params.put("snapshotTimestamp", snapshotTimestamp == null ? 0L : snapshotTimestamp);
params.put("timestamp", timestamp == null ? 0L : timestamp); params.put("timestamp", timestamp == null ? 0L : timestamp);
return (List<MetaHistoryDO>) getSqlMapClientTemplate().queryForList("meta_history.findByTimestamp", params); return getSqlSession().getMapper(MetaHistoryMapper.class).findByTimestamp(params);
} }
public Integer deleteByName(String destination) { public Integer deleteByName(String destination) {
HashMap params = Maps.newHashMapWithExpectedSize(2); HashMap params = Maps.newHashMapWithExpectedSize(2);
params.put("destination", destination); params.put("destination", destination);
return getSqlMapClientTemplate().delete("meta_history.deleteByName", params); return getSqlSession().getMapper(MetaHistoryMapper.class).deleteByName(params);
} }
/** /**
@@ -40,7 +40,7 @@ public class MetaHistoryDAO extends MetaBaseDAO {
long timestamp = System.currentTimeMillis() - interval * 1000; long timestamp = System.currentTimeMillis() - interval * 1000;
params.put("timestamp", timestamp); params.put("timestamp", timestamp);
params.put("destination", destination); params.put("destination", destination);
return getSqlMapClientTemplate().delete("meta_history.deleteByTimestamp", params); return getSqlSession().getMapper(MetaHistoryMapper.class).deleteByTimestamp(params);
} }
protected void initDao() throws Exception { protected void initDao() throws Exception {

15
parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/tsdb/dao/MetaHistoryMapper.java Normal file
View File

@@ -0,0 +1,15 @@
package com.alibaba.otter.canal.parse.inbound.mysql.tsdb.dao;
import java.util.List;
import java.util.Map;
public interface MetaHistoryMapper {
List<MetaHistoryDO> findByTimestamp(Map<String, Object> map);
long insert(MetaHistoryDO metaHistoryDO);
int deleteByName(Map<String, Object> map);
int deleteByTimestamp(Map<String, Object> map);
}

11
parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/tsdb/dao/MetaSnapshotDAO.java
View File

@@ -14,25 +14,24 @@ import com.google.common.collect.Maps;
public class MetaSnapshotDAO extends MetaBaseDAO { public class MetaSnapshotDAO extends MetaBaseDAO {
public Long insert(MetaSnapshotDO snapshotDO) { public Long insert(MetaSnapshotDO snapshotDO) {
return (Long) getSqlMapClientTemplate().insert("meta_snapshot.insert", snapshotDO); return getSqlSession().getMapper(MetaSnapshotMapper.class).insert(snapshotDO);
} }
public Long update(MetaSnapshotDO snapshotDO) { public Long update(MetaSnapshotDO snapshotDO) {
return (Long) getSqlMapClientTemplate().insert("meta_snapshot.update", snapshotDO); return getSqlSession().getMapper(MetaSnapshotMapper.class).update(snapshotDO);
} }
public MetaSnapshotDO findByTimestamp(String destination, Long timestamp) { public MetaSnapshotDO findByTimestamp(String destination, Long timestamp) {
HashMap params = Maps.newHashMapWithExpectedSize(2); HashMap params = Maps.newHashMapWithExpectedSize(2);
params.put("timestamp", timestamp == null ? 0L : timestamp); params.put("timestamp", timestamp == null ? 0L : timestamp);
params.put("destination", destination); params.put("destination", destination);
return getSqlSession().getMapper(MetaSnapshotMapper.class).findByTimestamp(params);
return (MetaSnapshotDO) getSqlMapClientTemplate().queryForObject("meta_snapshot.findByTimestamp", params);
} }
public Integer deleteByName(String destination) { public Integer deleteByName(String destination) {
HashMap params = Maps.newHashMapWithExpectedSize(2); HashMap params = Maps.newHashMapWithExpectedSize(2);
params.put("destination", destination); params.put("destination", destination);
return getSqlMapClientTemplate().delete("meta_snapshot.deleteByName", params); return getSqlSession().getMapper(MetaSnapshotMapper.class).deleteByName(params);
} }
/** /**
@@ -43,7 +42,7 @@ public class MetaSnapshotDAO extends MetaBaseDAO {
long timestamp = System.currentTimeMillis() - interval * 1000; long timestamp = System.currentTimeMillis() - interval * 1000;
params.put("timestamp", timestamp); params.put("timestamp", timestamp);
params.put("destination", destination); params.put("destination", destination);
return getSqlMapClientTemplate().delete("meta_snapshot.deleteByTimestamp", params); return getSqlSession().getMapper(MetaSnapshotMapper.class).deleteByTimestamp(params);
} }
protected void initDao() throws Exception { protected void initDao() throws Exception {

16
parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/tsdb/dao/MetaSnapshotMapper.java Normal file
View File

@@ -0,0 +1,16 @@
package com.alibaba.otter.canal.parse.inbound.mysql.tsdb.dao;
import java.util.Map;
public interface MetaSnapshotMapper {
MetaSnapshotDO findByTimestamp(Map<String, Object> map);
long insert(MetaSnapshotDO metaSnapshotDO);
long update(MetaSnapshotDO metaSnapshotDO);
int deleteByName(Map<String, Object> map);
int deleteByTimestamp(Map<String, Object> map);
}

6
parse/src/test/resources/tsdb/derby-tsdb.xml
View File

@@ -29,16 +29,16 @@
<property name="useUnfairLock" value="true" /> <property name="useUnfairLock" value="true" />
</bean> </bean>
<bean id="sqlMapClient" class="org.springframework.orm.ibatis.SqlMapClientFactoryBean"> <bean id="sqlSessionFactory" class="org.mybatis.spring.SqlSessionFactoryBean">
<property name="dataSource" ref="dataSource"/> <property name="dataSource" ref="dataSource"/>
<property name="configLocation" value="classpath:tsdb/sql-map/sqlmap-config.xml"/> <property name="configLocation" value="classpath:tsdb/sql-map/sqlmap-config.xml"/>
</bean> </bean>
<bean id="metaHistoryDAO" class="com.alibaba.otter.canal.parse.inbound.mysql.tsdb.dao.MetaHistoryDAO"> <bean id="metaHistoryDAO" class="com.alibaba.otter.canal.parse.inbound.mysql.tsdb.dao.MetaHistoryDAO">
<property name="sqlMapClient" ref="sqlMapClient"/> <property name="sqlSessionFactory" ref="sqlSessionFactory"/>
</bean> </bean>
<bean id="metaSnapshotDAO" class="com.alibaba.otter.canal.parse.inbound.mysql.tsdb.dao.MetaSnapshotDAO"> <bean id="metaSnapshotDAO" class="com.alibaba.otter.canal.parse.inbound.mysql.tsdb.dao.MetaSnapshotDAO">
<property name="sqlMapClient" ref="sqlMapClient"/> <property name="sqlSessionFactory" ref="sqlSessionFactory"/>
</bean> </bean>
</beans> </beans>

6
parse/src/test/resources/tsdb/h2-tsdb.xml
View File

@@ -29,16 +29,16 @@
<property name="useUnfairLock" value="true" /> <property name="useUnfairLock" value="true" />
</bean> </bean>
<bean id="sqlMapClient" class="org.springframework.orm.ibatis.SqlMapClientFactoryBean"> <bean id="sqlSessionFactory" class="org.mybatis.spring.SqlSessionFactoryBean">
<property name="dataSource" ref="dataSource"/> <property name="dataSource" ref="dataSource"/>
<property name="configLocation" value="classpath:tsdb/sql-map/sqlmap-config.xml"/> <property name="configLocation" value="classpath:tsdb/sql-map/sqlmap-config.xml"/>
</bean> </bean>
<bean id="metaHistoryDAO" class="com.alibaba.otter.canal.parse.inbound.mysql.tsdb.dao.MetaHistoryDAO"> <bean id="metaHistoryDAO" class="com.alibaba.otter.canal.parse.inbound.mysql.tsdb.dao.MetaHistoryDAO">
<property name="sqlMapClient" ref="sqlMapClient"/> <property name="sqlSessionFactory" ref="sqlSessionFactory"/>
</bean> </bean>
<bean id="metaSnapshotDAO" class="com.alibaba.otter.canal.parse.inbound.mysql.tsdb.dao.MetaSnapshotDAO"> <bean id="metaSnapshotDAO" class="com.alibaba.otter.canal.parse.inbound.mysql.tsdb.dao.MetaSnapshotDAO">
<property name="sqlMapClient" ref="sqlMapClient"/> <property name="sqlSessionFactory" ref="sqlSessionFactory"/>
</bean> </bean>
</beans> </beans>

6
parse/src/test/resources/tsdb/mysql-tsdb.xml
View File

@@ -32,16 +32,16 @@
<property name="useUnfairLock" value="true" /> <property name="useUnfairLock" value="true" />
</bean> </bean>
<bean id="sqlMapClient" class="org.springframework.orm.ibatis.SqlMapClientFactoryBean"> <bean id="sqlSessionFactory" class="org.mybatis.spring.SqlSessionFactoryBean">
<property name="dataSource" ref="dataSource"/> <property name="dataSource" ref="dataSource"/>
<property name="configLocation" value="classpath:tsdb/sql-map/sqlmap-config.xml"/> <property name="configLocation" value="classpath:tsdb/sql-map/sqlmap-config.xml"/>
</bean> </bean>
<bean id="metaHistoryDAO" class="com.alibaba.otter.canal.parse.inbound.mysql.tsdb.dao.MetaHistoryDAO"> <bean id="metaHistoryDAO" class="com.alibaba.otter.canal.parse.inbound.mysql.tsdb.dao.MetaHistoryDAO">
<property name="sqlMapClient" ref="sqlMapClient"/> <property name="sqlSessionFactory" ref="sqlSessionFactory"/>
</bean> </bean>
<bean id="metaSnapshotDAO" class="com.alibaba.otter.canal.parse.inbound.mysql.tsdb.dao.MetaSnapshotDAO"> <bean id="metaSnapshotDAO" class="com.alibaba.otter.canal.parse.inbound.mysql.tsdb.dao.MetaSnapshotDAO">
<property name="sqlMapClient" ref="sqlMapClient"/> <property name="sqlSessionFactory" ref="sqlSessionFactory"/>
</bean> </bean>
</beans> </beans>

20
parse/src/test/resources/tsdb/sql-map/sqlmap-config.xml
View File

@@ -1,8 +1,14 @@
<?xml version="1.0" encoding="UTF-8" ?> <?xml version="1.0" encoding="UTF-8" ?>
<!DOCTYPE sqlMapConfig PUBLIC "-//iBATIS.com//DTD SQL Map Config 2.0//EN" <!DOCTYPE configuration PUBLIC "-//mybatis.org//DTD Config 3.0//EN"
"http://www.ibatis.com/dtd/sql-map-config-2.dtd"> "http://mybatis.org/dtd/mybatis-3-config.dtd">
<sqlMapConfig> <configuration>
<settings useStatementNamespaces="true"/> <typeAliases>
<sqlMap resource="tsdb/sql-map/sqlmap_history.xml"/> <typeAlias alias="metaHistoryDO" type="com.alibaba.otter.canal.parse.inbound.mysql.tsdb.dao.MetaHistoryDO"/>
<sqlMap resource="tsdb/sql-map/sqlmap_snapshot.xml"/> <typeAlias alias="metaSnapshotDO" type="com.alibaba.otter.canal.parse.inbound.mysql.tsdb.dao.MetaSnapshotDO"/>
</sqlMapConfig> </typeAliases>
<mappers>
<mapper resource="tsdb/sql-map/sqlmap_history.xml"/>
<mapper resource="tsdb/sql-map/sqlmap_snapshot.xml"/>
</mappers>
</configuration>

23
parse/src/test/resources/tsdb/sql-map/sqlmap_history.xml
View File

@@ -1,7 +1,6 @@
<?xml version="1.0" encoding="UTF-8" ?> <?xml version="1.0" encoding="UTF-8" ?>
<!DOCTYPE sqlMap PUBLIC "-//ibatis.apache.org//DTD SQL Map 2.0//EN" "http://ibatis.apache.org/dtd/sql-map-2.dtd" > <!DOCTYPE mapper PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN" "http://mybatis.org/dtd/mybatis-3-mapper.dtd">
<sqlMap namespace="meta_history"> <mapper namespace="com.alibaba.otter.canal.parse.inbound.mysql.tsdb.dao.MetaHistoryMapper">
<typeAlias alias="metaHistoryDO" type="com.alibaba.otter.canal.parse.inbound.mysql.tsdb.dao.MetaHistoryDO"/>
<sql id="allColumns"> <sql id="allColumns">
<![CDATA[ <![CDATA[
gmt_create,gmt_modified,destination,binlog_file,binlog_offest,binlog_master_id,binlog_timestamp,use_schema,sql_schema,sql_table,sql_text,sql_type,extra gmt_create,gmt_modified,destination,binlog_file,binlog_offest,binlog_master_id,binlog_timestamp,use_schema,sql_schema,sql_table,sql_text,sql_type,extra
@@ -15,31 +14,31 @@
]]> ]]>
</sql> </sql>
<select id="findByTimestamp" parameterClass="java.util.Map" resultClass="metaHistoryDO"> <select id="findByTimestamp" parameterType="java.util.Map" resultType="metaHistoryDO">
select select
<include refid="allVOColumns"/> <include refid="allVOColumns"/>
from meta_history a from meta_history a
<![CDATA[ <![CDATA[
where destination = #destination# and binlog_timestamp >= #snapshotTimestamp# and binlog_timestamp <= #timestamp# where destination = #{destination} and binlog_timestamp >= #{snapshotTimestamp} and binlog_timestamp <= #{timestamp}
order by binlog_timestamp asc,id asc order by binlog_timestamp asc,id asc
]]> ]]>
</select> </select>
<insert id="insert" parameterClass="metaHistoryDO"> <insert id="insert" parameterType="metaHistoryDO">
insert into meta_history (<include refid="allColumns"/>) insert into meta_history (<include refid="allColumns"/>)
values(CURRENT_TIMESTAMP,CURRENT_TIMESTAMP,#destination#,#binlogFile#,#binlogOffest#,#binlogMasterId#,#binlogTimestamp#,#useSchema#,#sqlSchema#,#sqlTable#,#sqlText#,#sqlType#,#extra#) values(CURRENT_TIMESTAMP,CURRENT_TIMESTAMP,#{destination},#{binlogFile},#{binlogOffest},#{binlogMasterId},#{binlogTimestamp},#{useSchema},#{sqlSchema},#{sqlTable},#{sqlText},#{sqlType},#{extra})
</insert> </insert>
<delete id="deleteByName" parameterClass="java.util.Map"> <delete id="deleteByName" parameterType="java.util.Map">
delete from meta_history delete from meta_history
where destination=#destination# where destination=#{destination}
</delete> </delete>
<delete id="deleteByTimestamp" parameterClass="java.util.Map"> <delete id="deleteByTimestamp" parameterType="java.util.Map">
<![CDATA[ <![CDATA[
delete from meta_history delete from meta_history
where destination=#destination# and binlog_timestamp < #timestamp# where destination=#{destination} and binlog_timestamp < #{timestamp}
]]> ]]>
</delete> </delete>
</sqlMap> </mapper>

43
parse/src/test/resources/tsdb/sql-map/sqlmap_snapshot.xml
View File

@@ -1,9 +1,6 @@
<?xml version="1.0" encoding="UTF-8" ?> <?xml version="1.0" encoding="UTF-8" ?>
<!DOCTYPE sqlMap PUBLIC "-//ibatis.apache.org//DTD SQL Map 2.0//EN" "http://ibatis.apache.org/dtd/sql-map-2.dtd" > <!DOCTYPE mapper PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN" "http://mybatis.org/dtd/mybatis-3-mapper.dtd">
<sqlMap namespace="meta_snapshot"> <mapper namespace="com.alibaba.otter.canal.parse.inbound.mysql.tsdb.dao.MetaSnapshotMapper">
<typeAlias alias="metaSnapshotDO" type="com.alibaba.otter.canal.parse.inbound.mysql.tsdb.dao.MetaSnapshotDO"/>
<typeAlias alias="tableMetaSnapshotDO"
type="com.alibaba.middleware.jingwei.biz.dataobject.CanalTableMetaSnapshotDO"/>
<sql id="allColumns"> <sql id="allColumns">
<![CDATA[ <![CDATA[
gmt_create,gmt_modified,destination,binlog_file,binlog_offest,binlog_master_id,binlog_timestamp,data,extra gmt_create,gmt_modified,destination,binlog_file,binlog_offest,binlog_master_id,binlog_timestamp,data,extra
@@ -16,48 +13,36 @@
]]> ]]>
</sql> </sql>
<select id="findByTimestamp" parameterClass="java.util.Map" resultClass="metaSnapshotDO"> <select id="findByTimestamp" parameterType="java.util.Map" resultType="metaSnapshotDO">
select <include refid="allVOColumns"/> select <include refid="allVOColumns"/>
<![CDATA[ <![CDATA[
from meta_snapshot a from meta_snapshot a
where destination = #destination# and binlog_timestamp < #timestamp# where destination = #{destination} and binlog_timestamp < #{timestamp}
order by binlog_timestamp desc,id desc order by binlog_timestamp desc,id desc
limit 1 limit 1
]]> ]]>
</select> </select>
<select id="findByTimestampOnDerby" parameterClass="java.util.Map" resultClass="metaSnapshotDO"> <insert id="insert" parameterType="metaSnapshotDO">
select * FROM (
select ROW_NUMBER() OVER() AS rownum, <include refid="allVOColumns"/>
<![CDATA[
from meta_snapshot a
where destination = #destination# and binlog_timestamp < #timestamp#
order by binlog_timestamp desc,id desc
) AS tmp
WHERE rownum <= 5
]]>
</select>
<insert id="insert" parameterClass="metaSnapshotDO">
insert into meta_snapshot (<include refid="allColumns"/>) insert into meta_snapshot (<include refid="allColumns"/>)
values(CURRENT_TIMESTAMP,CURRENT_TIMESTAMP,#destination#,#binlogFile#,#binlogOffest#,#binlogMasterId#,#binlogTimestamp#,#data#,#extra#) values(CURRENT_TIMESTAMP,CURRENT_TIMESTAMP,#{destination},#{binlogFile},#{binlogOffest},#{binlogMasterId},#{binlogTimestamp},#{data},#{extra})
</insert> </insert>
<update id="update" parameterClass="metaSnapshotDO"> <update id="update" parameterType="metaSnapshotDO">
update meta_snapshot set gmt_modified=now(), update meta_snapshot set gmt_modified=now(),
binlog_file=#binlogFile#,binlog_offest=#binlogOffest#,binlog_master_id=#binlogMasterId#,binlog_timestamp=#binlogTimestamp#,data=#data#,extra=#extra# binlog_file=#{binlogFile},binlog_offest=#{binlogOffest},binlog_master_id=#{binlogMasterId},binlog_timestamp=#{binlogTimestamp},data=#{data},extra=#{extra}
where destination=#destination# and binlog_timestamp=0 where destination=#{destination} and binlog_timestamp=0
</update> </update>
<delete id="deleteByName" parameterClass="java.util.Map"> <delete id="deleteByName" parameterType="java.util.Map">
delete from meta_snapshot delete from meta_snapshot
where destination=#destination# where destination=#{destination}
</delete> </delete>
<delete id="deleteByTimestamp" parameterClass="java.util.Map"> <delete id="deleteByTimestamp" parameterType="java.util.Map">
<![CDATA[ <![CDATA[
delete from meta_snapshot delete from meta_snapshot
where destination=#destination# and binlog_timestamp < #timestamp# and binlog_timestamp > 0 where destination=#{destination} and binlog_timestamp < #{timestamp} and binlog_timestamp > 0
]]> ]]>
</delete> </delete>
</sqlMap> </mapper>

15
pom.xml
View File

@@ -48,7 +48,7 @@
<repositories> <repositories>
<repository> <repository>
<id>central</id> <id>central</id>
<url>http://repo1.maven.org/maven2</url> <url>https://repo1.maven.org/maven2</url>
<releases> <releases>
<enabled>true</enabled> <enabled>true</enabled>
</releases> </releases>
@@ -99,7 +99,7 @@
<java_target_version>1.8</java_target_version> <java_target_version>1.8</java_target_version>
<file_encoding>UTF-8</file_encoding> <file_encoding>UTF-8</file_encoding>
<javadoc_skip>true</javadoc_skip> <javadoc_skip>true</javadoc_skip>
<spring_version>3.2.18.RELEASE</spring_version> <spring_version>5.0.5.RELEASE</spring_version>
<maven-jacoco-plugin.version>0.8.3</maven-jacoco-plugin.version> <maven-jacoco-plugin.version>0.8.3</maven-jacoco-plugin.version>
<maven-surefire.version>2.22.1</maven-surefire.version> <maven-surefire.version>2.22.1</maven-surefire.version>
<argline>-server -Xms512m -Xmx1024m -Dfile.encoding=UTF-8 <argline>-server -Xms512m -Xmx1024m -Dfile.encoding=UTF-8
@@ -253,9 +253,14 @@
<version>3.6.1</version> <version>3.6.1</version>
</dependency> </dependency>
<dependency> <dependency>
<groupId>org.apache.ibatis</groupId> <groupId>org.mybatis</groupId>
<artifactId>ibatis-sqlmap</artifactId> <artifactId>mybatis-spring</artifactId>
<version>2.3.4.726</version> <version>2.0.4</version>
</dependency>
<dependency>
<groupId>org.mybatis</groupId>
<artifactId>mybatis</artifactId>
<version>3.5.4</version>
</dependency> </dependency>
<dependency> <dependency>
<groupId>com.alibaba</groupId> <groupId>com.alibaba</groupId>