golang elasticsearch Client的使用詳解
elasticsearch 的client ，通過 NewClient 建立連接，通過 NewClient 中的 Set.URL設置訪問的地址，SetSniff設置集群
獲得連接 後，通過 Index 方法插入數據，插入後可以通過 Get 方法獲得數據（最後的測試用例中會使用 elasticsearch client 的Get 方法）
func Save(item interface{}) {
client, err := elastic.NewClient(
elastic.SetURL("http://192.168.174.128:9200/"),
// Must turn off sniff in docker
elastic.SetSniff(false),
)
if err != nil {
panic(err)
}
resp, err := client.Index().
Index("dating_profile").
Type("zhenai").
BodyJson(item).
Do(context.Background()) //contex需要context 包
if err != nil {
panic(err)
}
fmt.Printf("%+v", resp)
}
測試程序，自行定義一個數據結構 Profile 進行測試
func TestSave(t *testing.T) {
profile := model.Profile{
Age: 34,
Height: 162,
Weight: 57,
Income: "3001-5000元",
Gender: "女",
Name: "安靜的雪",
XingZuo: "牡羊座",
Occupation: "人事/行政",
Marriage: "離異",
House: "已購房",
Hukou: "山東菏澤",
Education: "大學本科",
Car: "未購車",
}
Save(profile)
}
go test 成功
通過 Get 方法查看數據是否存在elasticsearch 中
我們在test中panic，在函數中講錯誤返回。在從elastisearch中 取出存入的數據，與我們定義的數據進行比較，
所以save中需要將插入數據的Id返回出來
func Save(item interface{}) (id string, err error) {
client, err := elastic.NewClient(
elastic.SetURL("http://192.168.174.128:9200/"),
// Must turn off sniff in docker
elastic.SetSniff(false),
)
if err != nil {
return "", err
}
resp, err := client.Index().
Index("dating_profile").
Type("zhenai").
BodyJson(item).
Do(context.Background())
if err != nil {
return "", err
}
return resp.Id, nil
}
測試用例
package persist
import (
"context"
"encoding/json"
"my_crawler_single/model"
"testing"
elastic "gopkg.in/olivere/elastic.v5"
)
func TestSave(t *testing.T) {
expected := model.Profile{
Age: 34,
Height: 162,
Weight: 57,
Income: "3001-5000元",
Gender: "女",
Name: "安靜的雪",
XingZuo: "牡羊座",
Occupation: "人事/行政",
Marriage: "離異",
House: "已購房",
Hukou: "山東菏澤",
Education: "大學本科",
Car: "未購車",
}
id, err := Save(expected)
if err != nil {
panic(err)
}
client, err := elastic.NewClient(
elastic.SetURL("http://192.168.174.128:9200/"),
elastic.SetSniff(false),
)
if err != nil {
panic(err)
}
resp, err := client.Get().
Index("dating_profile").
Type("zhenai").
Id(id). //查找指定id的那一條數據
Do(context.Background())
if err != nil {
panic(err)
}
t.Logf("%+v", resp)
//從打印得知，數據在resp.Source中，從rest client的截圖也可以知道
var actual model.Profile
//查看 *resp.Source 可知其數據類型為[]byte
err = json.Unmarshal(*resp.Source, &actual)
if err != nil {
panic(err)
}
if actual != expected {
t.Errorf("got %v;expected %v", actual, expected)
}
}
補充：go-elasticsearch: Elastic官方的Go語言客戶端
說明
Elastic官方鼓勵在項目中嘗試用這個包，但請記住以下幾點：
這個項目的工作還在進行中，並非所有計劃的功能和Elasticsearch官方客戶端中的標準（故障重試，節點自動發現等）都實現瞭。
API穩定性無法保證。 盡管公共API的設計非常謹慎，但它們可以根據進一步的探索和用戶反饋以不兼容的方式進行更改。
客戶端的目標是Elasticsearch 7.x版本。後續將添加對6.x和5.x版本API的支持。
安裝
用go get安裝這個包：
go get -u github.com/elastic/go-elasticsearch
或者將這個包添加到go.mod文件：
require github.com/elastic/go-elasticsearch v0.0.0
或者克隆這個倉庫：
git clone https://github.com/elastic/go-elasticsearch.git \u0026amp;\u0026amp; cd go-elasticsearch
一個完整的示例：
mkdir my-elasticsearch-app \u0026amp;\u0026amp; cd my-elasticsearch-appcat \u0026gt; go.mod \u0026lt;\u0026lt;-END module my-elasticsearch-app require github.com/elastic/go-elasticsearch v0.0.0ENDcat \u0026gt; main.go \u0026lt;\u0026lt;-END package main import ( \u0026quot;log\u0026quot; \u0026quot;github.com/elastic/go-elasticsearch\u0026quot; ) func main() { es, _ := elasticsearch.NewDefaultClient() log.Println(es.Info()) }ENDgo run main.go
用法
elasticsearch包與另外兩個包綁定在一起，esapi用於調用Elasticsearch的API，estransport通過HTTP傳輸數據。
使用elasticsearch.NewDefaultClient()函數創建帶有以下默認設置的客戶端：
es, err := elasticsearch.NewDefaultClient()if err != nil { log.Fatalf(\u0026quot;Error creating the client: %s\u0026quot;, err)}res, err := es.Info()if err != nil { log.Fatalf(\u0026quot;Error getting response: %s\u0026quot;, err)}log.Println(res)// [200 OK] {// \u0026quot;name\u0026quot; : \u0026quot;node-1\u0026quot;,// \u0026quot;cluster_name\u0026quot; : \u0026quot;go-elasticsearch\u0026quot;// ...
註意：當導出ELASTICSEARCH_URL環境變量時，它將被用作集群端點。
使用elasticsearch.NewClient()函數（僅用作演示）配置該客戶端：
cfg := elasticsearch.Config{ Addresses: []string{ \u0026quot;http://localhost:9200\u0026quot;, \u0026quot;http://localhost:9201\u0026quot;, }, Transport: \u0026amp;http.Transport{ MaxIdleConnsPerHost: 10, ResponseHeaderTimeout: time.Second, DialContext: (\u0026amp;net.Dialer{Timeout: time.Second}).DialContext, TLSClientConfig: \u0026amp;tls.Config{ MaxVersion: tls.VersionTLS11, InsecureSkipVerify: true, }, },}es, err := elasticsearch.NewClient(cfg)// ...
下面的示例展示瞭更復雜的用法。它從集群中獲取Elasticsearch版本，同時索引幾個文檔，並使用響應主體周圍的一個輕量包裝器打印搜索結果。
// $ go run _examples/main.gopackage mainimport ( \u0026quot;context\u0026quot; \u0026quot;encoding/json\u0026quot; \u0026quot;log\u0026quot; \u0026quot;strconv\u0026quot; \u0026quot;strings\u0026quot; \u0026quot;sync\u0026quot; \u0026quot;github.com/elastic/go-elasticsearch\u0026quot; \u0026quot;github.com/elastic/go-elasticsearch/esapi\u0026quot;)func main() { log.SetFlags(0) var ( r map[string]interface{} wg sync.WaitGroup ) // Initialize a client with the default settings. // // An `ELASTICSEARCH_URL` environment variable will be used when exported. // es, err := elasticsearch.NewDefaultClient() if err != nil { log.Fatalf(\u0026quot;Error creating the client: %s\u0026quot;, err) } // 1. Get cluster info // res, err := es.Info() if err != nil { log.Fatalf(\u0026quot;Error getting response: %s\u0026quot;, err) } // Deserialize the response into a map. if err := json.NewDecoder(res.Body).Decode(\u0026amp;r); err != nil { log.Fatalf(\u0026quot;Error parsing the response body: %s\u0026quot;, err) } // Print version number. log.Printf(\u0026quot;~~~~~~~\u0026gt; Elasticsearch %s\u0026quot;, r[\u0026quot;version\u0026quot;].(map[string]interface{})[\u0026quot;number\u0026quot;]) // 2. Index documents concurrently // for i, title := range []string{\u0026quot;Test One\u0026quot;, \u0026quot;Test Two\u0026quot;} { wg.Add(1) go func(i int, title string) { defer wg.Done() // Set up the request object directly. req := esapi.IndexRequest{ Index: \u0026quot;test\u0026quot;, DocumentID: strconv.Itoa(i + 1), Body: strings.NewReader(`{\u0026quot;title\u0026quot; : \u0026quot;` + title + `\u0026quot;}`), Refresh: \u0026quot;true\u0026quot;, } // Perform the request with the client. res, err := req.Do(context.Background(), es) if err != nil { log.Fatalf(\u0026quot;Error getting response: %s\u0026quot;, err) } defer res.Body.Close() if res.IsError() { log.Printf(\u0026quot;[%s] Error indexing document ID=%d\u0026quot;, res.Status(), i+1) } else { // Deserialize the response into a map. var r map[string]interface{} if err := json.NewDecoder(res.Body).Decode(\u0026amp;r); err != nil { log.Printf(\u0026quot;Error parsing the response body: %s\u0026quot;, err) } else { // Print the response status and indexed document version. log.Printf(\u0026quot;[%s] %s; version=%d\u0026quot;, res.Status(), r[\u0026quot;result\u0026quot;], int(r[\u0026quot;_version\u0026quot;].(float64))) } } }(i, title) } wg.Wait() log.Println(strings.Repeat(\u0026quot;-\u0026quot;, 37)) // 3. Search for the indexed documents // // Use the helper methods of the client. res, err = es.Search( es.Search.WithContext(context.Background()), es.Search.WithIndex(\u0026quot;test\u0026quot;), es.Search.WithBody(strings.NewReader(`{\u0026quot;query\u0026quot; : { \u0026quot;match\u0026quot; : { \u0026quot;title\u0026quot; : \u0026quot;test\u0026quot; } }}`)), es.Search.WithTrackTotalHits(true), es.Search.WithPretty(), ) if err != nil { log.Fatalf(\u0026quot;ERROR: %s\u0026quot;, err) } defer res.Body.Close() if res.IsError() { var e map[string]interface{} if err := json.NewDecoder(res.Body).Decode(\u0026amp;e); err != nil { log.Fatalf(\u0026quot;error parsing the response body: %s\u0026quot;, err) } else { // Print the response status and error information. log.Fatalf(\u0026quot;[%s] %s: %s\u0026quot;, res.Status(), e[\u0026quot;error\u0026quot;].(map[string]interface{})[\u0026quot;type\u0026quot;], e[\u0026quot;error\u0026quot;].(map[string]interface{})[\u0026quot;reason\u0026quot;], ) } } if err := json.NewDecoder(res.Body).Decode(\u0026amp;r); err != nil { log.Fatalf(\u0026quot;Error parsing the response body: %s\u0026quot;, err) } // Print the response status, number of results, and request duration. log.Printf( \u0026quot;[%s] %d hits; took: %dms\u0026quot;, res.Status(), int(r[\u0026quot;hits\u0026quot;].(map[string]interface{})[\u0026quot;total\u0026quot;].(map[string]interface{})[\u0026quot;value\u0026quot;].(float64)), int(r[\u0026quot;took\u0026quot;].(float64)), ) // Print the ID and document source for each hit. for _, hit := range r[\u0026quot;hits\u0026quot;].(map[string]interface{})[\u0026quot;hits\u0026quot;].([]interface{}) { log.Printf(\u0026quot; * ID=%s, %s\u0026quot;, hit.(map[string]interface{})[\u0026quot;_id\u0026quot;], hit.(map[string]interface{})[\u0026quot;_source\u0026quot;]) } log.Println(strings.Repeat(\u0026quot;=\u0026quot;, 37))}// ~~~~~~~\u0026gt; Elasticsearch 7.0.0-SNAPSHOT// [200 OK] updated; version=1// [200 OK] updated; version=1// -------------------------------------// [200 OK] 2 hits; took: 7ms// * ID=1, map[title:Test One]// * ID=2, map[title:Test Two]// =====================================
如上述示例所示，esapi包允許通過兩種不同的方式調用Elasticsearch API：通過創建結構（如IndexRequest），並向其傳遞上下文和客戶端來調用其Do（）方法，或者通過客戶端上可用的函數（如WithIndex（））直接調用其上的Search（）函數。更多信息請參閱包文檔。
estransport包處理與Elasticsearch之間的數據傳輸。 目前，這個實現隻占據很小的空間：它隻在已配置的集群端點上進行循環。後續將添加更多功能：重試失敗的請求，忽略某些狀態代碼，自動發現群集中的節點等等。
Examples
_examples文件夾包含許多全面的示例，可幫助你上手使用客戶端，包括客戶端的配置和自定義，模擬單元測試的傳輸，將客戶端嵌入自定義類型，構建查詢，執行請求和解析回應。
許可證
遵循Apache License 2.0版本。
以上為個人經驗，希望能給大傢一個參考，也希望大傢多多支持WalkonNet。如有錯誤或未考慮完全的地方，望不吝賜教。
推薦閱讀：
- docker安裝elastic search的詳細過程
- 使用Docker Compose搭建部署ElasticSearch的配置過程
- 在Django中使用ElasticSearch
- 一文詳解Golang的中間件設計模式
- ElasticSearch核心概念