go工具方法-elasticsearch

package es

import (
   "context"
   "fmt"
   "github.com/olivere/elastic"
   "go-pkg/pkg/cfg"
)
//可参考https://github.com/ashion89/go-pkg/tree/master/pkg/es中测试
var client *elastic.Client
var ctx = context.Background()
var config  = cfg.GetConfig()


//初始化es
func Init() error {
   var url = config.Es.Url
   var user = config.Es.User
   var password = config.Es.Password
 
   var err error
   client, err = elastic.NewClient(
      elastic.SetURL(url),
      elastic.SetBasicAuth(user, password),
      elastic.SetSniff(false),
   )
   if err != nil {
      fmt.Println("NewClient err ",err)
      return err
   }
   info, code, err := client.Ping(url).Do(ctx)
   if err != nil {
      fmt.Println("Ping err ",err)
      return err
   }
 
   if code == 200 {
      fmt.Printf("connected to es: %s ,version: %s 
", info.ClusterName, info.Version.Number)
   }
   return nil
}

//获取连接
func getClient() *elastic.Client {
   if client != nil {
      return client
   } else {
      err := Init()
      fmt.Println("init err ",err)
      return client
   }
}



//添加数据到es

import (
   "encoding/json"
   "errors"
   "fmt"
   "github.com/olivere/elastic"
   "go-pkg/pkg/util"
   "strconv"
)

const EsStdIdxCount = 10
const EsMsgIdxName = "message_"
const ALIASE = "_aliase" //别名
const TYPE = "_doc"

//account分库
func AddMessage(message map[string]interface{}, account string) error {
   idx := util.GetHashCode(account, EsStdIdxCount)
   id, _ := message["_id"].(string)
   delete(message, "_id")
   esClient := getClient()
   if esClient != nil {
      _, err := esClient.Index().Index(EsMsgIdxName + strconv.Itoa(idx) + ALIASE).Type(TYPE).Routing(account).Id(id).BodyJson(message).Do(ctx)
      if err != nil {
       
         return err
      }
      return nil
   } else {
     
      return errors.New("es client context deadlineexceeded")
   }
}

//查询数据

func SearchMessageAll(account string, msgContent string) (data []map[string]interface{}, err error) {
   idx := util.GetHashCode(account, EsStdIdxCount)
   qesMatch := elastic.NewMatchPhraseQuery("name", msgContent).Analyzer("standard") //匹配内容
   query := elastic.NewBoolQuery().Must(qesMatch)

   searchResult, err := getClient().Search().TrackTotalHits(false).
      Index(EsMsgIdxName + strconv.Itoa(idx) + ALIASE).
      Type(TYPE).Routing(account).
      Query(query).
      From(0).Size(1000).
      Do(ctx)
   if err != nil {
      fmt.Printf("-----search  doc from es err : %v ", err)
      return
   }
   for _, hit := range searchResult.Hits.Hits {
      var t map[string]interface{}
      err := json.Unmarshal(*hit.Source, &t) 
      if err != nil {
         fmt.Printf("search  doc es Unmarshal err: %v ", err)
      }
      data = append(data, t)
   }
   return data, nil
}

//util中方法

package util
import (
   "hash/crc32"
)

func GetHashCode(str string, count int) int {
   v := crc32.ChecksumIEEE([]byte(str))
   if v < 0 {
      v = -v
   }
   return int(v) % count
}