最新国产好看的视频,伊人天堂AV在线,国产Aaaaaa视频,蜜臀视频在线观看一区,人妻av色图,密臀久久久精品影片,青青视频免费观看毛片,久草在线观看视,国产三级精品色情在线

mysql使用Canal監(jiān)聽binlog日志,同步數(shù)據(jù)到MySQL、ElasticSearch、Redis、Kafka、RocketMQ和es

 更新時間:2026年03月14日 11:58:34   作者:fixAllenSun  
Canal是阿里巴巴開源的一款基于MySQL數(shù)據(jù)庫binlog的增量數(shù)據(jù)訂閱和消費組件,canal.adapter,相當于canal的客戶端,會從canal-deployer中獲取數(shù)據(jù),模擬MySQL slave協(xié)議解析binlog,捕獲數(shù)據(jù)變更事件,通過客戶端可以同步到MySQL、Elasticsearch和HBase等存儲中去

【一】安裝Canal

Canal [k?'næl] 是阿里巴巴開源的一款基于MySQL數(shù)據(jù)庫binlog的增量數(shù)據(jù)訂閱和消費組件。

MySQL Master → binlog → Canal偽裝Slave → 解析binlog → 數(shù)據(jù)分發(fā)
  • 偽裝機制:Canal模擬MySQL slave的交互協(xié)議,向MySQL master發(fā)送dump請求

  • 解析轉(zhuǎn)換:接收binlog并解析成JSON/AVRO等格式

  • 訂閱分發(fā):客戶端訂閱消費解析后的增量數(shù)據(jù)

【1】配置mysql

(1)mysql開啟binlog

修改 MySQL 配置文件my.cnf,重啟 MySQL

cd /root/NFTurbo_DockerCompose/mysql/config
vi my.cnf

開啟 Binlog 寫入功能,配置 binlog-format 為 ROW 模式,my.cnf 中配置如下:

[mysqld]
log-bin=mysql-bin # 開啟 binlog
binlog-format=ROW # 選擇 ROW 模式(Canal只支持ROW)
server_id=1 # 配置 MySQL replaction 需要定義,主庫ID,不要和 canal 的 slaveId 重復(fù)

binlog的路徑:/var/log/mysql/bin-log

重啟 MySQL

(2)授權(quán)

授權(quán) canal 鏈接 MySQL 賬號具有作為 MySQL slave 的權(quán)限, 如果已有賬戶可直接 grant

CREATE USER canal IDENTIFIED BY 'canal';

GRANT SELECT, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO 'canal'@'%';
-- GRANT ALL PRIVILEGES ON *.* TO 'canal'@'%' ;

FLUSH PRIVILEGES;

(3)檢查binlog是否已開啟

SHOW VARIABLES LIKE 'log_bin%'; -- Value為ON即成功
SHOW VARIABLES LIKE 'binlog_format'; -- Value為ROW即成功

重啟mysql

【2】安裝Canal-deployer(接收數(shù)據(jù))

(1)Canal-deployer下載

用的Canal-deployer是1.1.7版本。
下載地址:https://github.com/alibaba/canal/releases

下載canal.deployer,可以直接監(jiān)聽MySQL的binlog,把自己偽裝成MySQL的從庫,只負責(zé)接收數(shù)據(jù),并不做處理。

(2)解壓

(3)配置修改

修改canal.properties,增加
canal.ip = 127.0.0.1

cd /home/allensun/develop/canal/canal_deployer/conf
vim canal.properties

(4)修改example/instance.properties

(1)配置當前服務(wù)器的ip和端口
(2)配置監(jiān)聽mysql的用戶和密碼

cd /home/allensun/develop/canal/canal_deployer/conf/example
vi instance.properties

(5)啟動 Canal

到 bin 目錄下,執(zhí)行命令:./startup.sh

cd /home/allensun/develop/canal/canal_deployer/bin && ./stop.sh && ./startup.sh

有報錯

把jdk安裝上,再執(zhí)行

(6)啟動成功檢查

啟動成功,檢查canal/logs/canal.log
看到the canal server is running now …表示啟動成功。

tail -f /home/allensun/develop/canal/canal_deployer/logs/canal/canal.log

(7)查看主庫狀態(tài)的時候有報錯

tail -f /home/allensun/develop/canal/canal_deployer/logs/example/example.log

SHOW MASTER STATUS;語句報錯了,因為這是mysql8.0的語法,用來查看主庫狀態(tài)信息,mysql8.4應(yīng)該使用

SHOW BINARY LOG STATUS;

【3】安裝Canal-adapter(處理數(shù)據(jù))

(1)下載canal.adapter

同樣是使用1.1.7版本,和deployer 保持一致。
canal.adapter,相當于canal的客戶端,會從canal-deployer中獲取數(shù)據(jù),然后對數(shù)據(jù)進行同步,可以同步到MySQL、Elasticsearch和HBase等存儲中去。

到https://github.com/alibaba/canal/releases 下載

(2)解壓


Canal-Adapter 目錄結(jié)構(gòu)(默認):

canal-adapter/
├── conf/
│   ├── application.yml  # 全局核心配置
│   └── rdb/                    # MySQL同步專用目錄(放YAML映射文件)
│       └── mysql_sync.yml      # 自定義表映射YAML(名字隨意)
└── bin/startup.sh              # 啟動腳本

(3)全局核心配置:application.yml

全局配置:application.yml 負責(zé)連 Canal-Server + 目標 MySQL 從庫

server:
  port: 8081                       # REST API 端口
  tomcat:
    max-threads: 1                  # 最大線程數(shù)
    min-spare-threads: 1            # 最小空閑線程數(shù)
    
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
  canalServerHost: 127.0.0.1:11111  # Canal-Server地址
  
  # Canal-Server連接參數(shù)
  batchSize: 500                    # 每次獲取數(shù)據(jù)量
  syncBatchSize: 1000               # 同步批次提交量
  retries: 3                        # 重試次數(shù)
  timeout: 1000                     # 超時時間(ms)
  vhost: /                          # 虛擬主機
  filter: .*\\..*                   # 正則表達式過濾表
  
  # 源數(shù)據(jù)庫配置
  srcDataSources:
    defaultDS:                      # 數(shù)據(jù)源key,用于映射文件引用
      url: jdbc:mysql://127.0.0.1:3306/source_db?useUnicode=true&characterEncoding=UTF-8&useSSL=false
      username: root
      password: 123456
      lazy: false
      
  canalAdapters:                    # 適配器配置
  - instance: example               # Canal實例名,與Canal-Server配置一致
    groups:
    - groupId: g1                   # 分組ID
      outerAdapters:                # 目標數(shù)據(jù)源適配器
      - name: es                    # Elasticsearch適配器
        hosts: 127.0.0.1:9200       # ES地址
        properties:
          mode: rest                # 連接模式: rest/transport
          cluster.name: my-es       # 集群名
          
      - name: redis                 # Redis適配器
        hosts: 127.0.0.1:6379
        properties:
          mode: single              # 模式: single/sentinel/cluster
          database: 0
          
      - name: rdb                   # 關(guān)系型數(shù)據(jù)庫適配器(MySQL)
        key: mysql1                 # 適配器key
        properties:
          jdbc.driverClassName: com.mysql.cj.jdbc.Driver
          jdbc.url: jdbc:mysql://127.0.0.1:3306/target_db?useUnicode=true
          jdbc.username: root
          jdbc.password: 123456
          
      - name: kafka                 # Kafka適配器
        properties:
          bootstrap.servers: 127.0.0.1:9092
          acks: 1
          
      - name: rocketMQ              # RocketMQ適配器
        properties:
          namesrv.addr: 127.0.0.1:9876
          producer.group: canal-producer
          
  # 監(jiān)控配置
  monitor:
    enable: true
    port: 11112
    user: admin
    passwd: admin

(4)表映射配置:conf/xxx/xxx.yml

表映射:conf/rdb/*.yml 負責(zé)主表→從表的字段映射

# -------------------------- 固定全局配置 --------------------------
dataSourceKey: adapter       # 必須 = application.yml 中的 spring.datasource.adapter.key
destination: example         # 必須 = Canal 實例名
group: g1                    # 自定義分組,隨便寫
outerAdapterKey: rdb         # 固定:rdb

# -------------------------- 表映射規(guī)則(核心) --------------------------
tables:
  # 表 1:user 表同步
  - tableName: user            # 自定義標識,隨便寫
    sourceDb: source_db        # 主庫庫名
    sourceTable: user          # 主表名
    targetDb: target_db        # 從庫庫名
    targetTable: user          # 從表名
    mapAll: true               # 自動映射所有字段(主從結(jié)構(gòu)一致必須開)
    pk: id                     # 主鍵(不寫自動識別)

  # 表 2:product 表同步
  - tableName: product
    sourceDb: source_db
    sourceTable: product
    targetDb: target_db
    targetTable: product
    mapAll: true

  # 表 3:order 表同步
  - tableName: order
    sourceDb: source_db
    sourceTable: order
    targetDb: target_db
    targetTable: order
    mapAll: true

兩個 YAML 必須嚴格對應(yīng)(錯一個就失敗)

(5)啟動

# 進入Canal-Adapter根目錄
cd /your/path/canal-adapter
# 啟動(Linux/Mac)
sh bin/startup.sh
# 停止
sh bin/stop.sh

驗證啟動成功

# 查看日志(出現(xiàn)以下關(guān)鍵字即成功)
tail -f logs/adapter/adapter.log

? 成功日志:

Start to subscribe canal destination: example
Rdb adapter started successfully
All tables have been loaded

【4】工作原理

(1)MySQL主備復(fù)制原理
MySQL master 將數(shù)據(jù)變更寫入二進制日志( binary log, 其中記錄叫做二進制日志事件binary log events,可以通過 show binlog events 進行查看)
MySQL slave 將 master 的 binary log events 拷貝到它的中繼日志(relay log)
MySQL slave 重放 relay log 中事件,將數(shù)據(jù)變更反映它自己的數(shù)據(jù)

(2)canal 工作原理
canal 模擬 MySQL slave 的交互協(xié)議,偽裝自己為 MySQL slave ,向 MySQL master 發(fā)送dump 協(xié)議
MySQL master 收到 dump 請求,開始推送 binary log 給 slave (即 canal )
canal 解析 binary log 對象(原始為 byte 流)

【二】同步數(shù)據(jù)到MySQL、ElasticSearch、Redis、Kafka、RocketMQ

【1】前置準備

Canal-Server 已正常啟動,且監(jiān)聽 MySQL 主庫 binlog
MySQL從庫已創(chuàng)建好和主庫結(jié)構(gòu)一致的目標表
Canal-Adapter 目錄下:刪除原application.properties,新建application.yml(避免配置沖突)
表映射 YAML 必須放在 conf/rdb/ 目錄下(RDB 適配器固定目錄)

使用 Canal-Adapter 實現(xiàn)實時數(shù)據(jù)同步,核心是2 個配置文件:
(1)全局配置:conf/application.properties(也可以是conf/application.yml,連接 Canal-Server、配置同步目標、全局參數(shù))
(2)表映射配置:conf/xxx/xxx.yml(YAML 文件,指定哪張主表同步到哪張從表、字段映射規(guī)則)

關(guān)鍵:Canal-Adapter 同步到 MySQL 屬于RDB(關(guān)系型數(shù)據(jù)庫)適配器,YAML 文件必須放在 conf/rdb/ 目錄下!

MySQL Binlog
   ↓
Canal-Server
   ↓
Canal-Adapter(一套配置同步 5 個目的地)
   ├─ → MySQL 從庫(rdb)
   ├─ → ElasticSearch(es)
   ├─ → Redis(redis)
   ├─ → Kafka(kafka)
   └─ → RocketMQ(rocketmq)

【2】全局核心配置 conf/application.yml

作用:連接 Canal-Server + 配置目標 MySQL 從庫數(shù)據(jù)源 + 全局同步參數(shù)

server:
  port: 8081                       # REST API 端口
  tomcat:
    max-threads: 1                  # 最大線程數(shù)
    min-spare-threads: 1            # 最小空閑線程數(shù)
    
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
  canalServerHost: 127.0.0.1:11111  # Canal-Server地址
  
  # Canal-Server連接參數(shù)
  batchSize: 500                    # 每次獲取數(shù)據(jù)量
  syncBatchSize: 1000               # 同步批次提交量
  retries: 3                        # 重試次數(shù)
  timeout: 1000                     # 超時時間(ms)
  vhost: /                          # 虛擬主機
  filter: .*\\..*                   # 正則表達式過濾表
  
  # 源數(shù)據(jù)庫配置
  srcDataSources:
    defaultDS:                      # 數(shù)據(jù)源key,用于映射文件引用
      url: jdbc:mysql://127.0.0.1:3306/source_db?useUnicode=true&characterEncoding=UTF-8&useSSL=false
      username: root
      password: 123456
      lazy: false
      
  canalAdapters:                    # 適配器配置
  - instance: example               # Canal實例名,與Canal-Server配置一致
    groups:
    - groupId: g1                   # 分組ID
      outerAdapters:                # 目標數(shù)據(jù)源適配器
      - name: es                    # Elasticsearch適配器
        hosts: 127.0.0.1:9200       # ES地址
        properties:
          mode: rest                # 連接模式: rest/transport
          cluster.name: my-es       # 集群名
          
      - name: redis                 # Redis適配器
        hosts: 127.0.0.1:6379
        properties:
          mode: single              # 模式: single/sentinel/cluster
          database: 0
          
      - name: rdb                   # 關(guān)系型數(shù)據(jù)庫適配器(MySQL)
        key: mysql1                 # 適配器key
        properties:
          jdbc.driverClassName: com.mysql.cj.jdbc.Driver
          jdbc.url: jdbc:mysql://127.0.0.1:3306/target_db?useUnicode=true
          jdbc.username: root
          jdbc.password: 123456
          
      - name: kafka                 # Kafka適配器
        properties:
          bootstrap.servers: 127.0.0.1:9092
          acks: 1
          
      - name: rocketMQ              # RocketMQ適配器
        properties:
          namesrv.addr: 127.0.0.1:9876
          producer.group: canal-producer
          
  # 監(jiān)控配置
  monitor:
    enable: true
    port: 11112
    user: admin
    passwd: admin

【3】表映射配置 conf/rdb/mysql_sync.yml

作用:定義「主庫哪張表 → 目標數(shù)據(jù)源」的映射規(guī)則(支持多表同步)
?? YAML 嚴格縮進 2 空格,不能用 Tab!

5 套映射文件

(1)同步到 MySQL 從庫:conf/rdb/mysql.yml

# ====================== 表映射核心配置(必改:庫名、表名) ======================
# 1. 固定:和application.yml的spring.datasource.adapter.key一致
dataSourceKey: defaultDS
# 2. 固定:和application.yml的canal.conf.destinations一致
destination: example
# 3. 分組(自定義,默認即可)
group: g1
# 4. 適配器類型(固定rdb)
outerAdapterKey: mysql1

# 5. 表映射規(guī)則(核心:主表→從表,支持多表)
tables:
  # 表1:主庫user表 → 從庫user表
  - tableName: user                # 自定義標識(隨意)
    sourceDb: your_source_db       # 【需修改】主庫庫名
    sourceTable: user              # 【需修改】主庫表名
    targetDb: your_target_db       # 【需修改】從庫庫名
    targetTable: user              # 【需修改】從庫表名
    mapAll: true                   # 全字段自動映射(主從表結(jié)構(gòu)一致必開)

  # 表2:主庫product表 → 從庫product表
  - tableName: product
    sourceDb: your_source_db
    sourceTable: product
    targetDb: your_target_db
    targetTable: product
    mapAll: true

  # 表3:主庫order表 → 從庫order表
  - tableName: order
    sourceDb: your_source_db
    sourceTable: order
    targetDb: your_target_db
    targetTable: order
    mapAll: true

(2)同步到 ElasticSearch:conf/es/user.yml

dataSourceKey: defaultDS            # 對應(yīng)application.yml中的源數(shù)據(jù)源key
destination: example                # Canal實例名
groupId: g1                         # 分組ID
esMapping:
  _index: user_index                # ES索引名
  _type: _doc                       # ES類型(7.x后固定為_doc)
  _id: 'id'                         # 文檔ID字段,支持EL表達式: {id}
  upsert: true                      # 啟用upsert(存在則更新,不存在則插入)
  sql: "SELECT                      # 查詢SQL
    id,
    name,
    age,
    email,
    DATE_FORMAT(create_time, '%Y-%m-%d %H:%i:%s') as create_time,
    CASE status
      WHEN 1 THEN 'active'
      WHEN 0 THEN 'inactive'
      ELSE 'unknown'
    END as status_desc
    FROM user
    WHERE id > 0"
  etlCondition: "where create_time >= '{}'"  # ETL條件,{}會被參數(shù)替換
  commitBatch: 1000                  # 批量提交大小
  pk: id                            # 主鍵字段,用于增量判斷
  skips:                            # 跳過某些操作類型
    - INSERT
    - UPDATE
    - DELETE

(3)同步到 Redis:conf/redis/user.yml

dataSourceKey: defaultDS
destination: example
groupId: g1
key: 'user:{id}'                    # Redis key模板,支持{field}占位符
redisMapping:
  database: 0                       # Redis數(shù)據(jù)庫編號
  table: user                       # 源表名
  expireTime: 86400                 # 過期時間(秒),0表示永不過期
  pk:                               # 主鍵映射
    id: id
  type: string                      # Redis數(shù)據(jù)類型: string/hash/list/set/zset
  columns:                          # 字段映射
    id: id
    name: name
    age: age
    email: email
  sql: "SELECT id, name, age, email, status FROM user WHERE status = 1"
  operationType: ALL                # 操作類型: ALL/INSERT/UPDATE/DELETE

(4)同步到 Kafka:conf/kafka/user.yml

(1)application.yml Kafka專用配置

canal.conf:
  mode: kafka                       # 必須設(shè)置為kafka模式
  mqServers: 127.0.0.1:9092         # Kafka地址
  flatMessage: true                 # 是否使用扁平化消息格式
  parallelThreadSize: 8             # 并行處理線程數(shù)
  kafka:
    enable: true
    bootstrap.servers: 127.0.0.1:9092
    acks: 1
    compression.type: snappy
    batch.size: 16384
    linger.ms: 1

(2)映射文件:conf/kafka/user.yml

dataSourceKey: defaultDS
destination: example
groupId: g1
topic: canal.user.topic             # Kafka topic
partition: 0                        # 分區(qū)號,-1表示自動分配
partitionHash:                      # 分區(qū)哈希規(guī)則
  - id
tableMapping:                       # 表映射規(guī)則
  database: source_db               # 數(shù)據(jù)庫名
  table: user                       # 表名
  pk:                               # 主鍵字段
    id: id
  sql: "SELECT id, name, age, email FROM user"
  # 消息格式配置
  messageFormat: json               # 消息格式: json/protobuf
  includeFields:                    # 包含的字段
    - id
    - name
    - age
  excludeFields:                    # 排除的字段
    - password

(5)同步到 RocketMQ:conf/rocketmq/user.yml

(1)application.yml RocketMQ專用配置

canal.conf:
  mode: rocketMQ                    # 必須設(shè)置為rocketMQ模式
  mqServers: 127.0.0.1:9876         # NameServer地址
  flatMessage: true
  parallelThreadSize: 8
  rocketMQ:
    enable: true
    namesrv.addr: 127.0.0.1:9876
    producer.group: canal-producer
    enable.message.trace: false
    tag: canal                       # 默認tag

(2)映射文件:conf/rocketmq/user.yml

dataSourceKey: defaultDS
destination: example
groupId: g1
topic: CANAL_USER_TOPIC             # RocketMQ topic
tag: user_tag                       # 消息tag
key: '{id}'                         # 消息key
tableMapping:
  database: source_db
  table: user
  index: 1                          # 順序索引
  commitBatch: 100                  # 批量提交大小
  pk:
    id: id
  sql: "SELECT 
    id,
    name,
    age,
    email,
    create_time,
    update_time
    FROM user"
  # 消息過濾條件
  filterCondition: "status = 1"     # 只同步status=1的記錄
  # 字段轉(zhuǎn)換
  fieldMappings:
    id: userId
    name: userName
  # 消息屬性
  properties:
    retryTimes: 3
    timeout: 3000

【三】同步數(shù)據(jù)到es

【1】全局核心配置 conf/application.yml

修改 application.properties:

cd /root/canal_adapter/conf
vi 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
  flatMessage: true
  zookeeperHosts:
  syncBatchSize: 1000
  retries: -1
  timeout:
  accessKey:
  secretKey:
  consumerProperties:
    # canal tcp consumer
    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
    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
    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
    rabbitmq.host:
    rabbitmq.virtual.host:
    rabbitmq.username:
    rabbitmq.password:
    rabbitmq.resource.ownerId:

  srcDataSources:
    defaultDS:
      url: jdbc:mysql://rm-xxx.com:3306/nfturbo?useUnicode=true
      username: xxx
      password: xxx
  canalAdapters:
  - instance: example # canal instance Name or mq topic name
    groups:
    - groupId: g1
      outerAdapters:       
      - name: logger
      - name: es8
        hosts: localhost:9200 # 127.0.0.1:9200 for rest mode
        properties:
          mode: transport # or rest
          # security.auth: test:123456 #  only used for rest mode
          cluster.name: nfturbo-cluster

(1)srcDataSources配置的是監(jiān)聽Mysql數(shù)據(jù)庫的數(shù)據(jù)源信息
(2)canalAdapters配置的是es的連接信息,其中es8指向yaml配置文件,配置的是es的索引、主鍵id、查詢sql、批量提交數(shù)量等

【2】表映射配置 conf/rdb/mysql_sync.yml

因為在outerAdapters下面,我們配置的 name 是 es8(如果你自己安裝的是 es7,記得修改成對應(yīng)的版本。),所以adapter將會自動加載 conf/es8 下的所有.yml結(jié)尾的配置文件,適配器表映射文件,創(chuàng)建并修改 conf/es8/mytest_user.yml文件:

cd /root/canal_adapter/conf/es8
vi mytest_user.yml
dataSourceKey: defaultDS
destination: example
groupId: g1
esMapping:
  _index: nfturbo_users
  _id: _id
  #  upsert: true
  #  pk: id
  sql: "select t.id as _id, t.nick_name as  nick_name, t.state as state,t.telephone as telephone  from users as t"
  #  objFields:
  #    _labels: array:;
  #etlCondition: "where a.c_time>={}"
  commitBatch: 3000               

(4)啟動canal.adapter

同樣到 bin 目錄下,執(zhí)行命令:./startup.sh

cd /root/canal_adapter/bin && ./stop.sh && ./startup.sh
tail -f /root/canal_adapter/logs/adapter/adapter.log

【3】測試同步

(1)到 es 上創(chuàng)建 ES 索引
訪問 es(http://ip:5601/app/home#/ ),然后進入開發(fā)工具,在控制臺創(chuàng)建如下索引:
http://192.168.220.128:5601/app/dev_tools#/console

PUT nfturbo_users
{
  "mappings": {
    "properties": {
      "nickname": {
        "type": "text"
      },
      "telephone": {
        "type": "text"
      },
      "state": {
        "type": "text"
      }
    }
  }
}

(2)數(shù)據(jù)庫執(zhí)行INSERT

INSERT INTO `nfturbo`.`users` (`id`,`gmt_create`,`gmt_modified`,`nick_name`,`password_hash`,`state`,`telephone`,`user_role`) VALUES (14,'2024-04-18 17:47:40','2024-04-18 17:47:42','test11111','c2975f0faec10adca0ecd729c8cbc0aa','INIT','13000000000','CUSTOMER')

(3)從 ES 查詢:

GET nfturbo_users/_search
{"_source": ["nick_name","telephone","state"],
  "query": {
    "match": {
      "nick_name": "test11111"
    }
  }
}

到此這篇關(guān)于mysql使用Canal監(jiān)聽binlog日志,同步數(shù)據(jù)到MySQL、ElasticSearch、Redis、Kafka、RocketMQ和es的文章就介紹到這了,更多相關(guān)mysql使用Canal監(jiān)聽binlog日志來同步數(shù)據(jù)內(nèi)容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家!

相關(guān)文章

  • 5個MySQL GUI工具推薦,幫助你進行數(shù)據(jù)庫管理

    5個MySQL GUI工具推薦,幫助你進行數(shù)據(jù)庫管理

    這篇文章主要介紹了5個MySQL GUI工具推薦,幫助大家更好的進行MySQL數(shù)據(jù)庫管理,感興趣的朋友可以了解下
    2020-08-08
  • mysql數(shù)據(jù)表的基本操作之表結(jié)構(gòu)操作,字段操作實例分析

    mysql數(shù)據(jù)表的基本操作之表結(jié)構(gòu)操作,字段操作實例分析

    這篇文章主要介紹了mysql數(shù)據(jù)表的基本操作之表結(jié)構(gòu)操作,字段操作,結(jié)合實例形式分析了mysql表結(jié)構(gòu)操作,字段操作常見增刪改查實現(xiàn)技巧與操作注意事項,需要的朋友可以參考下
    2020-04-04
  • MySQL數(shù)據(jù)庫是如何實現(xiàn)XA規(guī)范的

    MySQL數(shù)據(jù)庫是如何實現(xiàn)XA規(guī)范的

    這篇文章主要介紹了MySQL數(shù)據(jù)庫是如何實現(xiàn)XA規(guī)范的,幫助大家更好的理解和使用MySQL數(shù)據(jù)庫,感興趣的朋友可以了解下
    2021-01-01
  • MYSQL實現(xiàn)連續(xù)簽到功能斷簽一天從頭開始(sql語句)

    MYSQL實現(xiàn)連續(xù)簽到功能斷簽一天從頭開始(sql語句)

    這篇文章主要介紹了MYSQL實現(xiàn)連續(xù)簽到功能斷簽一天從頭開始,非常不錯,具有參考借鑒價值,需要的朋友可以參考下
    2017-05-05
  • Linux下編譯安裝Mysql 5.5的簡單步驟

    Linux下編譯安裝Mysql 5.5的簡單步驟

    Linux下面因為從MySQL 5.5開始使用cmake來做config了,所以編譯安裝的會和5.1版本有些區(qū)別。不過總體來說還是差別不大
    2015-08-08
  • MySQL慢查詢之開啟慢查詢

    MySQL慢查詢之開啟慢查詢

    開啟慢查詢?nèi)罩?,可以讓MySQL記錄下查詢超過指定時間的語句,通過定位分析性能的瓶頸,才能更好的優(yōu)化數(shù)據(jù)庫系統(tǒng)的性能。下面這篇文章主要介紹了關(guān)于MySQL開啟慢查詢的相關(guān)資料,需要的朋友可以參考下。
    2017-03-03
  • 從MySQL轉(zhuǎn)換到PostgreSQL的遷移過程

    從MySQL轉(zhuǎn)換到PostgreSQL的遷移過程

    在數(shù)據(jù)庫遷移項目中,從MySQL轉(zhuǎn)換到PostgreSQL是一個常見但充滿挑戰(zhàn)的任務(wù),最近我在一個項目中遇到了這樣的需求,在轉(zhuǎn)換過程中遇到了各種語法錯誤和兼容性問題,本文將詳細記錄整個修復(fù)過程,希望能為遇到類似問題的開發(fā)者提供參考,需要的朋友可以參考下
    2026-04-04
  • MySql學(xué)習(xí)筆記之事務(wù)隔離級別詳解

    MySql學(xué)習(xí)筆記之事務(wù)隔離級別詳解

    這篇文章主要給大家介紹了關(guān)于MySql學(xué)習(xí)筆記之事務(wù)隔離級別的相關(guān)資料,文中通過示例代碼介紹的非常詳細,對大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價值,需要的朋友們下面隨著小編來一起學(xué)習(xí)學(xué)習(xí)吧
    2021-05-05
  • MySQL數(shù)據(jù)庫存儲引擎介紹及數(shù)據(jù)庫的操作詳解

    MySQL數(shù)據(jù)庫存儲引擎介紹及數(shù)據(jù)庫的操作詳解

    mysql面試中最常問的問題之一:小伙子,你說一下你們公司用的存儲引擎,以及你知道有哪些存儲引擎和他們之間的區(qū)別? 所以下面這篇文章主要給大家介紹了關(guān)于Mysql存儲引擎的相關(guān)資料,需要的朋友可以參考下
    2022-08-08
  • MySQL用limit方式實現(xiàn)分頁的實例方法

    MySQL用limit方式實現(xiàn)分頁的實例方法

    在本篇文章中小編給大家整理了一篇關(guān)于MySQL用limit方式實現(xiàn)分頁的實例方法,有需要的朋友們可以參考學(xué)習(xí)下。
    2020-01-01

最新評論

进贤县| 辉南县| 融水| 马鞍山市| 讷河市| 兴义市| 张家港市| 噶尔县| 文登市| 台中市| 郁南县| 泰来县| 台前县| 印江| 龙游县| 大渡口区| 科技| 淄博市| 民县| 武川县| 新泰市| 合山市| 航空| 邢台县| 定远县| 监利县| 全州县| 南川市| 长汀县| 巴马| 若羌县| 仁化县| 沁阳市| 台州市| 蓝田县| 济源市| 浑源县| 云南省| 格尔木市| 微山县| 西安市|