数据订阅库表到 Kafka

yum install java-1.8.0-openjdk-devel 

bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic testtop
Created topic "testtop".

import com.qcloud.dts.context.SubscribeContext;
import com.qcloud.dts.message.ClusterMessage;
import com.qcloud.dts.message.DataMessage;
import com.qcloud.dts.subscribe.ClusterListener;
import com.qcloud.dts.subscribe.DefaultSubscribeClient;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.StringSerializer;
import org.apache.log4j.Logger;

import java.util.List;
import java.util.Properties;

public class KafkaDemo {
 public static void main(String[] args) throws Exception {
     //初始化一个 kafka producer
     final String TOPIC = "testtop";
     Properties props = new Properties();
     props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "10.168.1.6:9092");
     props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
     props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);

     final Producer<String, String> producer = new KafkaProducer<String, String>(props);

     // 创建一个 context
     SubscribeContext context = new SubscribeContext();
     context.setSecretId("AKIDPko5fVtxxxxxdePt79uauy");
     context.setSecretKey("ECtY8F5exxxxx0EBqK");
     // 订阅通道所在 region
     context.setRegion("ap-beijing");
     final DefaultSubscribeClient client = new DefaultSubscribeClient(context);

     // 创建订阅监听者 listener
     ClusterListener listener = new ClusterListener() {
         @Override
         public void notify(List<ClusterMessage> messages) throws Exception {
             System.out.println("--------------------:" + messages.size());
             for(ClusterMessage m:messages){
                 DataMessage.Record record = m.getRecord();


                 if(record.getOpt() != DataMessage.Record.Type.BEGIN && record.getOpt() != DataMessage.Record.Type.COMMIT){
                     List<DataMessage.Record.Field> fields = record.getFieldList();

                     //打印每个列的信息
                     for (int i = 0; i < fields.size(); i++) {
                         DataMessage.Record.Field field = fields.get(i);
                         System.out.println("Database Name:" + record.getDbName());
                         System.out.println("Table Name:" + record.getTablename());
                         System.out.println("Field Value:" + field.getValue());
                         System.out.println("Field Value:" + field.getValue().length());
                         System.out.println("Field Encoding:" + field.getFieldEnc());
                     }

                     //将整个 record 发送到指定的 kafka topic 中
                     System.out.println("Record+++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++");
                     producer.send(new ProducerRecord<String, String>(TOPIC, record.toString()));
                 }

                 m.ackAsConsumed();
             }
         }
         @Override
         public void onException(Exception e){
             System.out.println("listen exception" + e);
         }
     };
     // 添加监听者
     client.addClusterListener(listener);
     client.askForGUID("dts-channel-p15e9eW9rn8hA68K");
     client.start();
 }

}

javac -classpath binlogsdk-2.9.1-jar-with-dependencies.jar:slf4j-api-1.7.25.jar:slf4j-log4j12-1.7.2.jar:kafka-clients-1.1.0.jar -encoding UTF-8 KafkaDemo.java 

java -XX:-UseGCOverheadLimit -Xms2g -Xmx2g  -classpath .:binlogsdk-2.9.1-jar-with-dependencies.jar:kafka-clients-1.1.0.jar:slf4j-api-1.7.25.jar:slf4j-log4j12-1.7.2.jar  KafkaDemo

MySQL [test]> insert into alantest values(123456,'alan');
Query OK, 1 row affected (0.02 sec)
[root@VM_71_10_centos kafka_2.11-1.1.0]#  bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic testtop --from-beginning 
checkpoint:144251@3@1275254@1153089
record_id:00000100000000001198410000000000000001
record_encoding:utf8
fields_enc:latin1,utf8
gtid:4f21864b-3bed-11e8-a44c-5cb901896188:5552
source_category:full_recorded
source_type:mysql
table_name:alantest
record_type:INSERT
db:test
timestamp:1524649133
primary:id
Field name: id
Field type: 3
Field length: 6
Field value: 123456
Field name: name
Field type: 253
Field length: 4
Field value: alan