男欢女爱,泰吉虎,小儿腮腺炎的治疗
在开发过程中,我们经常会遇到对业务数据进行模糊搜索的需求,例如电商网站对于商品的搜索,以及内容网站对于内容的关键字检索等等。对于这些高级的搜索功能,显然数据库的 like 是不合适的,通常我们采用 elasticsearch 来完成数据的搜索和分析,有了这个利器,我们可以轻松应对上述场景,实现关键字搜索等功能。
不过,由于增加了 elasticsearch 作为搜索引擎,随之而来的问题就是,如何将业务中的数据同步到 elasticsearch 中,主要有两种方式:
由于业务双写需要更改业务代码,一般不建议采用此种方式,除非有强一致性要求,或者对业务侵入不敏感的系统可以采取此种方式:
本文主要讨论非代码侵入的数据库同步方式,主要采用的是通过 logstash 定时扫描数据库来增量同步数据的方案。
数据库表结构中,需要有一个时间类型的字段作为增量更新的标识字段(例如 lastupdatetime),当该条数据更新时,必须同时更新该字段。
create table user ( `id` int(11) not null, `name` varchar(50) not null, `age` int(11) not null, `createtime` datetime(0) not null, `lastupdatetime` datetime(0) not null, primary key (`id`) using btree ) insert into `user` values(1,"jack",18,now(),now()) insert into `user` values(2,"william",18,now(),now()) select * from `user`
查询结果:
id | name | age | createtime | lastupdatetime |
---|---|---|---|---|
1 | jack | 18 | 2019-10-24 10:31:14 | 2019-10-24 10:31:14 |
2 | william | 18 | 2019-10-24 10:31:49 | 2019-10-24 10:31:49 |
logstash docker 安装脚本:
mkdir /opt/logstashsync/
mkdir /opt/logstashsync/pipeline
vi /opt/logstashsync/pipeline/logstash.conf
input { jdbc { jdbc_driver_library => "/app/mysql-connector-java-8.0.18.jar" jdbc_driver_class => "com.mysql.jdbc.driver" jdbc_connection_string => "jdbc:mysql://192.168.10.102:3306/synctest" jdbc_user => "root" jdbc_password => "123456" tracking_column => "unix_ts_in_secs" use_column_value => true schedule => "*/5 * * * * *" statement => "select *, unix_timestamp(lastupdatetime) as unix_ts_in_secs from user where (unix_timestamp(lastupdatetime) > :sql_last_value and lastupdatetime < now()) order by lastupdatetime asc" } } filter { mutate { copy => { "id" => "[@metadata][_id]"} remove_field => ["id", "@version", "unix_ts_in_secs"] } } output { elasticsearch { hosts => "192.168.10.102:9200" index => "syncuser" timeout => 300 document_id => "%{[@metadata][_id]}" } }
上述配置说明:
有了上述配置,我们把 logstath 的 docker 容器跑起来:
docker run -d \ -v /opt/logstashsync/config/logstash.yml:/usr/share/logstash/config/logstash.yml \ -v /opt/logstashsync/pipeline/logstash.conf:/usr/share/logstash/pipeline/logstash.conf \ -v /opt/logstashsync/mysql-connector-java-8.0.18.jar:/app/mysql-connector-java-8.0.18.jar \ --name=logstash \ logstash:6.7.1
注意:上述脚本可以看到,我们将本地 /opt/logstashsync/ 目录下的 mysql-connector-java-8.0.18.jar 映射到了容器的 /app 目录下,对应在上述 logstash.conf 中的配置的 jdbc_driver_library 的值
通过查看 logstash 容器运行日志,我们可以看到如下日志内容,说明该容易已经按照我们预期的每5s同步一次数据库:
[2019-10-25t06:27:59,056][info ][logstash.inputs.jdbc ] (0.039651s) select *, unix_timestamp(lastupdatetime) as unix_ts_in_secs from user where (unix_timestamp(lastupdatetime) > 0 and lastupdatetime < now()) order by lastupdatetime asc [2019-10-25t06:28:05,154][info ][logstash.inputs.jdbc ] (0.004232s) select *, unix_timestamp(lastupdatetime) as unix_ts_in_secs from user where (unix_timestamp(lastupdatetime) > 1571913109 and lastupdatetime < now()) order by lastupdatetime asc [2019-10-25t06:28:10,230][info ][logstash.inputs.jdbc ] (0.002832s) select *, unix_timestamp(lastupdatetime) as unix_ts_in_secs from user where (unix_timestamp(lastupdatetime) > 1571913109 and lastupdatetime < now()) order by lastupdatetime asc
在 kibana 中创建 syncuser index,即可以查看到已经同步的数据:
尝试对数据库数据做更新操作,将名为 william 的用户年龄修改为100(记得同时要更新lastupdatetime 字段):
update `user` set age=100, lastupdatetime=now() where `name`='william'; select * from `user`
再次查看 kibana 中的数据,可以看到该数据已经成功同步:
根据上述过程,我们完成了简单的单表数据定时同步至 elasticsearch 过程,但是在实际使用过程中,需要注意以下问题:
如对本文有疑问,请在下面进行留言讨论,广大热心网友会与你互动!! 点击进行留言回复
Blazor server side 自家的一些开源的, 实用型项目的进度之 CEF客户端
.NET IoC模式依赖反转(DIP)、控制反转(Ioc)、依赖注入(DI)
vue+.netcore可支持业务代码扩展的开发框架 VOL.Vue 2.0版本发布
网友评论