elasticsearch 的client ，通过 newclient 建立连接，通过 newclient 中的 set.url设置访问的地址，setsniff设置集群
获得连接 后，通过 index 方法插入数据，插入后可以通过 get 方法获得数据（最后的测试用例中会使用 elasticsearch client 的get 方法）
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
|
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 进行测试
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
|
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返回出来
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
|
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 } |
测试用例
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
|
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安装这个包：
|
1
|
go get -u github.com/elastic/go-elasticsearch |
或者将这个包添加到go.mod文件：
|
1
|
require github.com/elastic/go-elasticsearch v0.0.0 |
或者克隆这个仓库：
|
1
|
git clone https://github.com/elastic/go-elasticsearch.git \u0026amp;\u0026amp; cd go-elasticsearch |
一个完整的示例：
|
1
|
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()函数创建带有以下默认设置的客户端：
|
1
|
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()函数（仅用作演示）配置该客户端：
|
1
|
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版本，同时索引几个文档，并使用响应主体周围的一个轻量包装器打印搜索结果。
|
1
|
// $ 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版本。
以上为个人经验，希望能给大家一个参考，也希望大家多多支持服务器之家。如有错误或未考虑完全的地方，望不吝赐教。
原文链接：https://blog.csdn.net/insisxz/article/details/97612109
