tencent cloud

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("AKIDPko5fVtvTDE0WffffkCwd4NzKcdePt79uauy");
        context.setSecretKey("ECtY8F5e2QqtdXAe18yX0EBqK");
        // 订阅通道所在 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();
    }

}