想要实时搜索，最重要的是同步数据库-软件测试文章-51Testing软件测试圈

首页文章问答 51讲堂活动

写文章提问题

我的主页通知已购设置我的钱包
退出

想要实时搜索，最重要的是同步数据库
曼倩诙谐 2020-10-30 09:53:58 字数 11594 阅读 1411 收藏 3

其它

　　目前大部分mysql和elasticsearch同步机制使用的插件实现的，常用的插件包为：logstash-input-jdbc,go-mysql-elasticsearch, elasticsearch-jdbc、canal。

　　插件优缺点对比

　　1. logstash-input-jdbc

　　logstash官方插件,集成在logstash中,下载logstash即可,通过配置文件实现mysql与elasticsearch数据同步

　　优点

　　·能实现mysql数据全量和增量的数据同步,且能实现定时同步。

　　·版本更新迭代快,相对稳定。

　　·作为ES固有插件logstash一部分,易用。

　　缺点

　　·不能实现同步删除操作,MySQL数据删除后Elasticsearch中数据仍存在。

　　·同步最短时间差为一分钟,一分钟数据同步一次,无法做到实时同步。

　　2、go-mysql-elasticsearch

　　go-mysql-elasticsearch 是国内作者开发的一款插件

　　优点

　　·能实现mysql数据增加,删除,修改操作的实时数据同步

　　缺点

　　·无法实现数据全量同步Elasticsearch

　　·仍处理开发、相对不稳定阶段

　　3、elasticsearch-jdbc

　　目前最新的版本是2.3.4，支持的ElasticSearch的版本为2.3.4, 未实践

　　优点

　　·能实现mysql数据全量和增量的数据同步.

　　缺点

　　·目前最新的版本是2.3.4，支持的ElasticSearch的版本为2.3.4

　　·不能实现同步删除操作,MySQL数据删除后Elasticsearch中数据仍存在.

　　mysql安装

　　安装依赖

　　yum search libaio # 检索相关信息

　　yum install libaio # 安装依赖包

　　mysql是否安装

　　yum list installed | grep mysql

　　mysql卸载

　　yum -y remove mysql-libs.x86_64

　　mysql yum下载

　　wget -i -c http://dev.mysql.com/get/mysql57-community-release-el7-10.noarch.rpm

　　安装 MySQL

　　yum -y install mysql57-community-release-el7-10.noarch.rpm

　　yum -y install mysql-community-server

　　查看mysql安装位置

　　whereis mysql

　　启动mysql

　　systemctl start mysqld.service

　　systemctl status mysqld.service

　　关闭mysql

　　systemctl stop mysqld

　　查看密码

　　grep 'temporary password' /var/log/mysqld.log

　　mysql修改密码远程连接

　　SET PASSWORD = PASSWORD('/20as3SElksds0ew98');

　　GRANT ALL PRIVILEGES ON *.* TO 'root'@'%' IDENTIFIED BY '/20as3SElksds0ew98' WITH GRANT OPTION;

　　logstash-input-jdbc实现mysql数据库与elasticsearch同步

　　logstash5.x之后，集成了logstash-input-jdbc插件。安装logstash后通过命令安装logstash-input-jdbc插件

　　./logstash-plugin install logstash-input-jdbc

　　配置

　　在logstash/config文件夹下新建jdbc.conf,配置内容如下：

　# 输入部分
　　input {
　　  stdin {}
　　  jdbc {
　　    # mysql数据库驱动
　　    jdbc_driver_library => "../config/mysql-connector-java-5.1.30.jar"
　　    jdbc_driver_class => "com.mysql.jdbc.Driver"
　　    # mysql数据库链接，数据库名
　　    jdbc_connection_string => "jdbc:mysql://localhost:3306/cmr"
　　    # mysql数据库用户名，密码
　　    jdbc_user => "root"
　　    jdbc_password => "12345678"
　　    # 设置监听间隔  各字段含义（分、时、天、月、年），全部为*默认含义为每分钟更新一次
　　    schedule => "* * * * *"
　　    # 分页
　　    jdbc_paging_enabled => "true"
　　    # 分页大小
　　    jdbc_page_size => "50000"
　　    # sql语句执行文件，也可直接使用 statement => 'select * from t_school_archives_fold create_time >= 
　　                                :sql_last_value order by create_time limit 200000'
　　    statement_filepath => "/config/jdbc.sql"
　　    # elasticsearch索引类型名
　　    type => "t_employee"
　　  }
　　}
　　# 过滤部分(不是必须项）
　　filter {
　　    json {
　　        source => "message"
　　        remove_field => ["message"]
　　    }
　　}
　　# 输出部分
　　output {
　　    elasticsearch {
　　        # elasticsearch索引名
　　        index => "octopus"
　　        # 使用input中的type作为elasticsearch索引下的类型名
　　        document_type => "%{type}"   # <- use the type from each input
　　        # elasticsearch的ip和端口号
　　        hosts => "localhost:9200"
　　        # 同步mysql中数据id作为elasticsearch中文档id
　　        document_id => "%{id}"
　　    }
　　    stdout {
　　        codec => json_lines
　　    }
　　}
　　# 注: 使用时请去掉此文件中的注释，不然会报错

　在config 目录下新建jdbc.sql文件

　　select * from t_employee

　　运行

cd logstash-6.4.2
　　# 检查配置文件语法是否正确
　　bin/logstash -f config/jdbc.conf --config.test_and_exit
　　# 启动
　　bin/logstash -f config/jdbc.conf --config.reload.automatic
　　--config.reload.automatic: 会自动重新加载配置文件内容
　　在kibana中创建索引后查看同步数据
　　PUT octopus
　　GET octopus/_search

　Canal实现mysql数据库与elasticsearch同步

　　mysql

　　修改/etc/my.cnf

　　log-bin=mysql-bin
　　binlog-format=ROW  
　　server-id=1

　　创建授权

　　create user canal identified by 'Canal@2020!';   #创建canal账户
　　grant select,replication slave,replication client on *.* to 'canal'@'%'; #授权canal账户查询和复制权限
　　flush privileges;                                                        #刷新授权

　查看binlog是否正确启动

　　show variables like 'binlog_format%';

　　创建需要同步的数据库

　　create database canal_testdb character set utf8;  
　　CREATE TABLE canal_table (   
　　#创建canal_table表，字段为 id age name address
　　id int(11) NOT NULL,
　　age int(11) NOT NULL,
　　name varchar(200) NOT NULL,
　　address varchar(1000) DEFAULT NULL,
　　PRIMARY KEY (id)
　　) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
　　INSERT INTO canal_testdb.canal_table(id, age, name, address) VALUES (1, 88, '小明', '测试');

　Elasticsearch参考第一节

　　部署Canal-deployer服务端

　　下载并解压

　　# 没有的话新建
　　cd /usr/local/canal/canal-deployer/  
　　wget https://github.com/alibaba/canal/releases/download/canal-1.1.5-alpha-1/canal.deployer-1.1.5-SNAPSHOT.tar.gz
　　tar xf canal.deployer-1.1.5-SNAPSHOT.tar.gz

　修改配置文件 instance.properties

　　vim /usr/local/canal-deployer/conf/example/instance.properties
　　#################################################
　　## mysql serverId , v1.0.26+ will autoGen
　　canal.instance.mysql.slaveId=3
　　# enable gtid use true/false
　　canal.instance.gtidon=false
　　# position info
　　canal.instance.master.address=127.0.0.1:3306
　　canal.instance.master.journal.name=
　　canal.instance.master.position=
　　canal.instance.master.timestamp=
　　canal.instance.master.gtid=
　　# rds oss binlog
　　canal.instance.rds.accesskey=
　　canal.instance.rds.secretkey=
　　canal.instance.rds.instanceId=
　　# table meta tsdb info
　　canal.instance.tsdb.enable=true
　　#canal.instance.tsdb.url=jdbc:mysql://127.0.0.1:3306/canal_testdb
　　#canal.instance.tsdb.dbUsername=canal
　　#canal.instance.tsdb.dbPassword=canal
　　#canal.instance.standby.address =
　　#canal.instance.standby.journal.name =
　　#canal.instance.standby.position =
　　#canal.instance.standby.timestamp =
　　#canal.instance.standby.gtid=
　　# username/password
　　canal.instance.dbUsername=canal
　　canal.instance.dbPassword=Canal@2020!
　　canal.instance.connectionCharset = UTF-8
　　# enable druid Decrypt database password
　　canal.instance.enableDruid=false
　　#canal.instance.pwdPublicKey=MFwwDQYJKoZIhvcNAQEBBQADSwAwSAJBALK4BUxdDltRRE5/zXpVEVPUgunvscYFtEip3pmLlhrWpacX7y7GCMo2/JM6LeHmiiNdH1FWgGCpUfircSwlWKUCAwEAAQ==
　　# table regex
　　canal.instance.filter.regex=.*\\..*
　　# table black regex
　　canal.instance.filter.black.regex=
　　# table field filter(format: schema1.tableName1:field1/field2,schema2.tableName2:field1/field2)
　　#canal.instance.filter.field=test1.t_product:id/subject/keywords,test2.t_company:id/name/contact/ch
　　# table field black filter(format: schema1.tableName1:field1/field2,schema2.tableName2:field1/field2)
　　#canal.instance.filter.black.field=test1.t_product:subject/product_image,test2.t_company:id/name/contact/ch
　　# mq config
　　canal.mq.topic=example
　　# dynamic topic route by schema or table regex
　　#canal.mq.dynamicTopic=mytest1.user,mytest2\\..*,.*\\..*
　　canal.mq.partition=0
　　# hash partition config
　　#canal.mq.partitionsNum=3
　　#canal.mq.partitionHash=test.table:id^name,.*\\..*
　　#################################################

　启动canal-deployer

　　因为canal-depaloyer由java开发，所以需要jdk环境，jdk版本需要大于1.5

　　yum install java-1.8.0-openjdk.x86_64 java-1.8.0-openjdk-devel.x86_64 -y

　　/usr/local/canal/canal-deployer/bin/startup.sh

　　查看日志及端口

　　tail -f /usr/local/canal/logs/example/example.log

　　canal-deployer默认监听三个端口,11110、11111、11112

　　11110：为admin管理端口

　　11111：为canal deployer 服务器占用的端口

　　11112：为指标下拉端口

　　部署Canal-adapter客户端

　　下载并解压

　　cd /usr/local/canal/canal-adapter/  
　　wget https://github.com/alibaba/canal/releases/download/canal-1.1.5-alpha-1/canal.adapter-1.1.5-SNAPSHOT.tar.gz
　　tar xf canal.adapter-1.1.5-SNAPSHOT.tar.gz

　添加mysql8.0.18连接器

　　cd /usr/local/canal/canal-adapter/lib/
　　wget https://repo1.maven.org/maven2/mysql/mysql-connector-java/8.0.18/mysql-connector-java-8.0.18.jar
　　chmod 777 /usr/local/canal-adapter/lib/mysql-connector-java-8.0.18.jar                  #权限修改与其它lib库一致
　　chmod +st /usr/local/canal-adapter/lib/mysql-connector-java-8.0.18.jar

　修改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 # kafka rocketMQ
　　  canalServerHost: 127.0.0.1:11111
　　#  zookeeperHosts: slave1:2181
　　#  mqServers: 127.0.0.1:9092 #or rocketmq
　　#  flatMessage: true
　　  batchSize: 500
　　  syncBatchSize: 1000
　　  retries: 0
　　  timeout:
　　  srcDataSources:
　　    defaultDS:
　　      url: jdbc:mysql://127.0.0.1:3306/canal_testdb?useUnicode=true
　　      username: canal
　　      password: Canal@2020!
　　  canalAdapters:
　　  - instance: example # canal instance Name or mq topic name
　　    groups:
　　    - groupId: g1
　　      outerAdapters:
　　      - name: logger
　　      - name: es7
　　        hosts: 192.168.0.200:9300,192.168.0.200:8200
　　        properties:
　　          mode: rest # or rest
　　          # security.auth: test:123456 #  only used for rest mode
　　          cluster.name: my-es

　修改适配器映射文件

　　vim /usr/local/canal/canal-adapter/conf/es7/mytest_user.yml
　　dataSourceKey: defaultDS                                        #指定在application.yml文件中srcDataSources源数据源自定义的名称
　　destination: example                                            #cannal的instance或者MQ的topic，我们是把数据同步至es，所以不用修改，也用不到此处
　　groupId: g1                                                     #对应MQ模式下的groupId, 只会同步对应groupId的数据
　　esMapping:                                                      #es中的Mapping设置
　　  _index: canal_tsdb                                            #指定索引名称
　　  _id: _id                                                      #指定文档id，_id 此值则由es自动分配文档ID
　　  sql: "select a.id as _id,a.age,a.name,a.address from canal_table a"        #sql映射
　　  etlCondition: "where a.c_time>={}"                            #etl的条件参数
　　  commitBatch: 3000                                             #提交批大小

　Elasticsearch创建索引

　　POST canal_tsdb/_doc
　　{
　　    "mappings":{
　　        "_doc":{
　　            "properties":{
　　                "age":{
　　                    "type":"long"
　　                },
　　                "name":{
　　                    "type":"text"
　　                },
　　                "address":{
　　                    "type":"text"
　　                }
　　            }
　　        }
　　    }

　　启动Canal-adapter并写入数据

　　/usr/local/canal/canal-adapter/bin/startup.sh

　　tail -f /usr/local/canal/canal-adapter/logs/adapter/adapter.log

　　在MySQL再次插入一条数据并查看日志

　　INSERT INTO canal_tsdb.canal_table(id, age, name, address) VALUES (2, 88, '小明', '测试');

　　查看Canal-deployer服务端日志

　　tail -f /usr/local/canal/canal-deployer/logs/example/meta.log

　　在es里面可以看到数据

　　部署Canal Admin

　　canal-admin的限定依赖：

　　1.MySQL，用于存储配置和节点等相关数据

　　2.canal版本，要求>=1.1.4 (需要依赖canal-server提供面向admin的动态运维管理接口)

　　Canal Admin下载并解压

　　mkdir /usr/local/canal/canal-admin
　　wget https://github.com/alibaba/canal/releases/download/canal-1.1.5-alpha-1/canal.admin-1.1.5-SNAPSHOT.tar.gz
　　tar xf canal.admin-1.1.5-SNAPSHOT.tar.gz

　application.yml

server:
　　  port: 8089                                    #Canal Admin监听端口
　　spring:
　　  jackson:
　　    date-format: yyyy-MM-dd HH:mm:ss            #时间格式
　　    time-zone: GMT+8                            #时区
　　spring.datasource:                              #数据库信息
　　  address: 192.168.0.200:8809                  #指定Canal Admin所使用的数据库地址及端口
　　  database: canal_manager                       #指定数据库名称
　　  username: cadmin                              #指定数据库账户
　　  password: Cadmin@2020!                        #指定数据库密码
　　  driver-class-name: com.mysql.jdbc.Driver      #指定数据库驱动
　　  url: jdbc:mysql://${spring.datasource.address}/${spring.datasource.database}?useUnicode=true&characterEncoding=UTF-8&useSSL=false
　　  hikari:
　　    maximum-pool-size: 30
　　    minimum-idle: 1
　　canal:                                          #Canal UI界面默认账号密码
　　  adminUser: admin
　　  adminPasswd: admin

　创建数据库及授权用户

　　create database canal_manager character set utf8;
　　create database canal_manager character set utf8;
　　create user cadmin identified by 'Cadmin@2020!';
　　grant all on canal_manager.* to 'cadmin'@'%';
　　flush privileges;

　　/usr/local/canal/canal-admin/conf/canal_manager.sql;数据库数据导入

　　启动

　　/usr/local/canal/canal-admin/bin/startup.sh

作者：wmburst

来源：51Testing软件测试网原创

0.00 查看剩余0%
【留下美好印记】

赞赏支持

登录后发表评论

曼倩诙谐测试

+ 关注

热门文章

最新讲堂