加一个表

src/main/java/com/huazheng/project/greenplum/mapper/GpAbc2Mapper.java

+package com.huazheng.project.greenplum.mapper;
+import com.baomidou.mybatisplus.core.mapper.BaseMapper;
+import com.huazheng.project.greenplum.model.GpAbc2;
+
+public interface GpAbc2Mapper extends BaseMapper<GpAbc2> {
+	
+}
\ No newline at end of file

src/main/java/com/huazheng/project/greenplum/mapper/GpTb2Mapper.java

+package com.huazheng.project.greenplum.mapper;
+import com.baomidou.mybatisplus.core.mapper.BaseMapper;
+import com.huazheng.project.greenplum.model.GpTb2;
+
+public interface GpTb2Mapper extends BaseMapper<GpTb2> {
+	
+}
\ No newline at end of file

src/main/java/com/huazheng/project/greenplum/mapper/GpTest2Mapper.java

+package com.huazheng.project.greenplum.mapper;
+import com.baomidou.mybatisplus.core.mapper.BaseMapper;
+import com.huazheng.project.greenplum.model.GpTest2;
+
+public interface GpTest2Mapper extends BaseMapper<GpTest2> {
+	
+}
\ No newline at end of file

src/main/java/com/huazheng/project/greenplum/model/GpAbc2.java

+package com.huazheng.project.greenplum.model;
+
+import java.io.Serializable;
+
+import com.baomidou.mybatisplus.annotation.TableField;
+import com.baomidou.mybatisplus.annotation.TableId;
+import com.baomidou.mybatisplus.annotation.TableName;
+import com.baomidou.mybatisplus.extension.activerecord.Model;
+
+import io.swagger.annotations.ApiModel;
+import io.swagger.annotations.ApiModelProperty;
+import lombok.AllArgsConstructor;
+import lombok.Builder;
+import lombok.Data;
+import lombok.EqualsAndHashCode;
+import lombok.NoArgsConstructor;
+import lombok.experimental.Accessors;
+
+@Data
+@Builder
+@EqualsAndHashCode(callSuper = false)
+@Accessors(chain = true)
+@TableName("mssql_abc2")
+@NoArgsConstructor
+@AllArgsConstructor
+@ApiModel(value="GpAbc2对象", description="测试表")
+public class GpAbc2 extends Model<GpAbc2> {
+	
+	private static final long serialVersionUID = 1L;
+	
+	@ApiModelProperty(value = "主键")
+    @TableId("id")
+	private Integer id;
+	
+	@ApiModelProperty(value = "用户")
+	@TableField("username")
+	private String username;
+	
+	@ApiModelProperty(value = "密码")
+	@TableField("userpswd")
+	private String userpswd;
+	
+	@Override
+    protected Serializable pkVal() {
+        return this.id;
+    }
+	
+}

src/main/java/com/huazheng/project/greenplum/model/GpTb2.java

+package com.huazheng.project.greenplum.model;
+
+import java.io.Serializable;
+
+import com.baomidou.mybatisplus.annotation.TableField;
+import com.baomidou.mybatisplus.annotation.TableId;
+import com.baomidou.mybatisplus.annotation.TableName;
+import com.baomidou.mybatisplus.extension.activerecord.Model;
+
+import io.swagger.annotations.ApiModel;
+import io.swagger.annotations.ApiModelProperty;
+import lombok.AllArgsConstructor;
+import lombok.Builder;
+import lombok.Data;
+import lombok.EqualsAndHashCode;
+import lombok.NoArgsConstructor;
+import lombok.experimental.Accessors;
+
+@Data
+@Builder
+@EqualsAndHashCode(callSuper = false)
+@Accessors(chain = true)
+@TableName("mysql_tb2")
+@NoArgsConstructor
+@AllArgsConstructor
+@ApiModel(value="GpTb2对象", description="测试表")
+public class GpTb2 extends Model<GpTb2> {
+	
+	private static final long serialVersionUID = 1L;
+	
+	@ApiModelProperty(value = "主键")
+    @TableId("id")
+	private Integer id;
+	
+	@ApiModelProperty(value = "用户")
+	@TableField("username")
+	private String username;
+	
+	@ApiModelProperty(value = "密码")
+	@TableField("userpswd")
+	private String userpswd;
+	
+	@Override
+    protected Serializable pkVal() {
+        return this.id;
+    }
+	
+}

src/main/java/com/huazheng/project/greenplum/model/GpTest2.java

+package com.huazheng.project.greenplum.model;
+
+import java.io.Serializable;
+import java.util.Date;
+
+import com.baomidou.mybatisplus.annotation.TableField;
+import com.baomidou.mybatisplus.annotation.TableId;
+import com.baomidou.mybatisplus.annotation.TableName;
+import com.baomidou.mybatisplus.extension.activerecord.Model;
+
+import io.swagger.annotations.ApiModel;
+import io.swagger.annotations.ApiModelProperty;
+import lombok.AllArgsConstructor;
+import lombok.Builder;
+import lombok.Data;
+import lombok.EqualsAndHashCode;
+import lombok.NoArgsConstructor;
+import lombok.experimental.Accessors;
+
+@Data
+@Builder
+@EqualsAndHashCode(callSuper = false)
+@Accessors(chain = true)
+@TableName("hana_test2")
+@NoArgsConstructor
+@AllArgsConstructor
+@ApiModel(value="GpTest2对象", description="测试表")
+public class GpTest2 extends Model<GpTest2> {
+	
+	private static final long serialVersionUID = 1L;
+	
+	@ApiModelProperty(value = "主键")
+    @TableId("id")
+	private Integer id;
+	
+	@ApiModelProperty(value = "用户")
+	@TableField("\"user\"")
+	private String user;
+	
+	@ApiModelProperty(value = "密码")
+	@TableField("pswd")
+	private String pswd;
+	
+	@ApiModelProperty(value = "时间")
+	@TableField("times")
+	private Date times;
+	
+	@Override
+    protected Serializable pkVal() {
+        return this.id;
+    }
+	
+}

src/main/java/com/huazheng/project/service/impl/GpAbc2ServiceImpl.java

+package com.huazheng.project.service.impl;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.kafka.annotation.KafkaListener;
+import org.springframework.stereotype.Service;
+
+import com.huazheng.project.greenplum.mapper.GpAbc2Mapper;
+import com.huazheng.project.greenplum.model.GpAbc2;
+
+import cn.hutool.json.JSONObject;
+import cn.hutool.json.JSONUtil;
+
+@Service
+public class GpAbc2ServiceImpl {
+
+	private Log log = LogFactory.getLog(getClass());
+
+	@Autowired
+	private GpAbc2Mapper mapper;
+
+	@KafkaListener(topics = "sqlserver.test2.dbo.abc2")
+	public void mssqlData(ConsumerRecord<?, ?> record) {
+		Object value = record.value();
+		if (value != null) {
+			String data = value.toString();
+			JSONObject json = JSONUtil.parseObj(data);
+			JSONObject payload = json.getJSONObject("payload");
+			String op = payload.getStr("op");
+			JSONObject before = payload.getJSONObject("before");
+			JSONObject after = payload.getJSONObject("after");
+			if (op.equals("c")) {
+				GpAbc2 bean = JSONUtil.toBean(after, GpAbc2.class);
+				mapper.insert(bean);
+				log.info("mssql insert --> " + bean);
+			} else if (op.equals("u")) {
+				GpAbc2 bean = JSONUtil.toBean(after, GpAbc2.class);
+				mapper.updateById(bean);
+				log.info("mssql update --> " + bean);
+			} else if (op.equals("d")) {
+				GpAbc2 bean = JSONUtil.toBean(before, GpAbc2.class);
+				mapper.deleteById(bean.getId());
+				log.info("mssql delete --> " + bean);
+			}
+		}
+	}
+	
+}

src/main/java/com/huazheng/project/service/impl/GpTb2ServiceImpl.java

+package com.huazheng.project.service.impl;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.kafka.annotation.KafkaListener;
+import org.springframework.stereotype.Service;
+
+import com.huazheng.project.greenplum.mapper.GpTb2Mapper;
+import com.huazheng.project.greenplum.model.GpTb2;
+
+import cn.hutool.json.JSONObject;
+import cn.hutool.json.JSONUtil;
+
+@Service
+public class GpTb2ServiceImpl {
+
+	private Log log = LogFactory.getLog(getClass());
+
+	@Autowired
+	private GpTb2Mapper mapper;
+	
+	@KafkaListener(topics = "mysql.abc.abc.tb2")
+	public void mysqlData(ConsumerRecord<?, ?> record) {
+		Object value = record.value();
+		if (value != null) {
+			String data = value.toString();
+			JSONObject json = JSONUtil.parseObj(data);
+			JSONObject payload = json.getJSONObject("payload");
+			String op = payload.getStr("op");
+			JSONObject before = payload.getJSONObject("before");
+			JSONObject after = payload.getJSONObject("after");
+			if (op.equals("c")) {
+				GpTb2 bean = JSONUtil.toBean(after, GpTb2.class);
+				mapper.insert(bean);
+				log.info("mysql insert --> " + bean);
+			} else if (op.equals("u")) {
+				GpTb2 bean = JSONUtil.toBean(after, GpTb2.class);
+				mapper.updateById(bean);
+				log.info("mysql update --> " + bean);
+			} else if (op.equals("d")) {
+				GpTb2 bean = JSONUtil.toBean(before, GpTb2.class);
+				mapper.deleteById(bean.getId());
+				log.info("mysql delete --> " + bean);
+			}
+		}
+	}
+	
+}

src/main/java/com/huazheng/project/service/impl/GpTest2ServiceImpl.java

+package com.huazheng.project.service.impl;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.kafka.annotation.KafkaListener;
+import org.springframework.stereotype.Service;
+
+import com.huazheng.project.greenplum.mapper.GpTest2Mapper;
+import com.huazheng.project.greenplum.model.GpTest2;
+
+import cn.hutool.json.JSONObject;
+import cn.hutool.json.JSONUtil;
+
+@Service
+public class GpTest2ServiceImpl {
+
+	private Log log = LogFactory.getLog(getClass());
+
+	@Autowired
+	private GpTest2Mapper mapper;
+	
+	@KafkaListener(topics = "test_topic_3")
+	public void hanaData(ConsumerRecord<?, ?> record) {
+		Object value = record.value();
+		if (value != null) {
+			String data = value.toString();
+			JSONObject json = JSONUtil.parseObj(data);
+			JSONObject payload = json.getJSONObject("payload");
+			String op = payload.getStr("TYPE");
+			JSONObject before = payload.getJSONObject("BEFORE");
+			JSONObject after = payload.getJSONObject("AFTER");
+			if (op.equals("c")) {
+				after.set("times", after.getStr("times").substring(0, 19));
+				GpTest2 bean = JSONUtil.toBean(after, GpTest2.class);
+				mapper.insert(bean);
+				log.info("hana insert --> " + bean);
+			} else if (op.equals("u")) {
+				after.set("times", after.getStr("times").substring(0, 19));
+				GpTest2 bean = JSONUtil.toBean(after, GpTest2.class);
+				mapper.updateById(bean);
+				log.info("hana update --> " + bean);
+			} else if (op.equals("d")) {
+				before.set("times", before.getStr("times").substring(0, 19));
+				GpTest2 bean = JSONUtil.toBean(before, GpTest2.class);
+				mapper.deleteById(bean.getId());
+				log.info("hana delete --> " + bean);
+			}
+		}
+	}
+	
+}