GoLang学习之路，对Elasticsearch的使用，一文足以（包括泛型使用思想）（二）

书写上回，上回讲到，Elasticsearch的使用前提即：语法，表结构，使用类型结构等。要学这个必须要看前面这个：GoLang学习之路，对Elasticsearch的使用，一文足以（包括泛型使用思想）（一），因为这篇是基础！！！！！！！

使用ElasticSearch

使用前提

1. 必须要有一个ElasticSearch服务器
2. 必须要有一个可视化工具
3. 安装API包，"github.com/elastic/go-elasticsearch/v8"

import "github.com/elastic/go-elasticsearch/v8"

但是这个包下面其实还有一些包，这些包非常的重要。当时我在使用的时候，根本不知道，走了不少的弯路的，找了官网的文档，又找了一些博客，都没有详细的说明情况和要点。要不就少些，要不就只把部分给列出来。但是现在我将这些无私的奉献给各位。

因为这个v8的包非常的多，所以很难将所有的放进去。这里我做一些解释：

1. 客户端：
   • 调用NewDefaultClient()和NewClient(cfg Config)方法会返回一个普通客户端
     • NewDefaultClient() 不需要去配置链接时的配置参数，默认参数链接，并返回一个普通客户端
     • NewClient(cfg Config)需要按照总共需要的配置需求去配置参数，并返回一个普通客户端
   • 调用NewTypedClient(cfg Config)会返回一个属性客户端（相比普通客户端强大，但是有局限，后面再说）
2. 工具包：
   • 这个工具包主要是普通客户端进行调用的，使用的范围是对于批量处理数据的情况
3. 参数类型包：
   • 我们在对ElasticSearch进行处理的时候会有很多中情况：
     • 首先是对于语法的选择，ElasticSearch有独属于他自己的一套语法。
     • 查询时会有很多选择，比如对于字段是模糊查询，还是精确查询，还是对地图进行查询。这些参数都有，也有对于AI进行处理的参数。（建议下一个翻译软件，去看看。那个参数太多了。。。也就是说功能非常齐全）

…很多内容在GoLang学习之路，对Elasticsearch的使用，一文足以（包括泛型使用思想）（一）

接下来正式开始

使用API实现对Elasticsearch的增删改查

为了实现这些CRUD，我总结了几个基本的使用步骤。（可以不按我这个创建客户端）

1. 创建客户端
   • 生成yaml配置文件
   • 读取配置文件信息，并保存到客户端上
2. 创建索引结构
3. 插入数据
4. 然后调用API

创建客户端

根据上面所说，客户端在创建的时候，分为两种，一种为普通客户端，一种是属性客户端。而后者的功能更为强大。但是前者的某些功能，属性客户端是没办法的。比如批量处理数据（bulk）

在实际的生产中我们需要创建两个客户端，以便我们在需求变化中获取主动权。

创建yaml文件

文件名： config.yaml
文件中的参数按自己配，千万别一样，你们是连不上 的。

es:adders:- http://8.210.237.26:9200username: elasticpassword:  +Svn3a*I*b2xxbCe9

yaml 中为何要实现数组结构，其本质是，Elasticsearch为了给以后分布式扩展提供渠道。到时候只要将IP地址，填充到配置文件就可以了

创建客户端

建议可以看看配置方法中的源码。

import (myElasticSearch "elasticsearch/common/esll""github.com/elastic/go-elasticsearch/v8""net""net/http""time"
)type ESConfig struct {Adders   []string `mapstructure:"adders" json:"adders" yaml:"adders"`Password string   `mapstructure:"password" json:"password" yaml:"password"`Username string   `mapstructure:"username" json:"username" yaml:"username"`
}func NewES(config *Config) *myElasticSearch.ElasticSearch {//强化版客户端clientType, err := elasticsearch.NewTypedClient(elasticsearch.Config{Addresses: config.ElasticSearch.Adders,Username:  config.ElasticSearch.Username,Password:  config.ElasticSearch.Password,Transport: &http.Transport{//每个host的idle状态的最大连接数目MaxConnsPerHost: 10,//发送完request后等待serve response的时间ResponseHeaderTimeout: 3 * time.Second,//(net.Conn, error) 创建未加密的tcp连接DialContext: (&net.Dialer{Timeout: time.Second}).DialContext,//连接保持idle状态的最大时间，超时关闭pconn// todo 看需求是否使用tls证书链接/*TLSClientConfig: &tls.Config{MaxVersion:         tls.VersionTLS11,InsecureSkipVerify: true,},*/},EnableDebugLogger: true,})if err != nil {panic("ElasticSearch clientType connect ping failed:" + err.Error())}//一般客户端client, err := elasticsearch.NewClient(elasticsearch.Config{Addresses: config.ElasticSearch.Adders,Username:  config.ElasticSearch.Username,Password:  config.ElasticSearch.Password,Transport: &http.Transport{//每个host的idle状态的最大连接数目MaxConnsPerHost: 10,//发送完request后等待serve response的时间ResponseHeaderTimeout: 3 * time.Second,//(net.Conn, error) 创建未加密的tcp连接DialContext: (&net.Dialer{Timeout: time.Second}).DialContext,//连接保持idle状态的最大时间，超时关闭pconn// todo 看需求是否使用tls证书链接/*TLSClientConfig: &tls.Config{MaxVersion:         tls.VersionTLS11,InsecureSkipVerify: true,},*/},EnableDebugLogger: true,})if err != nil {panic("ElasticSearch client connect ping failed:" + err.Error())}return &myElasticSearch.ElasticSearch{ClientTyped: clientType,Client:      client,}
}

将配置文件加载到客户端对象中

viper，这个读取配置文件的工具包：详细请看：文章

import ("fmt""github.com/fsnotify/fsnotify""github.com/go-playground/validator/v10""github.com/google/wire""github.com/spf13/viper"
)type Config struct {ElasticSearch *ESConfig `mapstructure:"es" validate:"required"`
}var Cfg *Configfunc ProvideConfig() *Config {var cfg Configv := viper.New()//索引配置文件位置v.SetConfigName("config.yaml")v.AddConfigPath("./")v.SetConfigType("yaml")err := v.ReadInConfig()if err != nil {panic(fmt.Errorf("open error of config file:%s", err))}//监视器v.WatchConfig()v.OnConfigChange(func(in fsnotify.Event) {fmt.Println("config file changed:", in.Name)err := v.Unmarshal(&cfg)if err != nil {fmt.Println(err)}})//反序列化if err := v.Unmarshal(&cfg); err != nil {panic(fmt.Errorf("fatal error config file : %s", err))}vs := validator.New()//校验结构err = vs.Struct(&cfg)if err != nil {panic(err)}Cfg = &cfgreturn &cfg
}

创建索引结构

定义索引结构

在 esll.go 文件中写入

const MappingTpl = `{"mappings":{"properties":{"categoryId":     { "type": "long" },"productName":   {"type": "keyword" },"masterPic":   {"type": "text"},"desc":   {"type": "keyword" },"price":    { "type": "long"},"startProvinceCode":  {"type": "text" },"startCityCode":         {"type": "text" },"update_time":  { "type": "long"},"create_time":  { "type": "long"}}}}`

定义客户端结构体

包：package esll

type ElasticSearch struct {ClientTyped *elasticsearchV8.TypedClientClient      *elasticsearchV8.Client
}

这里强调说明一下。这里为什么要用两个客户端？因为对于真正的实际运用中会有各种各样的问题出现，不仅会有一个一个查询，一个一个插入的情况，更会有一批一批的查询，插入的。所以这里的客户端对应的都会各有不同。

ClientTyped：功能强大，但是不支持批量处理
Client ：调用复杂，但是支持批量处理

定义创建索引结构的方法

包：package esll

// CreateIndex 创建所用的索引结构
func (e *ElasticSearch) CreateIndex(ctx context.Context, indexName string, mappings string) error {mapping := types.NewTypeMapping()err := mapping.UnmarshalJSON([]byte(mappings))if err != nil {return err}_, err = e.ClientTyped.Indices.Exists(indexName).Do(ctx)if err != nil {log.Printf("索引已经存在")return err}_, err = e.ClientTyped.Indices.Create(indexName).Mappings(mapping).Do(ctx)if err != nil {log.Printf("索引创建失败")return err}return nil
}

写一个测试方法

func TestMepping(t *testing.T) {ctx, _ := context.WithTimeout(context.Background(), 50*time.Second)cfg := config.ProvideConfig()client := config.NewES(cfg)mapping := types.NewTypeMapping()err := mapping.UnmarshalJSON([]byte(esll.MappingTpl))if err != nil {return}_, err = client.ClientTyped.Indices.Create("test2").Mappings(mapping).Do(ctx)if err != nil {fmt.Println(err)}
}

这里的types文件，是参数的文件，具体可以看看源码详情，根据需求选择

插入一条数据的方法

// IndexDocument 创建一条索引进入文档
func (e *ElasticSearch) IndexDocument(ctx context.Context, indexName string, document interface{}) error {do, err := e.ClientTyped.Index(indexName).Document(document ).Do(ctx)result := do.Resultfmt.Println(result)if err != nil {log.Printf("创建索引文档失败：%s", err)return err}return nil
}

判断是否存在索引，不存在就创建一个

// IsExists 是否存在索引,不存在就创建一个
func (e *ElasticSearch) IsExists(ctx context.Context, indexName string, mappings string) error {_, err2 := e.ClientTyped.Indices.Exists(indexName).Do(ctx)if err2 != nil {//不存在就重新创建一个索引err := e.CreateIndex(ctx, indexName, mappings)if err != nil {return err}}return nil
}

批量处理

方式一

func (e *ElasticSearch) IndexDocumentList(ctx context.Context, indexName string, anyList any, mapping string) error {//验证索引是否存在if err := e.IsExists(ctx, indexName, mapping); err != nil {return err}//RW := &sync.RWMutex{}slice, err := transitionSlice(anyList)if err != nil {return err}buf := buffer(ctx, e, indexName, slice, "index")//获取当前索引下的文档个数//todo:这里诺是出现超量的索引，可以通过for循环确定索要令牌(技术上限流)，或者通过协程处理//写入缓存中，并绑定索引,//转换成json格式//结果我发现这个官方已经实现了。。。。bulk, err := e.Client.Bulk(bytes.NewReader(buf.Bytes()),e.Client.Bulk.WithIndex(indexName),e.Client.Bulk.WithContext(ctx),e.Client.Bulk.WithRefresh("true"))//先关闭缓存defer bulk.Body.Close()if err != nil {log.Fatal("ElasticSearch 批量写入 失败：", err)return err}return nil
}

// 上传的缓存逻辑
func buffer(ctx context.Context, client *ElasticSearch, indexName string, slice []any, CRUD string) bytes.Buffer {c, _ := client.ClientTyped.Cat.Count().Index(indexName).Do(ctx)num, _ := strconv.Atoi(*c[0].Count)//创建缓存var buf bytes.Bufferfor i := num; i < len(slice)+num; i++ {index := []byte(fmt.Sprintf(`{ "%s" : { "_id" : "%d" } }%s`, CRUD, i, "\n"))//这里可以优化通过算法插入datas, _ := json.Marshal(slice[i-num])datas = append(datas, "\n"...)buf.Grow(len(index) + len(datas))buf.Write(index)buf.Write(datas)}return buf}

// todo 数量过多的话可以通过，三种方式，一种通过创建协程，一种通过二叉树递归的方式，另一种通过创建协程加递归的方式
func transitionSlice(anyl any) ([]any, error) {val, ok := isSlice(anyl)if !ok {return nil, errors.New("切片转换失败")}sliceLen := val.Len()list := make([]any, sliceLen)for i := 0; i < sliceLen; i++ {list[i] = val.Index(i).Interface()}return list, nil
}

// 判断是否为切片类型
func isSlice(anySlice any) (val1 reflect.Value, ok bool) {val := reflect.ValueOf(anySlice)if val.Kind() == reflect.Slice {ok = true}val1 = valreturn val1, ok
}

测试

func TestDuck2(t *testing.T) {var (buf bytes.Bufferres *esapi.Responseerr error)cfg := config.ProvideConfig()client := config.NewES(cfg).Clientfor j := 1; j <= 1000; j++ {meta := []byte(fmt.Sprintf(`{ "index" : { "_id" : "%d" } }%s`, j, "\n"))data := []byte(`{"content":"` + strings.Repeat("ABC", 100) + `"}`)data = append(data, "\n"...)buf.Grow(len(meta) + len(data))buf.Write(meta)buf.Write(data)}res, err = client.Bulk(bytes.NewReader(buf.Bytes()), client.Bulk.WithIndex("test"), client.Bulk.WithRefresh("true"))if err != nil {t.Fatalf("Failed to index data: %s", err)}res.Body.Close()if res.IsError() {t.Fatalf("Failed to index data: %s", res.Status())}
}

方法二

unc (e *ElasticSearch) UpdateDocumentList(ctx context.Context, indexName *string, anyList any, typeBulk string, mappings string) error {//验证索引是否存在if err := e.IsExists(ctx, *indexName, mappings); err != nil {return err}slice, err := transitionSlice(anyList)if err != nil {return err}start := time.Now().UTC()//设置批量配置文件indexer, err := esutil.NewBulkIndexer(esutil.BulkIndexerConfig{Index:         *indexName,Client:        e.Client,NumWorkers:    5,FlushBytes:    1024000,FlushInterval: 30 * time.Second,})if err != nil {return err}i := 1002//将数据一条一条塞入缓存中for _, data := range slice {marsha, _ := json.Marshal(data)m := fmt.Sprintf(`%s`, marsha)i++doc := esutil.BulkIndexerItem{Index:      *indexName,Action:     typeBulk,DocumentID: strconv.Itoa(i),Body:       strings.NewReader(m),OnSuccess: func(ctx context.Context, item esutil.BulkIndexerItem, item2 esutil.BulkIndexerResponseItem) {fmt.Printf("[%d] %s test/%s", item2.Status, item2.Result, item.DocumentID)},OnFailure: func(ctx context.Context, item esutil.BulkIndexerItem, item2 esutil.BulkIndexerResponseItem, err error) {if err != nil {fmt.Printf(" ERROR: %s \n", err)} else {fmt.Printf("ERROR: %s: %s \n", item2.Error.Type, item2.Error.Reason)}},}err := indexer.Add(ctx, doc)if err != nil {fmt.Println(" bulk upsert Add doc fail,", err)}}stats := indexer.Stats()fmt.Println(strings.Repeat("-", 80))dur := time.Since(start)m := int64(1000.0 / float64(dur/time.Millisecond) * float64(stats.NumFlushed))if stats.NumFailed > 0 {fmt.Printf("[%s.bulk:%s]总数据[%d]行，其中失败[%d]， 耗时 %v (速度：%d docs/秒)\n",*indexName,typeBulk,stats.NumAdded,stats.NumFailed,dur.Truncate(time.Millisecond), m)} else {fmt.Printf("[%s.bulk:%s]处理数据[%d]行，耗时%v (速度：%d docs/秒)\n",*indexName,typeBulk,stats.NumAdded,dur.Truncate(time.Millisecond), m)}err = indexer.Close(ctx)//如果没有关闭就需要循环关闭直到彻底关闭if err != nil {go func(ctx2 context.Context) {for {err = indexer.Close(ctx2)if err != nil {return}}}(ctx)}return nil
}

方式三

func (e *ElasticSearch) Findne(ctx context.Context, indexName string, queryStr any, size uint, offset uint) error {if err, _ := e.ClientTyped.Indices.Exists(indexName).Do(ctx); err == false {return fmt.Errorf("该索引不存在，不能查找：%s", err)}typeList := reflect.TypeOf(queryStr)if typeList.Kind() == reflect.Ptr {typeList = typeList.Elem()}val := reflect.ValueOf(queryStr)if val.Kind() == reflect.Ptr {val = val.Elem()}var name stringvar value stringquery := make(map[string]types.MatchQuery, typeList.NumField())var chouse boolchouse = truevar dol types.HitsMetadatafor i := 0; i < typeList.NumField(); i++ {que := &types.MatchQuery{Lenient:             &chouse,FuzzyTranspositions: &chouse,}name = typeList.Field(i).Namevalue = val.FieldByName(name).String()que.Query = valuequery[name] = *quedo, _ := e.ClientTyped.Search().Index(indexName).Query(&types.Query{Match: map[string]types.MatchQuery{"price": {Query: "123456"},},},).From(int(offset)).Size(int(size)).Do(ctx)dol = do.Hits}mapp["*esll.ProductES"] = &ProductES{}m := mappfmt.Println(m)//获取类型typeLs := reflect.TypeOf(queryStr)nam := typeLs.String()fmt.Println(nam)//从map中找到对应的结构体key := mapp[typeLs.String()]list := make([]any, 0)//深拷贝/*valo := &keykey2 := *valokey3 := &key2*/var co anyfor i := 0; i < len(dol.Hits); i++ {co = deepcopy.Copy(key)//转换为json字符数组marshalJSON, _ := dol.Hits[i].Source_.MarshalJSON()//解码并绑定_ = json.Unmarshal(marshalJSON, &co)list = append(list, co)}return nil
}

查询

var mapp = make(map[string]any, 0)func (e *ElasticSearch) FindOne(ctx context.Context, searchStruct *SearchStruct) (map[string]any, error) {do, err := e.ClientTyped.Search().Index(searchStruct.IndexName).Query(&types.Query{MatchAll: &types.MatchAllQuery{Boost:      &searchStruct.Boost,QueryName_: &searchStruct.FieldName,},},).From(searchStruct.Form).Size(searchStruct.Size).Do(ctx)if err != nil {return nil, err}//name := make(map[string]any)marshalJSON, err := do.Hits.Hits[0].Source_.MarshalJSON()if err != nil {return nil, err}var p ProductES_ = json.Unmarshal(marshalJSON, &p)fmt.Println(p)//listMap := make(map[string]any)//for i := 0; i < len(do.Hits.Hits); i++ {//	structName := name["structName"]//	stringE, _ := ToStringE(structName)//	structs := mapp[stringE]//	structl := deepcopy.Copy(structs)//	_ = json.Unmarshal(marshalJSON, &structl)//	toStringE, _ := ToStringE(i)//	listMap[toStringE] = stringE//}//return nil, err}// ToStringE 字符串转换工具
func ToStringE(i any) (string, error) {i = indirectToStringerOrError(i)switch s := i.(type) {case string:return s, nilcase bool:return strconv.FormatBool(s), nilcase float64:return strconv.FormatFloat(s, 'f', -1, 64), nilcase float32:return strconv.FormatFloat(float64(s), 'f', -1, 32), nilcase int:return strconv.Itoa(s), nilcase int64:return strconv.FormatInt(s, 10), nilcase int32:return strconv.Itoa(int(s)), nilcase int16:return strconv.FormatInt(int64(s), 10), nilcase int8:return strconv.FormatInt(int64(s), 10), nilcase uint:return strconv.FormatUint(uint64(s), 10), nilcase uint64:return strconv.FormatUint(uint64(s), 10), nilcase uint32:return strconv.FormatUint(uint64(s), 10), nilcase uint16:return strconv.FormatUint(uint64(s), 10), nilcase uint8:return strconv.FormatUint(uint64(s), 10), nilcase json.Number:return s.String(), nilcase []byte:return string(s), nilcase template.HTML:return string(s), nilcase template.URL:return string(s), nilcase template.JS:return string(s), nilcase template.CSS:return string(s), nilcase template.HTMLAttr:return string(s), nilcase nil:return "", nilcase fmt.Stringer:return s.String(), nilcase error:return s.Error(), nildefault:return "", fmt.Errorf("unable to cast %#v of type %T to string", i, i)}
}var (errorType       = reflect.TypeOf((*error)(nil)).Elem()fmtStringerType = reflect.TypeOf((*fmt.Stringer)(nil)).Elem()
)// Copied from html/template/content.go.
// indirectToStringerOrError returns the value, after dereferencing as many times
// as necessary to reach the base type (or nil) or an implementation of fmt.Stringer
// or error,
func indirectToStringerOrError(a any) any {if a == nil {return nil}v := reflect.ValueOf(a)for !v.Type().Implements(fmtStringerType) && !v.Type().Implements(errorType) && v.Kind() == reflect.Pointer && !v.IsNil() {v = v.Elem()}return v.Interface()
}