[TOC] ## 搜索引擎 ### Elasticsearch Elasticsearch是開源分布式搜索引擎的霸主，其依賴于Lucene實現，在部署和運維方面做了很多優化。當今搭建一個分布式搜索引擎比起Sphinx的時代已經是容易很多很多了。只要簡單配置客戶端IP和端口就可以了 #### go sdk ``` // 選用 elastic 版本時 // 注意與自己使用的 elasticsearch 要對應 import ( elastic "gopkg.in/olivere/elastic.v3" ) var esClient *elastic.Client func initElasticsearchClient(host string, port string) { var err error esClient, err = elastic.NewClient( elastic.SetURL(fmt.Sprintf("http://%s:%s", host, port)), elastic.SetMaxRetries(3), ) if err != nil { // log error } } ``` 插入： ``` func insertDocument(db string, table string, obj map[string]interface{}) { id := obj["id"] var indexName, typeName string // 數據庫中的 database/table 概念，可以簡單映射到 es 的 index 和 type // 不過需要注意，因為 es 中的 _type 本質上只是 document 的一個字段 // 所以單個 index 內容過多會導致性能問題 // 在新版本中 type 已經廢棄 // 為了讓不同表的數據落入不同的 index，這里我們用 table+name 作為 index 的名字 indexName = fmt.Sprintf("%v_%v", db, table) typeName = table // 正常情況 res, err := esClient.Index().Index(indexName).Type(typeName).Id(id).BodyJson(obj).Do() if err != nil { // handle error } else { // insert success } } ``` 獲取： ``` func query(indexName string, typeName string) (*elastic.SearchResult, error) { // 通過 bool must 和 bool should 添加 bool 查詢條件 q := elastic.NewBoolQuery().Must(elastic.NewMatchPhraseQuery("id", 1), elastic.NewBoolQuery().Must(elastic.NewMatchPhraseQuery("male", "m"))) q = q.Should( elastic.NewMatchPhraseQuery("name", "alex"), elastic.NewMatchPhraseQuery("name", "xargin"), ) searchService := esClient.Search(indexName).Type(typeName) res, err := searchService.Query(q).Do() if err != nil { // log error return nil, err } return res, nil } ``` 刪除： ``` func deleteDocument( indexName string, typeName string, obj map[string]interface{}, ) { id := obj["id"] res, err := esClient.Delete().Index(indexName).Type(typeName).Id(id).Do() if err != nil { // handle error } else { // delete success } } ``` #### 將 sql 轉換為 DSL ``` github.com/cch123/elasticsql ``` demo ``` package main import ( "fmt" "github.com/cch123/elasticsql" ) var sql = ` select * from aaa where a=1 and x = '三個男人' and create_time between '2015-01-01T00:00:00+0800' and '2016-01-01T00:00:00+0800' and process_id > 1 order by id desc limit 100,10 ` func main() { dsl, esType, _ := elasticsql.Convert(sql) fmt.Println(dsl) fmt.Println(esType) } ``` ## 異構數據同步 在實際應用中，我們很少直接向搜索引擎中寫入數據。更為常見的方式是，將MySQL或其它關系型數據中的數據同步到搜索引擎中。而搜索引擎的使用方只能對數據進行查詢，無法進行修改和刪除 ### 通過時間戳進行增量數據同步 把最近十分鐘創建的所有出庫單取出，批量存入es中，取數據的操作需要執行的邏輯可以表達為下面的SQL： ``` select * from wms_orders where update_time >= date_sub(now(), interval 10 minute); ``` 當然，考慮到邊界情況，我們可以讓這個時間段的數據與前一次的有一些重疊： ``` select * from wms_orders where update_time >= date_sub( now(), interval 11 minute ); ``` ### 通過 binlog 進行數據同步  業界使用較多的是阿里開源的Canal，來進行binlog解析與同步。**canal會偽裝成MySQL的從庫**，然后解析好行格式的binlog，再以更容易解析的格式（例如json）發送到消息隊列