kafka-es.go

package es

import (
"context"
"encoding/json"
"fmt"
"kafka/mongo"
"log"
"os"

"github.com/olivere/elastic"

)

var client *elastic.Client

var host = "http://192.168.184.10:9200"

//初始化es驱动
func Connect() {
client, _ = elastic.NewClient(
// ...
// 将sniff设置为false后，便不会自动转换地址
elastic.SetSniff(false),
)
errorlog := log.New(os.Stdout, "app", log.LstdFlags)

var err error
client, err = elastic.NewClient(elastic.SetErrorLog(errorlog), elastic.SetURL(host), elastic.SetSniff(false))
if err != nil {
	panic(err)
}
info, code, err := client.Ping(host).Do(context.Background())
if err != nil {
	panic(err)
}
fmt.Printf("Es return with code %d and version %s \n", code, info.Version.Number)
esversionCode, err := client.ElasticsearchVersion(host)
if err != nil {
	fmt.Println("err:")
	panic(err)
}
fmt.Printf("es version %s\n", esversionCode)

}

//查找
func Get() {
get, err := client.Get().Index("test").Type("_doc").Id("Mk33VIkB0JvaroO0nbrb").Do(context.Background())
if err != nil {
panic(err)
}
if get.Found {
fmt.Printf("got document %s in version %d from index %s,type %s \n", get.Id, get.Version, get.Index, get.Type)

	var user mongo.User
	// 这里别忘了*，因为res.Source是地址
	if err := json.Unmarshal(*get.Source, &user); err != nil {
		fmt.Println("转换成json失败")
		return
	}
	fmt.Println(user)
}

}
func Insert() {
name := "test"
data := { "name": "wali", "age": 30, "city": "shenzhen" }
_, err := client.Index().Index(name).Type("_doc").Id("1").BodyJson(data).Do(context.Background())
fmt.Println(err)
}