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版本。

以上为个人经验，希望能给大家一个参考，也希望大家多多支持。如有错误或未考虑完全的地方，望不吝赐教。