This commit is contained in:
rewerma
2019-07-11 01:05:11 +08:00
parent a93af74539
commit 64ff140304
5 changed files with 37 additions and 24 deletions

View File

@@ -0,0 +1,15 @@
package com.alibaba.otter.canal.admin.config;
import org.apache.commons.dbutils.QueryRunner;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import javax.sql.DataSource;
@Configuration
public class Config {
@Bean
public QueryRunner queryRunner(DataSource dataSource) {
return new QueryRunner(dataSource);
}
}

View File

@@ -1,43 +1,40 @@
package com.alibaba.otter.canal.admin.dao; package com.alibaba.otter.canal.admin.dao;
import com.alibaba.otter.canal.admin.model.CanalConfig; import com.alibaba.otter.canal.admin.model.CanalConfig;
import com.alibaba.otter.canal.admin.model.Person;
import org.apache.commons.dbutils.QueryRunner; import org.apache.commons.dbutils.QueryRunner;
import org.apache.commons.dbutils.handlers.BeanHandler; import org.apache.commons.dbutils.handlers.BeanHandler;
import org.javalite.activejdbc.Base;
import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.stereotype.Repository; import org.springframework.stereotype.Repository;
import javax.annotation.PostConstruct; import javax.annotation.PostConstruct;
import javax.sql.DataSource;
import java.sql.SQLException; import java.sql.SQLException;
import java.util.Date;
@Repository @Repository
public class CanalConfigDao { public class CanalConfigDao {
@Autowired @Autowired
private DataSource dataSource; private QueryRunner runner;
// public void open() {
// Base.open(dataSource);
// }
@PostConstruct @PostConstruct
public void init() { public void init() {
// CanalConfig canalConfig = new CanalConfig();
// canalConfig.setId(1L);
UpdateContent(1L);
}
public void UpdateContent(Long id) {
QueryRunner runner = new QueryRunner(dataSource);
try { try {
CanalConfig canalConfig = runner.query("select * from canal_config where id=1", new BeanHandler<>(CanalConfig.class)); CanalConfig canalConfig = findById(1L);
canalConfig = canalConfig; canalConfig = canalConfig;
canalConfig.setContent(canalConfig.getContent()+" xxx");
updateContent(canalConfig);
} catch (SQLException e) { } catch (SQLException e) {
e.printStackTrace(); e.printStackTrace();
} }
// Base.open(dataSource); }
// Person.where("id=?", id);
// Base.close(); public CanalConfig findById(Long id) throws SQLException {
String sql = "select id,name,content,modified_time as modifiedTime" +
" from canal_config where id=?";
return runner.query(sql, new BeanHandler<>(CanalConfig.class), id);
}
public void updateContent(CanalConfig canalConfig) throws SQLException {
String sql = "update canal_config set content=?, modified_time=? where id=?";
runner.update(sql, canalConfig.getContent(), new Date(), canalConfig.getId());
} }
} }

View File

@@ -10,7 +10,7 @@ public class CanalConfig {
private Long id; private Long id;
private String name; private String name;
private String content; private String content;
private Timestamp modifiedTime; private Date modifiedTime;
public Long getId() { public Long getId() {
return id; return id;
@@ -36,11 +36,11 @@ public class CanalConfig {
this.content = content; this.content = content;
} }
public Timestamp getModifiedTime() { public Date getModifiedTime() {
return modifiedTime; return modifiedTime;
} }
public void setModifiedTime(Timestamp modifiedTime) { public void setModifiedTime(Date modifiedTime) {
this.modifiedTime = modifiedTime; this.modifiedTime = modifiedTime;
} }
} }

View File

@@ -126,6 +126,7 @@
<module>example</module> <module>example</module>
<module>prometheus</module> <module>prometheus</module>
<module>client-adapter</module> <module>client-adapter</module>
<module>canal-admin</module>
</modules> </modules>
<dependencyManagement> <dependencyManagement>
@@ -253,7 +254,7 @@
<dependency> <dependency>
<groupId>com.alibaba.fastsql</groupId> <groupId>com.alibaba.fastsql</groupId>
<artifactId>fastsql</artifactId> <artifactId>fastsql</artifactId>
<version>2.0.0_preview_914</version> <version>2.0.0_preview_896</version>
</dependency> </dependency>
<dependency> <dependency>
<groupId>com.alibaba</groupId> <groupId>com.alibaba</groupId>

View File

@@ -133,7 +133,7 @@ public class CanalKafkaProducer implements CanalMQProducer {
canalDestination.getDynamicTopic()); canalDestination.getDynamicTopic());
for (Map.Entry<String, Message> entry : messageMap.entrySet()) { for (Map.Entry<String, Message> entry : messageMap.entrySet()) {
String topicName = entry.getKey().replace('.', '_'); String topicName = entry.getKey(); //.replace('.', '_');
Message messageSub = entry.getValue(); Message messageSub = entry.getValue();
if (logger.isDebugEnabled()) { if (logger.isDebugEnabled()) {
logger.debug("## Send message to kafka topic: " + topicName); logger.debug("## Send message to kafka topic: " + topicName);