• 哨兵（centennial）負責接待客人，直接與調用方對接。
• 哨兵的核心組件包括service HUB和connection pool。
• service HUB用于與服務中心通信，獲取可提供服務的節點信息。
• connection pool用于緩存與index worker的連接，避免每次搜索時重新建立連接。
• 連接池初始化為空map。
• 提供函數獲取指定endpoint的GRPC連接。
• 函數首先檢查本地緩存中是否有可用連接，若無則創建新連接。
• 創建連接時默認立即返回，可選阻塞模式直到連接可用。
• 連接建立后放入緩存并返回。
• 哨兵提供添加、刪除和搜索三個核心功能。
• 添加功能：隨機選擇一臺index worker添加新文檔。
• 刪除功能：遍歷所有endpoint，并行刪除指定文檔。
• 搜索功能：將搜索請求發送到所有endpoint，合并搜索結果。
• 使用channel進行并發搜索結果的收集。
• 上游并發寫入channel，下游讀取channel數據到切片。
• 使用wait group等待所有搜索任務完成。
• 關閉channel后仍可讀取，確保讀取到所有數據。

package index_serviceimport ("context""fmt""github.com/jmh000527/criker-search/index_service/service_hub""github.com/jmh000527/criker-search/types""github.com/jmh000527/criker-search/utils""google.golang.org/grpc""google.golang.org/grpc/connectivity""google.golang.org/grpc/credentials/insecure""sync""sync/atomic""time"
)// Sentinel 哨兵前臺，與外部系統對接的接口。
type Sentinel struct {hub      service_hub.ServiceHub // 從 Hub 中獲取 IndexServiceWorker 的集合。可以直接訪問 ServiceHub，也可能通過代理模式進行訪問。connPool sync.Map               // 與各個 IndexServiceWorker 建立的 gRPC 連接池。緩存連接以避免每次請求都重新建立連接，提升效率。
}// NewSentinel 創建并返回一個 Sentinel 實例。
//
// 參數:
//   - etcdServers: 一個字符串數組，包含了 etcd 服務器的地址。
//
// 返回值:
//   - *Sentinel: 一個新的 Sentinel 實例。
func NewSentinel(etcdServers []string) *Sentinel {return &Sentinel{// hub: GetServiceHub(etcdServers, 10), // 直接訪問 ServiceHubhub:      service_hub.GetServiceHubProxy(etcdServers, 3, 100), // 使用代理模式訪問 ServiceHubconnPool: sync.Map{},                                          // 初始化 gRPC 連接池}
}// GetGrpcConn 向指定的 endpoint 建立 gRPC 連接。
// 如果連接已經存在于緩存中且狀態可用，則直接返回緩存的連接。
// 如果連接狀態不可用或不存在，則重新建立連接并存儲到緩存中。
//
// 參數:
//   - endpoint: 要連接的 gRPC 服務的地址。
//
// 返回值:
//   - *grpc.ClientConn: 返回與 endpoint 建立的 gRPC 連接，如果連接失敗則返回 nil。
func (sentinel *Sentinel) GetGrpcConn(endpoint string) *grpc.ClientConn {v, exists := sentinel.connPool.Load(endpoint)// 連接緩存中存在if exists {conn := v.(*grpc.ClientConn)// 如果連接狀態不可用，則從連接緩存中刪除if conn.GetState() == connectivity.TransientFailure || conn.GetState() == connectivity.Shutdown {utils.Log.Printf("連接到 endpoint %s 的狀態為 %s", endpoint, conn.GetState().String())conn.Close()sentinel.connPool.Delete(endpoint)} else {return conn}}// 連接到服務，控制連接超時ctx, cancel := context.WithTimeout(context.Background(), 200*time.Millisecond)defer cancel()// 獲取 gRPC 連接// grpc.Dial 是異步連接，連接狀態為正在連接。// 如果設置了 grpc.WithBlock 選項，則會阻塞等待（等待握手成功）。// 需要注意的是，當未設置 grpc.WithBlock 時，ctx 超時控制對其無任何效果。grpcConn, err := grpc.DialContext(ctx, endpoint, grpc.WithTransportCredentials(insecure.NewCredentials()), grpc.WithBlock())if err != nil {utils.Log.Printf("連接到 %s 的 gRPC 失敗，錯誤: %s", endpoint, err.Error())return nil}utils.Log.Printf("連接到 %s 的 gRPC 成功", endpoint)// 將 gRPC 連接緩存到連接池中sentinel.connPool.Store(endpoint, grpcConn)return grpcConn
}// AddDoc 向集群中的 IndexService 添加文檔。如果文檔已存在，會先刪除舊文檔再添加新文檔。
//
// 參數:
//   - doc: 要添加的文檔，類型為 types.Document。
//
// 返回值:
//   - int: 成功添加的文檔數量。
//   - error: 如果在添加文檔時出現錯誤，返回相應的錯誤信息。
func (sentinel *Sentinel) AddDoc(doc types.Document) (int, error) {// 根據負載均衡策略，選擇一個 IndexService 節點，將文檔添加到該節點endpoint := sentinel.hub.GetServiceEndpoint(IndexService)if len(endpoint) == 0 {return 0, fmt.Errorf("未找到服務 %s 的有效節點", IndexService)}// 創建到該節點的 gRPC 連接grpcConn := sentinel.GetGrpcConn(endpoint)if grpcConn == nil {return 0, fmt.Errorf("連接到 %s 的 gRPC 失敗", endpoint)}// 創建 gRPC 客戶端并進行調用client := NewIndexServiceClient(grpcConn)affected, err := client.AddDoc(context.Background(), &doc)if err != nil {return 0, err}utils.Log.Printf("成功向 worker %s 添加 %d 個文檔", endpoint, affected.Count)return int(affected.Count), nil
}// DeleteDoc 從集群中刪除與 docId 對應的文檔，返回成功刪除的文檔數量（通常不會超過 1）。
//
// 參數:
//   - docId: 要刪除的文檔的唯一標識符。
//
// 返回值:
//   - int: 成功刪除的文檔數量。
func (sentinel *Sentinel) DeleteDoc(docId string) int {// 獲取該服務的所有 endpointsendpoints := sentinel.hub.GetServiceEndpoints(IndexService)if len(endpoints) == 0 {return 0}var n int32wg := sync.WaitGroup{}wg.Add(len(endpoints))for _, endpoint := range endpoints {// 并行地向各個 IndexServiceWorker 刪除對應的 docId 的文檔。// 正常情況下，只有一個 worker 上有該文檔。go func(endpoint string) {defer wg.Done()grpcConn := sentinel.GetGrpcConn(endpoint)if grpcConn == nil {utils.Log.Printf("連接到 %s 的 gRPC 失敗", endpoint)return}client := NewIndexServiceClient(grpcConn)affected, err := client.DeleteDoc(context.Background(), &DocId{docId})if err != nil {utils.Log.Printf("從 worker %s 刪除文檔 %s 失敗，錯誤: %s", endpoint, docId, err)return}if affected.Count > 0 {atomic.AddInt32(&n, affected.Count)utils.Log.Printf("從 worker %s 刪除文檔 %s 成功", endpoint, docId)}}(endpoint)}wg.Wait()return int(atomic.LoadInt32(&n))
}// Search 執行檢索操作，并返回文檔列表。
//
// 參數:
//   - query: 指定的檢索查詢條件，類型為 *types.TermQuery。
//   - onFlag: 開啟的標志位，類型為 uint64。
//   - offFlag: 關閉的標志位，類型為 uint64。
//   - orFlags: OR 標志位的切片，類型為 []uint64。
//
// 返回值:
//   - []*types.Document: 經過檢索的文檔列表，可能為空。
//
// 詳細描述:
//  1. 從服務中心獲取所有的 endpoints。
//  2. 使用 goroutines 并行地對每個 endpoint 執行檢索操作。
//  3. 將每個檢索結果發送到 resultChan 通道中。
//  4. 在另一個 goroutine 中，從 resultChan 通道中讀取結果，并將其存儲在 docs 切片中。
//  5. 等待所有的檢索操作完成后，關閉 resultChan，并等待從 resultChan 中讀取完所有結果。
//  6. 返回存儲的文檔列表。
func (sentinel *Sentinel) Search(query *types.TermQuery, onFlag, offFlag uint64, orFlags []uint64) []*types.Document {// 獲取該服務所有的 endpointsendpoints := sentinel.hub.GetServiceEndpoints(IndexService)if len(endpoints) == 0 {return nil}// 用于存儲檢索結果的切片和通道docs := make([]*types.Document, 0, 1000)resultChan := make(chan *types.Document, 1000)// 使用 WaitGroup 并行開啟協程去每個 endpoint 執行檢索操作var wg sync.WaitGroupwg.Add(len(endpoints))for _, endpoint := range endpoints {go func(endpoint string) {defer wg.Done()// 獲取 gRPC 連接grpcConn := sentinel.GetGrpcConn(endpoint)if grpcConn == nil {utils.Log.Printf("連接到 %s 的 gRPC 連接失敗", endpoint)return}client := NewIndexServiceClient(grpcConn)// 執行檢索請求searchResult, err := client.Search(context.Background(), &SearchRequest{Query:   query,OnFlag:  onFlag,OffFlag: offFlag,OrFlags: orFlags,})if err != nil {utils.Log.Printf("向 worker %s 執行查詢 %s 失敗，錯誤: %s", endpoint, query, err)return}if len(searchResult.Results) > 0 {utils.Log.Printf("向 worker %s 執行查詢 %s 成功，獲取到 %v 個文檔", endpoint, query, len(searchResult.Results))for _, result := range searchResult.Results {resultChan <- result}}}(endpoint)}// 啟動另一個 goroutine 從 resultChan 中獲取結果signalChan := make(chan struct{})go func() {for doc := range resultChan {docs = append(docs, doc)}// 讀取完成，通知主 goroutinesignalChan <- struct{}{}}()// 等待所有檢索操作完成wg.Wait()// 關閉 resultChan 通道close(resultChan)// 等待結果讀取完畢<-signalChanreturn docs
}// Count 獲取所有服務中的搜索條目數量。
//
// 參數:
//   - 無參數。
//
// 返回值:
//   - int: 所有服務中的文檔總數量。
//
// 詳細描述:
//  1. 從服務中心獲取所有的 endpoints。
//  2. 使用 goroutines 并行地對每個 endpoint 執行計數操作。
//  3. 將每個 worker 中的文檔數量累加到總計數中。
//  4. 等待所有計數操作完成后，返回文檔總數量。
func (sentinel *Sentinel) Count() int {var n int32// 獲取所有服務的 endpointsendpoints := sentinel.hub.GetServiceEndpoints(IndexService)if len(endpoints) == 0 {return 0}var wg sync.WaitGroupwg.Add(len(endpoints))for _, endpoint := range endpoints {go func(endpoint string) {defer wg.Done()// 獲取 gRPC 連接grpcConn := sentinel.GetGrpcConn(endpoint)if grpcConn != nil {client := NewIndexServiceClient(grpcConn)// 執行計數請求affected, err := client.Count(context.Background(), new(CountRequest))if err != nil {utils.Log.Printf("從 worker %s 獲取文檔數量失敗: %s", endpoint, err)}if affected.Count > 0 {// 累加計數atomic.AddInt32(&n, affected.Count)utils.Log.Printf("worker %s 共有 %d 個文檔", endpoint, affected.Count)}}}(endpoint)}// 等待所有計數操作完成wg.Wait()return int(atomic.LoadInt32(&n))
}// Close 關閉各個grpc client連接，關閉etcd client連接
func (sentinel *Sentinel) Close() (err error) {sentinel.connPool.Range(func(key, value any) bool {conn := value.(*grpc.ClientConn)err = conn.Close()return true})sentinel.hub.Close()return
}