golang elasticsearch Client的使用詳解

更新時間：2021年05月04日 11:01:16 作者：insisxz

這篇文章主要介紹了golang elasticsearch Client的使用詳解，具有很好的參考價值，希望對大家有所幫助。一起跟隨小編過來看看吧

elasticsearch 的client ，通過 NewClient 建立連接，通過 NewClient 中的 Set.URL設(shè)置訪問的地址，SetSniff設(shè)置集群

獲得連接后，通過 Index 方法插入數(shù)據(jù)，插入后可以通過 Get 方法獲得數(shù)據(jù)（最后的測試用例中會使用 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)
 
}

測試程序，自行定義一個數(shù)據(jù)結(jié)構(gòu) Profile 進(jìn)行測試

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:  "大學(xué)本科",
  Car:        "未購車",
 }
 
 Save(profile)
}

go test 成功

通過 Get 方法查看數(shù)據(jù)是否存在elasticsearch 中

我們在test中panic，在函數(shù)中講錯誤返回。在從elastisearch中取出存入的數(shù)據(jù)，與我們定義的數(shù)據(jù)進(jìn)行比較，

所以save中需要將插入數(shù)據(jù)的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:  "大學(xué)本科",
		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的那一條數(shù)據(jù)
		Do(context.Background())
	if err != nil {
		panic(err)
	}
 
	t.Logf("%+v", resp)
	//從打印得知，數(shù)據(jù)在resp.Source中，從rest client的截圖也可以知道
 
	var actual model.Profile
	//查看 *resp.Source 可知其數(shù)據(jù)類型為[]byte
	err = json.Unmarshal(*resp.Source, &actual)
	if err != nil {
		panic(err)
	}
 
	if actual != expected {
		t.Errorf("got %v;expected %v", actual, expected)
	}
}

補(bǔ)充：go-elasticsearch: Elastic官方的Go語言客戶端

說明

Elastic官方鼓勵在項目中嘗試用這個包，但請記住以下幾點：

這個項目的工作還在進(jìn)行中，并非所有計劃的功能和Elasticsearch官方客戶端中的標(biāo)準(zhǔn)（故障重試，節(jié)點自動發(fā)現(xiàn)等）都實現(xiàn)了。

API穩(wěn)定性無法保證。盡管公共API的設(shè)計非常謹(jǐn)慎，但它們可以根據(jù)進(jìn)一步的探索和用戶反饋以不兼容的方式進(jìn)行更改。

客戶端的目標(biāo)是Elasticsearch 7.x版本。后續(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用于調(diào)用Elasticsearch的API，estransport通過HTTP傳輸數(shù)據(jù)。

使用elasticsearch.NewDefaultClient()函數(shù)創(chuàng)建帶有以下默認(rèn)設(shè)置的客戶端：

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;// ...

注意：當(dāng)導(dǎo)出ELASTICSEARCH_URL環(huán)境變量時，它將被用作集群端點。

使用elasticsearch.NewClient()函數(shù)（僅用作演示）配置該客戶端：

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)// ...

下面的示例展示了更復(fù)雜的用法。它從集群中獲取Elasticsearch版本，同時索引幾個文檔，并使用響應(yīng)主體周圍的一個輕量包裝器打印搜索結(jié)果。

// $ 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包允許通過兩種不同的方式調(diào)用Elasticsearch API：通過創(chuàng)建結(jié)構(gòu)（如IndexRequest），并向其傳遞上下文和客戶端來調(diào)用其Do（）方法，或者通過客戶端上可用的函數(shù)（如WithIndex（））直接調(diào)用其上的Search（）函數(shù)。更多信息請參閱包文檔。

estransport包處理與Elasticsearch之間的數(shù)據(jù)傳輸。目前，這個實現(xiàn)只占據(jù)很小的空間：它只在已配置的集群端點上進(jìn)行循環(huán)。后續(xù)將添加更多功能：重試失敗的請求，忽略某些狀態(tài)代碼，自動發(fā)現(xiàn)群集中的節(jié)點等等。