Go操作elasticsearch 基础操作

安装ES集群（Liunx版本）

参考链接​​这个博主的很不错​​

连接客户端

// 连接Es
func connectEs() (*elastic.Client, error) {
  return elastic.NewClient(
    // 设置Elastic服务地址
    elastic.SetURL("http://192.168.6.39:9201"),
    // 允许指定弹性是否应该定期检查集群，默认为true,会把请求http://ip:port/_nodes/http，
    // 并将其返回的publish_address作为请求路径
    //elastic.SetSniff(true),
    // 心跳检查,间隔时间
    elastic.SetHealthcheckInterval(time.Second*5),
    // 设置错误日志
    elastic.SetErrorLog(log.New(os.Stderr, "ES-ERROR ", log.LstdFlags)),
    // 设置info日志
    elastic.SetInfoLog(log.New(os.Stdout, "ES-INFO ", log.LstdFlags)),
  )
}

简单的一些查看操作

func TestBeego(t *testing.T) {
  client, err := connectEs()
  if err != nil {
    t.Error(err)
    return
  }
  // 健康检查
  Health, _ := client.ClusterHealth().Index().Do(context.TODO())
  // 查看索引
  Indexs, _ := client.CatIndices().Do(context.TODO())
  // 模糊查询某个索引
  Index, _ := client.CatIndices().Index(".apm*").Do(context.TODO())
  // 查看所有分片
  Nodes, _ := client.CatAllocation().Do(context.TODO())
  // 查看谁是主分片
  MasterNode, _ := client.CatMaster().Do(context.TODO())
  fmt.Println("健康检查: ", Health)
  fmt.Println("索引: ", Indexs)
  fmt.Println("查看所有分片: ", Nodes)
  fmt.Println("查看谁是主分片: ", MasterNode)
  fmt.Println("模糊查看索引 : ", Index)

}

创建索引

const indexName = "user_test"

// 创建索引 （自己指定mapping）
func TestCreateIndexMapping(t *testing.T) {
    userMapping := `{
    "mappings":{
        "properties":{
            "name":{
                "type":"keyword"
            },
            "age":{
                "type":"byte"
            },
            "birth":{
                "type":"date"
            }
        }
    }
}`
    // 获取客户端
    client, _ := connectEs()
    // 检索当前有没有该索引

    exist, err := client.IndexExists(indexName).Do(context.Background())
    if err != nil {
        t.Errorf("检测索引失败:%s", err)
        return
    }
    if exist {
        t.Error("索引已经存在，无需重复创建！")
        return
    }
    // 不存在就创建
    //  如果想直接创建索引，只需删除BodyString(userMapping)
    res, err := client.CreateIndex(indexName).BodyString(userMapping).Do(context.Background())
    if err != nil {
        t.Errorf("创建索引失败:%s", err)
        return
    }
    fmt.Println("创建成功： ", res)
}

添加数据（单条数据）

/ 添加数据（单条添加）
type UserInfo struct {
  Name  string `json:"name"`
  Age   int    `json:"age"`
  Birth string `json:"birth"`
}

func TestAddOne(t *testing.T) {
  client, _ := connectEs()
  // 创建userInfo
  userInfo := UserInfo{
    Age:   18,
    Name:  "冯消水",
    Birth: "2000-01-15",
  }
  // 这里可以选择指定或者不指定 Id 不指定ID 就会用默认的
  response, err := client.Index().Index(indexName).Id("1").BodyJson(userInfo).Do(context.Background())
  if err != nil {
    t.Errorf("添加失败:%s", err)
  }
  fmt.Println("添加成功", response)
}

批量添加

// 批量添加
func TestBatchAdd(t *testing.T) {
  client, _ := connectEs()
  userMap := make(map[string]string)
  userMap["fjj"] = "2000-01-02"
  userMap["mjd"] = "2000-02-02"
  userMap["liu"] = "2000-03-02"
  userMap["admin"] = "2000-04-02"
  rand.Seed(time.Now().Unix())
  // 创建bulk
  bulkService := client.Bulk().Index(indexName)
  id := 2
  for name, birth := range userMap {
    userinfos := UserInfo{
      Age:   rand.Intn(50),
      Name:  name,
      Birth: birth,
    }
    // 批量添加
    doc := elastic.NewBulkCreateRequest().Id(strconv.Itoa(id)).Doc(userinfos)
    bulkService.Add(doc)
    id++
  }
  // 检查添加的数据是否为空
  if bulkService.NumberOfActions() < 1 {
    t.Error("被添加的数据不能为空！")
    return
  }
  // 保存
  response, err := bulkService.Do(context.Background())
  if err != nil {
    t.Errorf("保存失败:%s", err)
    return
  }
  fmt.Println("保存成功: ", response)
}

通过id 修改数据

// 修改数据（script) 根据id更新
func TestUpdateOneByScript(t *testing.T) {
  client, _ := connectEs()
  // 通过id 修改
  updateResponse, err := client.Update().Index(indexName).Id("1").Script(elastic.NewScript("ctx._source.birth='2099-12-31'")).Do(context.Background())
  if err != nil {
    t.Errorf("根据ID更新单条记录失败:%s", err)
    return
  }
  fmt.Println("根据ID更新成功:", updateResponse.Result)
}

根据条件更新

// 根据条件更新
func TestUpdate(t *testing.T) {
  client, _ := connectEs()
  res, err := client.UpdateByQuery(indexName).Query(elastic.NewTermQuery("name", "mjd")).Script(elastic.NewScript("ctx._source.birth='2099-12-31'")).ProceedOnVersionConflict().Do(context.Background())
  if err != nil {
    t.Errorf("根据条件更新单条记录失败:%s", err)
    return
  }
  fmt.Println("根据条件更新成功:", res.Updated)
}

多字段更新

// 多字段更新
func TestUpdateByDoc(t *testing.T) {
  client, _ := connectEs()
  response, err := client.Update().Index(indexName).Id("2").Doc(map[string]interface{}{
    "name": "mjd2", "age": 39,
  }).Do(context.Background())
  if err != nil {
    t.Errorf("更新记录失败:%s", err)
    return
  }
  fmt.Println("根据条件更新成功:", response.Result)
}

批量更新

func TestBatchUpdate(t *testing.T) {
  client, _ := connectEs()
  bulkService := client.Bulk().Index(indexName)
  for _, id := range []string{"1", "2", "3", "4", "5"} {
    doc := elastic.NewBulkUpdateRequest().Id(id).Doc(map[string]interface{}{"age": 18})
    bulkService.Add(doc)
  }
  // 被更新的数量不能小于0
  if bulkService.NumberOfActions() < 0 {
    t.Error("被更新的数量不能为空")
    return
  }
  // 执行操作
  do, err := bulkService.Do(context.Background())
  if err != nil {
    t.Errorf("批量更新失败:%v", err)
    return
  }
  fmt.Println("更新成功:", do.Updated())
}

单条查询

// 单条查询
func TestSearchOneEs(t *testing.T) {
  client, _ := connectEs()
  getResult, err := client.Get().Index(indexName).Id("1").Do(context.Background())
  if err != nil {
    t.Errorf("获取失败: %s", err)
    return
  }
  // 把查出来的结果json
  json, _ := getResult.Source.MarshalJSON()
  fmt.Printf("查询单条结果:%s \n", json)
}

批量查询

func TestSearchMoreEs(t *testing.T) {
  client, _ := connectEs()
  searchResult, err := client.Search(indexName).Query(elastic.NewMatchQuery("age", 18)).From(0).Size(10).Pretty(true).Do(context.Background())
  if err != nil {
    t.Errorf("获取失败: %s", err)
    return
  }
  // 创建集合 用来存放
  var userList []UserInfo
  for _, i := range searchResult.Each(reflect.TypeOf(UserInfo{})) {
    tmp := i.(UserInfo)
    userList = append(userList, tmp)
  }
  fmt.Printf("查询结果:%v\n", userList)
}

精确匹配单个字段

func TestSearchByWhere(t *testing.T) {
  client, _ := connectEs()
  // 创建term
  termQuery := elastic.NewTermQuery("age", 18)
  searchResult, err := client.Search().
    // 设置索引名
    Index(indexName).
    // 设置查询条件
    Query(termQuery).
    // 设置排序字段 false 是逆序
    Sort("birth", true).
    // 分页参数
    From(0).Size(10). //设置每页的大小
    Pretty(true).     // 查询结果返回JSON格式
    // 执行请求
    Do(context.Background())
  if err != nil {
    t.Errorf("查询失败:%s", err)
    return
  }
  fmt.Printf("查询消耗时间 %d ms, 结果总数: %d\n", searchResult.TookInMillis, searchResult.TotalHits())
  if searchResult.TotalHits() > 0 {
    // 创建集合 用来存放
    var userList []UserInfo
    for _, i := range searchResult.Each(reflect.TypeOf(UserInfo{})) {
      tmp := i.(UserInfo)
      userList = append(userList, tmp)
    }
    fmt.Printf("查询结果:%v\n", userList)
  }
}

通过terms 实现Sql的in查询

// 通过terms 实现Sql  in 查询
func TestSearchByIn(t *testing.T) {
  client, _ := connectEs()
  termsQuery := elastic.NewTermsQuery("name", "mjd", "fjj")
  searchResult, err := client.Search().
    // 设置索引名
    Index(indexName).
    // 设置查询条件
    Query(termsQuery).
    // 设置排序字段 false 是逆序
    Sort("birth", true).
    // 分页参数
    From(0).Size(10). //设置每页的大小
    Pretty(true).     // 查询结果返回JSON格式
    // 执行请求
    Do(context.Background())
  if err != nil {
    t.Errorf("查询失败:%s", err)
    return
  }
  fmt.Printf("查询消耗时间 %d ms, 结果总数: %d\n", searchResult.TookInMillis, searchResult.TotalHits())
  if searchResult.TotalHits() > 0 {
    // 创建集合 用来存放
    var userList []UserInfo
    for _, i := range searchResult.Each(reflect.TypeOf(UserInfo{})) {
      tmp := i.(UserInfo)
      userList = append(userList, tmp)
    }
    fmt.Printf("查询结果:%v\n", userList)
  }
}

范围查询 类似于 某个值 >‘xx’ and 某个值 < ‘xxx’

// 范围查询
func TestSearchByRange(t *testing.T) {
  client, _ := connectEs()
  // >  <
  //elastic.NewRangeQuery("age").Gt("19").Lt("20")
  // >=  <=
  rangeQuery := elastic.NewRangeQuery("age").Gte("18").Lte("20")
  searchResult, err := client.Search().Index(indexName).Query(rangeQuery).Do(context.Background())
  if err != nil {
    t.Errorf("查询失败:%s", err)
    return
  }
  fmt.Printf("查询消耗时间 %d ms, 结果总数: %d\n", searchResult.TookInMillis, searchResult.TotalHits())
  if searchResult.TotalHits() > 0 {
    // 创建集合 用来存放
    var userList []UserInfo
    for _, i := range searchResult.Each(reflect.TypeOf(UserInfo{})) {
      tmp := i.(UserInfo)
      userList = append(userList, tmp)
    }
    fmt.Printf("查询结果:%v\n", userList)
  }

}

bool 组合查询

must 条件

// bool 组合查询
// 1 must 条件
func TestSearchMust(t *testing.T) {
  client, _ := connectEs()
  boolQuery := elastic.NewBoolQuery()
  // 创建term
  termQuery := elastic.NewTermQuery("name", "fjj")
  matchQuery := elastic.NewMatchQuery("age", 18)

  // 设置bool查询的must条件, 组合了两个子查询
  // 表示搜索匹配name=fjj且age匹配18
  boolQuery.Must(termQuery, matchQuery)

  searchResult, err := client.Search().
    Index(indexName).        // 设置索引名
    Query(boolQuery).        // 设置查询条件
    From(0).                 // 设置分页参数 - 起始偏移量，从第0行记录开始
    Size(10).                // 设置分页参数 - 每页大小
    Do(context.Background()) // 执行请求
  if err != nil {
    t.Errorf("查询失败:%s", err)
    return
  }
  fmt.Printf("查询消耗时间 %d ms, 结果总数: %d\n", searchResult.TookInMillis, searchResult.TotalHits())
  if searchResult.TotalHits() > 0 {
    // 创建集合 用来存放
    var userList []UserInfo
    for _, i := range searchResult.Each(reflect.TypeOf(UserInfo{})) {
      tmp := i.(UserInfo)
      userList = append(userList, tmp)
    }
    fmt.Printf("查询结果:%v\n", userList)
  }

}

must_not

// must_not
func TestSearchMustNot(t *testing.T) {
  client, _ := connectEs()
  // 创建bool
  boolQuery := elastic.NewBoolQuery()
  // 创建term
  termQuery := elastic.NewTermQuery("age", 18)
  // 设置条件
  query := boolQuery.MustNot(termQuery)
  searchResult, err := client.Search().Index(indexName).Query(query).Do(context.Background())
  if err != nil {
    t.Errorf("查询失败:%s", err)
    return
  }
  fmt.Printf("查询消耗时间 %d ms, 结果总数: %d\n", searchResult.TookInMillis, searchResult.TotalHits())
  if searchResult.TotalHits() > 0 {
    // 创建集合 用来存放
    var userList []UserInfo
    for _, i := range searchResult.Each(reflect.TypeOf(UserInfo{})) {
      tmp := i.(UserInfo)
      userList = append(userList, tmp)
    }
    fmt.Printf("查询结果:%v\n", userList)
  }
}

should 条件

// should 类似与OR
func TestSearchShould(t *testing.T) {
  client, _ := connectEs()
  // 创建bool查询
  boolQuery := elastic.NewBoolQuery().Must()

  // 创建term
  termQuery := elastic.NewTermQuery("name", "fjj")
  matchQuery := elastic.NewMatchQuery("age", 18)

  // 设置bool查询的should条件, 组合了两个子查询
  // 表示搜索匹配name=fjj且age匹配18
  boolQuery.Should(termQuery, matchQuery)
  searchResult, err := client.Search().
    Index(indexName).        // 设置索引名
    Query(boolQuery).        // 设置查询条件
    From(0).                 // 设置分页参数 - 起始偏移量，从第0行记录开始
    Size(10).                // 设置分页参数 - 每页大小
    Do(context.Background()) // 执行请求
  if err != nil {
    t.Errorf("查询失败:%s", err)
    return
  }
  fmt.Printf("查询消耗时间 %d ms, 结果总数: %d\n", searchResult.TookInMillis, searchResult.TotalHits())
  if searchResult.TotalHits() > 0 {
    // 创建集合 用来存放
    var userList []UserInfo
    for _, i := range searchResult.Each(reflect.TypeOf(UserInfo{})) {
      tmp := i.(UserInfo)
      userList = append(userList, tmp)
    }
    fmt.Printf("查询结果:%v\n", userList)
  }
}

删除

// 通过Id 删除
func TestDelById(t *testing.T) {
  client, _ := connectEs()
  response, err := client.Delete().Index(indexName).Id("cxDhC4YBndr2knWC-YJt").Do(context.Background())
  if err != nil {
    t.Errorf("删除失败:%s", err)
    return
  }
  fmt.Println("删除成功: ", response.Result)
}

根据条件删除

// 根据条件删除
func TestDelByWhere(t *testing.T) {
  client, _ := connectEs()
  scrollResponse, err := client.DeleteByQuery(indexName).Query(elastic.NewTermQuery("name", "mjd")).ProceedOnVersionConflict().Do(context.Background())
  if err != nil {
    t.Errorf("删除失败:%s", err)
    return
  }
  fmt.Println("删除成功: ", scrollResponse.Deleted)
}