canal 适配器说明

Love The Way You Lie 2022-10-12 12:12 182阅读 0赞

--------------------

## canal 适配器说明 ##

**官网：[https://github.com/alibaba/canal/wiki/ClientAdapter][https_github.com_alibaba_canal_wiki_ClientAdapter]**

**docker 仓库：[https://registry.hub.docker.com/r/slpcat/canal-adapter][https_registry.hub.docker.com_r_slpcat_canal-adapter]**

## \*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\* ##

**适配器说明**

**canal 1.1.1之后，提供了适配器功能，****可将canal server的数据直接输出到目的地，不需要用户编写客户端****（个性化需求还需要用户编写客户端实现）**

**canal adapter分为启动器、适配器，1.1.5提供的适配器如下：**

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MzkzMTYyNQ_size_16_color_FFFFFF_t_70][]

**launcher：**启动器

**logger：**日志适配器

**rdb：**支持jdbc的关系型数据库适配器**（mysql、oracle、postgress、sqlserver等）**

**hbase：**hbase适配器

**kudu：**kudu适配器

## \*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\* ##

**启动器配置文件**

![20210704205930899.png][]

**application.yml**

server:
      port: 8081
    spring:
      jackson:
        date-format: yyyy-MM-dd HH:mm:ss
        time-zone: GMT+8
        default-property-inclusion: non_null
    
    canal.conf:
      mode: tcp #tcp kafka rocketMQ rabbitMQ      #客户端消费模式，对应下面的consumerProperties
      flatMessage: true                           #是否以json字符串传递数据，仅对mq生效
      #canalServerHost:               #canal adapter直连canal server
                                      #canalServerHost在1.1.5后不起作用
                                      #tcp mode相关属性在consumerProperties tcp中设置
                                      #kafka、rocketmq、rabbitmq的属性也在对应的 consumerProperties中设置
    
      zookeeperHosts:                 #canal server集群部署时，创建curator客户端
                                      #tcp mode需要在consumerProperties tcp中设置
      batchSize: 50                               #每次获取的数据大小，单位为 k
      syncBatchSize: 1000                         #每次同步的批数量
      retries: 0                                  #重试次数，-1为无限次
      timeout:
      accessKey:
      secretKey:
      consumerProperties:
        # canal tcp consumer                      #canal adapter连接的canal server
        canal.tcp.server.host: 127.0.0.1:11111
        canal.tcp.zookeeper.hosts:
        canal.tcp.batch.size: 500
        canal.tcp.username:
        canal.tcp.password:
        # kafka consumer                          #canal adapter连接的kafka
        kafka.bootstrap.servers: 127.0.0.1:9092
        kafka.enable.auto.commit: false
        kafka.auto.commit.interval.ms: 1000
        kafka.auto.offset.reset: latest
        kafka.request.timeout.ms: 40000
        kafka.session.timeout.ms: 30000
        kafka.isolation.level: read_committed
        kafka.max.poll.records: 1000
        # rocketMQ consumer                       #canal adapter连接的rocketmq
        rocketmq.namespace:
        rocketmq.namesrv.addr: 127.0.0.1:9876
        rocketmq.batch.size: 1000
        rocketmq.enable.message.trace: false
        rocketmq.customized.trace.topic:
        rocketmq.access.channel:
        rocketmq.subscribe.filter:
        # rabbitMQ consumer                       #canal adapter连接的rabbitmq
        rabbitmq.host:
        rabbitmq.virtual.host:
        rabbitmq.username:
        rabbitmq.password:
        rabbitmq.resource.ownerId:
    
    #  srcDataSources:                 #数据源
    #    defaultDS:
    #      url: jdbc:mysql://127.0.0.1:3306/mytest?useUnicode=true
    #      username: root
    #      password: 121212
      canalAdapters:
      - instance: example # canal instance Name or mq topic name
                          # tcp mode：instance name
                          # mq  mode：topic name
                          # 注意：instance name、topic name不支持通配符匹配
        groups:
        - groupId: g1                  #一份数据可被多个groupId消费
                                       #不同groupId并发执行，
                                       #同一groupId内的adapters顺序执行
          outerAdapters:
          - name: logger               #输出到日志
    #      - name: rdb                 #输出到rdb（关系型数据库）
    #        key: mysql1                     #输出到mysql数据库
    #        properties:
    #          jdbc.driverClassName: com.mysql.jdbc.Driver
    #          jdbc.url: jdbc:mysql://127.0.0.1:3306/mytest2?useUnicode=true
    #          jdbc.username: root
    #          jdbc.password: 121212
    #      - name: rdb
    #        key: oracle1                    #输出到oracle数据库
    #        properties:
    #          jdbc.driverClassName: oracle.jdbc.OracleDriver
    #          jdbc.url: jdbc:oracle:thin:@localhost:49161:XE
    #          jdbc.username: mytest
    #          jdbc.password: m121212
    #      - name: rdb
    #        key: postgres1                  #输出到postgress数据库
    #        properties:
    #          jdbc.driverClassName: org.postgresql.Driver
    #          jdbc.url: jdbc:postgresql://localhost:5432/postgres
    #          jdbc.username: postgres
    #          jdbc.password: 121212
    #          threads: 1
    #          commitSize: 3000
    #      - name: hbase               #输出到hbase
    #        properties:
    #          hbase.zookeeper.quorum: 127.0.0.1
    #          hbase.zookeeper.property.clientPort: 2181
    #          zookeeper.znode.parent: /hbase
    #      - name: es                  #输出到es
    #        hosts: 127.0.0.1:9300 # 127.0.0.1:9200 for rest mode
    #        properties:
    #          mode: transport # or rest
    #          # security.auth: test:123456 #  only used for rest mode
    #          cluster.name: elasticsearch
    #        - name: kudu              #输出到kudu
    #          key: kudu
    #          properties:
    #            kudu.master.address: 127.0.0.1 # ',' split multi address

**tcp 属性**

package com.alibaba.otter.canal.connector.tcp.config;
    
    public class TCPConstants {
    
        public static final String ROOT                 = "canal";
    
        public static final String CANAL_TCP_HOST       = ROOT + "." + "tcp.server.host";
        public static final String CANAL_TCP_ZK_HOSTS   = ROOT + "." + "tcp.zookeeper.hosts";
        public static final String CANAL_TCP_USERNAME   = ROOT + "." + "tcp.username";
        public static final String CANAL_TCP_PASSWORD   = ROOT + "." + "tcp.password";
        public static final String CANAL_TCP_BATCH_SIZE = ROOT + "." + "tcp.batch.size";
    }

**mq 公共属性**

public class CanalConstants {
    
        public static final String ROOT                           = "canal";
    
        public static final String CANAL_FILTER_TRANSACTION_ENTRY = ROOT + "." + "instance.filter.transaction.entry";
    
        public static final String CANAL_MQ_ACCESS_CHANNEL        = ROOT + "." + "mq.accessChannel";
        public static final String CANAL_MQ_CANAL_BATCH_SIZE      = ROOT + "." + "mq.canalBatchSize";
        public static final String CANAL_MQ_CANAL_GET_TIMEOUT     = ROOT + "." + "mq.canalGetTimeout";
        public static final String CANAL_MQ_FLAT_MESSAGE          = ROOT + "." + "mq.flatMessage";
    
        public static final String CANAL_MQ_DATABASE_HASH         = ROOT + "." + "mq.database.hash";
        public static final String CANAL_MQ_BUILD_THREAD_SIZE     = ROOT + "." + "mq.build.thread.size";
        public static final String CANAL_MQ_SEND_THREAD_SIZE      = ROOT + "." + "mq.send.thread.size";
    
        public static final String CANAL_ALIYUN_ACCESS_KEY        = ROOT + "." + "aliyun.accessKey";
        public static final String CANAL_ALIYUN_SECRET_KEY        = ROOT + "." + "aliyun.secretKey";
        public static final String CANAL_ALIYUN_UID               = ROOT + "." + "aliyun.uid";
    
    }

**kafka 属性**

@SPI("kafka")
    public class CanalKafkaConsumer implements CanalMsgConsumer {
    
        private static final String      PREFIX_KAFKA_CONFIG = "kafka.";  //前缀kafka
    
        private KafkaConsumer<String, ?> kafkaConsumer;
        private boolean                  flatMessage         = true;
        private String                   topic;
    
        private Map<Integer, Long>       currentOffsets      = new ConcurrentHashMap<>();
        private Properties               kafkaProperties     = new Properties();
    
        @Override
        public void init(Properties properties, String topic, String groupId) {
            this.topic = topic;
    
            Boolean flatMessage = (Boolean) properties.get(CanalConstants.CANAL_MQ_FLAT_MESSAGE);
            if (flatMessage != null) {
                this.flatMessage = flatMessage;
            }
            for (Map.Entry<Object, Object> entry : properties.entrySet()) {
                String k = (String) entry.getKey();
                Object v = entry.getValue();
                if (k.startsWith(PREFIX_KAFKA_CONFIG) && v != null) {
                    kafkaProperties.put(k.substring(PREFIX_KAFKA_CONFIG.length()), v);
                }   //将前缀为kafka的属性添加到kafkaProperties中
            }
            kafkaProperties.put("group.id", groupId);
            kafkaProperties.put("key.deserializer", StringDeserializer.class);
            kafkaProperties.put("client.id", UUID.randomUUID().toString().substring(0, 6));
        }
    
        。。。
    
    
    *****************
    kafkaConsumer
    
    public class KafkaConsumer<K, V> implements Consumer<K, V> {
        private static final long NO_CURRENT_THREAD = -1L;
        private static final AtomicInteger CONSUMER_CLIENT_ID_SEQUENCE = new AtomicInteger(1);
        private static final String JMX_PREFIX = "kafka.consumer";
        static final long DEFAULT_CLOSE_TIMEOUT_MS = 30000L;
        final Metrics metrics;
        private final Logger log;
        private final String clientId;
        private final ConsumerCoordinator coordinator;
        private final Deserializer<K> keyDeserializer;
        private final Deserializer<V> valueDeserializer;
        private final Fetcher<K, V> fetcher;
        private final ConsumerInterceptors<K, V> interceptors;
        private final Time time;
        private final ConsumerNetworkClient client;
        private final SubscriptionState subscriptions;
        private final Metadata metadata;
        private final long retryBackoffMs;
        private final long requestTimeoutMs;
        private volatile boolean closed;
        private List<PartitionAssignor> assignors;
        private final AtomicLong currentThread;
        private final AtomicInteger refcount;
    
        public KafkaConsumer(Map<String, Object> configs) {
            this((Map)configs, (Deserializer)null, (Deserializer)null);
        }
    
        public KafkaConsumer(Map<String, Object> configs, Deserializer<K> keyDeserializer, Deserializer<V> valueDeserializer) {
            this(new ConsumerConfig(ConsumerConfig.addDeserializerToConfig(configs, keyDeserializer, valueDeserializer)), keyDeserializer, valueDeserializer);
        }
    
        public KafkaConsumer(Properties properties) {
            this((Properties)properties, (Deserializer)null, (Deserializer)null);
        }
    
        public KafkaConsumer(Properties properties, Deserializer<K> keyDeserializer, Deserializer<V> valueDeserializer) {
            this(new ConsumerConfig(ConsumerConfig.addDeserializerToConfig(properties, keyDeserializer, valueDeserializer)), keyDeserializer, valueDeserializer);
        }
    
        private KafkaConsumer(ConsumerConfig config, Deserializer<K> keyDeserializer, Deserializer<V> valueDeserializer) {
            this.closed = false;
            this.currentThread = new AtomicLong(-1L);
            this.refcount = new AtomicInteger(0);
    
            try {
                String clientId = config.getString("client.id");
                if (clientId.isEmpty()) {
                    clientId = "consumer-" + CONSUMER_CLIENT_ID_SEQUENCE.getAndIncrement();
                }
    
                this.clientId = clientId;
                String groupId = config.getString("group.id");
                LogContext logContext = new LogContext("[Consumer clientId=" + clientId + ", groupId=" + groupId + "] ");
                this.log = logContext.logger(this.getClass());
                this.log.debug("Initializing the Kafka consumer");
                this.requestTimeoutMs = (long)config.getInt("request.timeout.ms");
                int sessionTimeOutMs = config.getInt("session.timeout.ms");
                int fetchMaxWaitMs = config.getInt("fetch.max.wait.ms");
                if (this.requestTimeoutMs > (long)sessionTimeOutMs && this.requestTimeoutMs > (long)fetchMaxWaitMs) {
                    this.time = Time.SYSTEM;
                    Map<String, String> metricsTags = Collections.singletonMap("client-id", clientId);
                    MetricConfig metricConfig = (new MetricConfig()).samples(config.getInt("metrics.num.samples")).timeWindow(config.getLong("metrics.sample.window.ms"), TimeUnit.MILLISECONDS).recordLevel(RecordingLevel.forName(config.getString("metrics.recording.level"))).tags(metricsTags);
                    List<MetricsReporter> reporters = config.getConfiguredInstances("metric.reporters", MetricsReporter.class);
                    reporters.add(new JmxReporter("kafka.consumer"));
                    this.metrics = new Metrics(metricConfig, reporters, this.time);
                    this.retryBackoffMs = config.getLong("retry.backoff.ms");
                    Map<String, Object> userProvidedConfigs = config.originals();
                    userProvidedConfigs.put("client.id", clientId);
                    List<ConsumerInterceptor<K, V>> interceptorList = (new ConsumerConfig(userProvidedConfigs, false)).getConfiguredInstances("interceptor.classes", ConsumerInterceptor.class);
                    this.interceptors = new ConsumerInterceptors(interceptorList);
                    if (keyDeserializer == null) {
                        this.keyDeserializer = (Deserializer)config.getConfiguredInstance("key.deserializer", Deserializer.class);
                        this.keyDeserializer.configure(config.originals(), true);
                    } else {
                        config.ignore("key.deserializer");
                        this.keyDeserializer = keyDeserializer;
                    }
    
                    if (valueDeserializer == null) {
                        this.valueDeserializer = (Deserializer)config.getConfiguredInstance("value.deserializer", Deserializer.class);
                        this.valueDeserializer.configure(config.originals(), false);
                    } else {
                        config.ignore("value.deserializer");
                        this.valueDeserializer = valueDeserializer;
                    }
    
                    ClusterResourceListeners clusterResourceListeners = this.configureClusterResourceListeners(keyDeserializer, valueDeserializer, reporters, interceptorList);
                    this.metadata = new Metadata(this.retryBackoffMs, config.getLong("metadata.max.age.ms"), true, false, clusterResourceListeners);
                    List<InetSocketAddress> addresses = ClientUtils.parseAndValidateAddresses(config.getList("bootstrap.servers"));
                    this.metadata.update(Cluster.bootstrap(addresses), Collections.emptySet(), 0L);
                    String metricGrpPrefix = "consumer";
                    ConsumerMetrics metricsRegistry = new ConsumerMetrics(metricsTags.keySet(), "consumer");
                    ChannelBuilder channelBuilder = ClientUtils.createChannelBuilder(config);
                    IsolationLevel isolationLevel = IsolationLevel.valueOf(config.getString("isolation.level").toUpperCase(Locale.ROOT));
                    Sensor throttleTimeSensor = Fetcher.throttleTimeSensor(this.metrics, metricsRegistry.fetcherMetrics);
                    int heartbeatIntervalMs = config.getInt("heartbeat.interval.ms");
                    NetworkClient netClient = new NetworkClient(new Selector(config.getLong("connections.max.idle.ms"), this.metrics, this.time, metricGrpPrefix, channelBuilder, logContext), this.metadata, clientId, 100, config.getLong("reconnect.backoff.ms"), config.getLong("reconnect.backoff.max.ms"), config.getInt("send.buffer.bytes"), config.getInt("receive.buffer.bytes"), config.getInt("request.timeout.ms"), this.time, true, new ApiVersions(), throttleTimeSensor, logContext);
                    this.client = new ConsumerNetworkClient(logContext, netClient, this.metadata, this.time, this.retryBackoffMs, (long)config.getInt("request.timeout.ms"), heartbeatIntervalMs);
                    OffsetResetStrategy offsetResetStrategy = OffsetResetStrategy.valueOf(config.getString("auto.offset.reset").toUpperCase(Locale.ROOT));
                    this.subscriptions = new SubscriptionState(offsetResetStrategy);
                    this.assignors = config.getConfiguredInstances("partition.assignment.strategy", PartitionAssignor.class);
                    this.coordinator = new ConsumerCoordinator(logContext, this.client, groupId, config.getInt("max.poll.interval.ms"), config.getInt("session.timeout.ms"), heartbeatIntervalMs, this.assignors, this.metadata, this.subscriptions, this.metrics, metricGrpPrefix, this.time, this.retryBackoffMs, config.getBoolean("enable.auto.commit"), config.getInt("auto.commit.interval.ms"), this.interceptors, config.getBoolean("exclude.internal.topics"), config.getBoolean("internal.leave.group.on.close"));
                    this.fetcher = new Fetcher(logContext, this.client, config.getInt("fetch.min.bytes"), config.getInt("fetch.max.bytes"), config.getInt("fetch.max.wait.ms"), config.getInt("max.partition.fetch.bytes"), config.getInt("max.poll.records"), config.getBoolean("check.crcs"), this.keyDeserializer, this.valueDeserializer, this.metadata, this.subscriptions, this.metrics, metricsRegistry.fetcherMetrics, this.time, this.retryBackoffMs, this.requestTimeoutMs, isolationLevel);
                    config.logUnused();
                    AppInfoParser.registerAppInfo("kafka.consumer", clientId, this.metrics);
                    this.log.debug("Kafka consumer initialized");
                } else {
                    throw new ConfigException("request.timeout.ms should be greater than session.timeout.ms and fetch.max.wait.ms");
                }
            } catch (Throwable var24) {
                this.close(0L, true);
                throw new KafkaException("Failed to construct kafka consumer", var24);
            }
        }
    
        KafkaConsumer(LogContext logContext, String clientId, ConsumerCoordinator coordinator, Deserializer<K> keyDeserializer, Deserializer<V> valueDeserializer, Fetcher<K, V> fetcher, ConsumerInterceptors<K, V> interceptors, Time time, ConsumerNetworkClient client, Metrics metrics, SubscriptionState subscriptions, Metadata metadata, long retryBackoffMs, long requestTimeoutMs, List<PartitionAssignor> assignors) {

**rocketmq 属性**

/**
     * RocketMQ 配置常量类
     *
     * @author rewerma 2020-01-27
     * @version 1.0.0
     */
    public class RocketMQConstants {
    
        public static final String ROOT                                  = "rocketmq";
    
        public static final String ROCKETMQ_PRODUCER_GROUP               = ROOT + "." + "producer.group";
        public static final String ROCKETMQ_ENABLE_MESSAGE_TRACE         = ROOT + "." + "enable.message.trace";
        public static final String ROCKETMQ_CUSTOMIZED_TRACE_TOPIC       = ROOT + "." + "customized.trace.topic";
        public static final String ROCKETMQ_NAMESPACE                    = ROOT + "." + "namespace";
        public static final String ROCKETMQ_NAMESRV_ADDR                 = ROOT + "." + "namesrv.addr";
        public static final String ROCKETMQ_RETRY_TIMES_WHEN_SEND_FAILED = ROOT + "." + "retry.times.when.send.failed";
        public static final String ROCKETMQ_VIP_CHANNEL_ENABLED          = ROOT + "." + "vip.channel.enabled";
        public static final String ROCKETMQ_TAG                          = ROOT + "." + "tag";
    
        public static final String ROCKETMQ_ACCESS_CHANNEL               = ROOT + "." + "access.channel";
        public static final String ROCKETMQ_BATCH_SIZE                   = ROOT + "." + "batch.size";
        public static final String ROCKETMQ_SUBSCRIBE_FILTER             = ROOT + "." + "subscribe.filter";
    
    }
    
    
    ****************
    CanalRocketMQConsumer
    
    /**
     * RocketMQ consumer SPI 实现
     *
     * @author rewerma 2020-02-01
     * @version 1.0.0
     */
    @SPI("rocketmq")
    public class CanalRocketMQConsumer implements CanalMsgConsumer {
    
        private static final Logger                                logger               = LoggerFactory.getLogger(CanalRocketMQConsumer.class);
        private static final String                                CLOUD_ACCESS_CHANNEL = "cloud";
    
        private String                                             nameServer;
        private String                                             topic;
        private String                                             groupName;
        private DefaultMQPushConsumer                              rocketMQConsumer;
        private BlockingQueue<ConsumerBatchMessage<CommonMessage>> messageBlockingQueue;
        private int                                                batchSize            = -1;
        private long                                               batchProcessTimeout  = 60 * 1000;
        private boolean                                            flatMessage;
        private volatile ConsumerBatchMessage<CommonMessage>       lastGetBatchMessage  = null;
        private String                                             accessKey;
        private String                                             secretKey;
        private String                                             customizedTraceTopic;
        private boolean                                            enableMessageTrace   = false;
        private String                                             accessChannel;
        private String                                             namespace;
        private String                                             filter               = "*";
    
        @Override
        public void init(Properties properties, String topic, String groupName) {
            this.topic = topic;
            this.groupName = groupName;
            this.flatMessage = (Boolean) properties.get(CanalConstants.CANAL_MQ_FLAT_MESSAGE);
            this.messageBlockingQueue = new LinkedBlockingQueue<>(1024);
            this.accessKey = properties.getProperty(CanalConstants.CANAL_ALIYUN_ACCESS_KEY);
            this.secretKey = properties.getProperty(CanalConstants.CANAL_ALIYUN_SECRET_KEY);
            String enableMessageTrace = properties.getProperty(RocketMQConstants.ROCKETMQ_ENABLE_MESSAGE_TRACE);
            if (StringUtils.isNotEmpty(enableMessageTrace)) {
                this.enableMessageTrace = Boolean.parseBoolean(enableMessageTrace);
            }
            this.customizedTraceTopic = properties.getProperty(RocketMQConstants.ROCKETMQ_CUSTOMIZED_TRACE_TOPIC);
            this.accessChannel = properties.getProperty(RocketMQConstants.ROCKETMQ_ACCESS_CHANNEL);
            this.namespace = properties.getProperty(RocketMQConstants.ROCKETMQ_NAMESPACE);
            this.nameServer = properties.getProperty(RocketMQConstants.ROCKETMQ_NAMESRV_ADDR);
            String batchSize = properties.getProperty(RocketMQConstants.ROCKETMQ_BATCH_SIZE);
            if (StringUtils.isNotEmpty(batchSize)) {
                this.batchSize = Integer.parseInt(batchSize);
            }
            String subscribeFilter = properties.getProperty(RocketMQConstants.ROCKETMQ_SUBSCRIBE_FILTER);
            if (StringUtils.isNotEmpty(subscribeFilter)) {
                this.filter = subscribeFilter;
            }
        }
    
        。。。

**rabbitmq 属性**

package com.alibaba.otter.canal.connector.rabbitmq.config;
    
    /**
     * RabbitMQ 配置常量类
     *
     * @author rewerma 2020-01-27
     * @version 1.0.0
     */
    public class RabbitMQConstants {
    
        public static final String ROOT                      = "rabbitmq";
    
        public static final String RABBITMQ_HOST             = ROOT + "." + "host";
        public static final String RABBITMQ_EXCHANGE         = ROOT + "." + "exchange";
        public static final String RABBITMQ_VIRTUAL_HOST     = ROOT + "." + "virtual.host";
        public static final String RABBITMQ_USERNAME         = ROOT + "." + "username";
        public static final String RABBITMQ_PASSWORD         = ROOT + "." + "password";
    
        public static final String RABBITMQ_RESOURCE_OWNERID = ROOT + "." + "rabbitmq.resource.ownerId";
    }
    
    
    
    ***************
    CanalRabbitMQConsumer
    
    @SPI("rabbitmq")
    public class CanalRabbitMQConsumer implements CanalMsgConsumer {
    
        private static final Logger                                logger              = LoggerFactory.getLogger(CanalRabbitMQConsumer.class);
    
        // 链接地址
        private String                                             nameServer;
        // 主机名
        private String                                             vhost;
        private String                                             queueName;
    
        // 一些鉴权信息
        private String                                             accessKey;
        private String                                             secretKey;
        private Long                                               resourceOwnerId;
        private String                                             username;
        private String                                             password;
    
        private boolean                                            flatMessage;
    
        private Connection                                         connect;
        private Channel                                            channel;
    
        private long                                               batchProcessTimeout = 60 * 1000;
        private BlockingQueue<ConsumerBatchMessage<CommonMessage>> messageBlockingQueue;
        private volatile ConsumerBatchMessage<CommonMessage>       lastGetBatchMessage = null;
    
        @Override
        public void init(Properties properties, String topic, String groupId) {
            this.nameServer = PropertiesUtils.getProperty(properties, "rabbitmq.host");
            this.vhost = PropertiesUtils.getProperty(properties, "rabbitmq.virtual.host");
            this.queueName = topic;
            this.accessKey = PropertiesUtils.getProperty(properties, CanalConstants.CANAL_ALIYUN_ACCESS_KEY);
            this.secretKey = PropertiesUtils.getProperty(properties, CanalConstants.CANAL_ALIYUN_SECRET_KEY);
            this.username = PropertiesUtils.getProperty(properties, RabbitMQConstants.RABBITMQ_USERNAME);
            this.password = PropertiesUtils.getProperty(properties, RabbitMQConstants.RABBITMQ_PASSWORD);
            Long resourceOwnerIdPro = (Long) properties.get(RabbitMQConstants.RABBITMQ_RESOURCE_OWNERID);
            if (resourceOwnerIdPro != null) {
                this.resourceOwnerId = resourceOwnerIdPro;
            }
            this.flatMessage = (Boolean) properties.get(CanalConstants.CANAL_MQ_FLAT_MESSAGE);
            this.messageBlockingQueue = new LinkedBlockingQueue<>(1024);
        }
    
        。。。

## \*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\* ##

**适配器配置文件**

**rdb 配置文件**

dataSourceKey: defaultDS         #对应启动器配置文件中的数据源key
    destination: example             #canal server中的instance name
    groupId: g1                      #对应启动器配置文件中的groupId
    outerAdapterKey: mysql1          #对应启动器配置文件mysql1
    concurrent: true
    dbMapping:
      database: mytest               #源数据库、表
      table: user
      targetTable: mytest2.user      #目的地数据库、表
      targetPk:                      #主键
        id: id
    #  mapAll: true
      targetColumns:                 #列
        id:
        name:
        role_id:
        c_time:
        test1:
      etlCondition: "where c_time>={}"     #筛选条件
      commitBatch: 3000 # 批量提交的大小
    
    
    ## Mirror schema synchronize config
    #dataSourceKey: defaultDS
    #destination: example
    #groupId: g1
    #outerAdapterKey: mysql1
    #concurrent: true
    #dbMapping:
    #  mirrorDb: true
    #  database: mytest

**es7.x 配置文件**

mytest_user.yml
    
    dataSourceKey: defaultDS
    destination: example
    groupId: g1
    esMapping:
      _index: mytest_user
      _id: _id
    #  upsert: true
    #  pk: id
      sql: "select a.id as _id, a.name, a.role_id, b.role_name,
            a.c_time from user a
            left join role b on b.id=a.role_id"
    #  objFields:
    #    _labels: array:;
      etlCondition: "where a.c_time>={}"
      commitBatch: 3000
    
    
    *******************
    customer.yml
    
    dataSourceKey: defaultDS
    destination: example
    groupId: g1
    esMapping:
      _index: customer
      _id: id
      relations:
        customer_order:
          name: customer
      sql: "select t.id, t.name, t.email from customer t"
      etlCondition: "where t.c_time>={}"
      commitBatch: 3000
    
    # 自定义映射
    #{
    #  "mappings":{
    #    "_doc":{
    #      "properties":{
    #        "id": {
    #          "type": "long"
    #        },
    #        "name": {
    #          "type": "text"
    #        },
    #        "email": {
    #          "type": "text"
    #        },
    #        "order_id": {
    #          "type": "long"
    #        },
    #        "order_serial": {
    #          "type": "text"
    #        },
    #        "order_time": {
    #          "type": "date"
    #        },
    #        "customer_order":{
    #          "type":"join",
    #          "relations":{
    #            "customer":"order"
    #          }
    #        }
    #      }
    #    }
    #  }
    #}
    
    
    *******************
    biz_order.yml
    
    dataSourceKey: defaultDS
    destination: example
    groupId: g1
    esMapping:
      _index: customer
      _id: _id
      relations:
        customer_order:
          name: order
          parent: customer_id
      sql: "select concat('oid_', t.id) as _id,
            t.customer_id,
            t.id as order_id,
            t.serial_code as order_serial,
            t.c_time as order_time
            from biz_order t"
      skips:
        - customer_id
      etlCondition: "where t.c_time>={}"
      commitBatch: 3000

## \*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\* ##

**adapter rest管理接口**

**查看订阅的canal instance、mq**

curl http://127.0.0.1:8081/destinations

**数据同步开关****（针对具体的instance、mq）**

curl -X PUT http://127.0.0.1:8081/syncSwitch/example/on
    curl -X PUT http://127.0.0.1:8081/syncSwitch/example/off

**查看instance、mq同步状态**

curl http://127.0.0.1:8081/syncSwitch/example

**传递etlCondition 中的参数**

curl -X POST http://127.0.0.1:8081/etl/hbase/mytest_person2.yml \
    -d "params=2018-10-21 00:00:00"
    
    # 对应的adapter配置文件，etcCondition：where c_time>={}读取参数params
    dataSourceKey: defaultDS
    destination: example
    groupId: g1
    hbaseMapping:
      mode: STRING  #NATIVE   #PHOENIX
      database: mytest                             # 数据库名
      table: person2                               # 数据库表名
      hbaseTable: MYTEST.PERSON2                   # HBase表名
      family: CF                                   # 默认统一Family名称
      uppercaseQualifier: true                     # 字段名转大写, 默认为true
      etlCondition: "where c_time>={}"
      commitBatch: 3000                            # 批量提交的大小
      #rowKey: id,type                             # 复合字段rowKey不能和columns中的rowKey重复
      columns:                                     # 数据库字段:HBase对应字段
        id: ROWKEY LEN:15
        name: NAME
        email: EMAIL
        type:
        c_time: C_TIME
        birthday: BIRTHDAY

**查看相关库总数据**

curl http://127.0.0.1:8081/count/hbase/mytest_person2.yml

[https_github.com_alibaba_canal_wiki_ClientAdapter]: https://github.com/alibaba/canal/wiki/ClientAdapter
[https_registry.hub.docker.com_r_slpcat_canal-adapter]: https://registry.hub.docker.com/r/slpcat/canal-adapter
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MzkzMTYyNQ_size_16_color_FFFFFF_t_70]: /images/20221005/309b1550a94d4486b6c82f89f97b3823.png
[20210704205930899.png]: /images/20221005/67187f2ff6554d9c9601538624ee54f2.png