Go 创建一个监视系统，通过启动协程并使用上下文控制

电玩女神 2023-10-12 09:15 3阅读 0赞

#### 阅读目录 ####

*  Go 创建一个监视系统，通过启动协程并使用上下文控制

## Go 创建一个监视系统，通过启动协程并使用上下文控制 ##

package main
    
    import (
    	"context"
    	"errors"
    	"fmt"
    	"log"
    	"time"
    
    	"golang.org/x/sync/errgroup"
    )
    
    // 定义 MiraMenu 结构体，其中包含一个函数类型的字段 startWatch
    type MiraMenu struct {
        
    	startWatch func(context.Context, string) error
    }
    
    // MiraMenu 结构体定义了一个方法 SetWatchStart，用于设置 startWatch 字段的值
    func (s *MiraMenu) SetWatchStart(starter func(context.Context, string) error) {
        
    	s.startWatch = starter
    }
    
    // MiraWatcher 结构体，其中包含两个通道类型的字段 Rx 和 Tx，用于协程之间的通信
    type MiraWatcher struct {
        
    	Rx chan interface{
        }
    	Tx chan interface{
        }
    }
    
    // NewWatcher 函数，用于创建一个新的 MiraWatcher 实例，并初始化其中的通道
    func NewWatcher() *MiraWatcher {
        
    	return &MiraWatcher{
        
    		Rx: make(chan interface{
        }),
    		Tx: make(chan interface{
        }),
    	}
    }
    
    /*
    MiraWatcher 结构体的 Start 方法，该方法启动一个定时器，
    并在每秒触发的操作中检查后台服务是否正常运行。
    */
    func (w *MiraWatcher) Start(ctx context.Context, LC string) error {
        
    	ticker := time.NewTicker(time.Second) // 每秒触发的定时器
    	after := time.After(time.Second * 5)  // 5秒后触发的定时器
    
    	w.WatchDaemon(ctx, LC)
    
    	for {
        
    		select {
        
    		case <-ticker.C:
    			fmt.Println("每秒触发的操作")
    		case <-after:
    			err := errors.New("后台服务未正常运行")
    			fmt.Println(err)
    			w.Tx <- err
    			return err
    		}
    	}
    }
    
    /*
    MiraWatcher 结构体的 WatchDaemon 方法，
    该方法用于启动守护协程并建立管道通信。
    在建立管道过程中进行重试，直到建立成功或达到重试次数上限。
    */
    func (w *MiraWatcher) WatchDaemon(ctx context.Context, LC string) {
        
    	watcher, cancelWatch, err := w.GetWatcherWithTimeout(ctx, LC, 5*time.Second)
    	defer cancelWatch()
    
    	retryCounter := 2
    	for {
        
    		if err.Error() == "Watcher is OK" {
        
    			log.Printf("Watcher 守护进程管道已建立")
    			w.Tx <- &WatcherUpEvent{
        }
    			break
    		} else if retryCounter == 0 {
        
    			err = errors.New("未能建立观察者守护进程管道: " + err.Error())
    			w.Tx <- err
    			return
    		}
    		log.Printf("无法建立 watcher 守护进程管道，1 秒后重试: " + err.Error())
    		<-time.After(time.Second * 1)
    		watcher, cancelWatch, err = w.GetWatcherWithTimeout(ctx, LC, 15*time.Second)
    		defer cancelWatch()
    		retryCounter--
    	}
    
    	defer func() {
        
    		if watcher != "" {
        
    			log.Println("Closing watcher")
    		}
    	}()
    }
    
    /*
    MiraWatcher 结构体的 GetWatcherWithTimeout 方法，
    该方法用于模拟获取 watcher 并设置超时。
    使用 errgroup 来控制两个协程的执行，一个用于获取 watcher，另一个用于设置超时。
    */
    func (w *MiraWatcher) GetWatcherWithTimeout(ctx context.Context, LC string, timeout time.Duration) (watcher string, cancelWatch context.CancelFunc, err error) {
        
    	ctx, cancelWatch = context.WithCancel(ctx)
    
    	eg, egctx := errgroup.WithContext(ctx)
    
    	eg.Go(func() error {
        
    		select {
        
    		case <-egctx.Done():
    			log.Printf("Egctx cancelled")
    			return egctx.Err()
    		default:
    			watcher = "dummy-watcher"
    			if err == nil {
        
    				err = errors.New("虚拟观察者 watcher is OK")
    			}
    			return err
    		}
    	})
    
    	eg.Go(func() error {
        
    		select {
        
    		case <-egctx.Done():
    			log.Printf("Egctx cancelled 第二个")
    			return egctx.Err()
    		case <-time.After(timeout):
    			err := errors.New("watcher up time out")
    			cancelWatch()
    			return err
    		}
    	})
    
    	err = eg.Wait()
    	return
    }
    
    type WatcherUpEvent struct{
        }
    
    var MM *MiraMenu
    var watcher *MiraWatcher
    
    /*
    main 函数中，创建了一个 MiraMenu 实例和一个 MiraWatcher 实例，并将它们关联起来。
    然后，启动一个协程来启动 Watcher，并在一段时间后取消上下文，停止 Watcher 的运行。
    */
    func main() {
        
    	MM = &MiraMenu{
        }
    	watcher = NewWatcher()
    	//将 watcher.Start 函数设置为 MiraMenu 类型实例 MM 的 startWatch 成员变量
    	MM.SetWatchStart(watcher.Start)
    
    	ctx := context.Background()
    	ctx, cancel := context.WithCancel(ctx)
    
    	go func() {
        
    		/*
    			调用 MiraMenu 类实例 MM 的 startWatch 成员变量所指向的函数，
    			并传递 ctx 和 "dummy-LC" 作为参数。
    		*/
    		err := MM.startWatch(ctx, "dummy-LC")
    		if err != nil {
        
    			log.Println("Error starting watch:", err)
    		}
    	}()
    
    	time.Sleep(10 * time.Second)
    	cancel() // 取消上下文，停止Watcher
    }

PS E:\TEXT\test> go run .\test.go
    2023/06/02 17:49:42 Egctx cancelled 第二个
    2023/06/02 17:49:42 无法建立 watcher 守护进程管道，1 秒后重试: 虚拟观察者 watcher is OK
    2023/06/02 17:49:43 Egctx cancelled 第二个
    2023/06/02 17:49:43 无法建立 watcher 守护进程管道，1 秒后重试: 虚拟观察者 watcher is OK
    2023/06/02 17:49:44 Egctx cancelled 第二个
    PS E:\TEXT\test>

当 main() 函数运行时，它执行以下操作：

1 创建了 MiraMenu 实例 MM 和 MiraWatcher 实例 watcher。  
2 通过调用 MM.SetWatchStart() 方法，将 watcher.Start 方法设置为 MiraMenu 实例的 `startWatch` 函数。

3 创建了一个 context.Context 对象 ctx，用于控制协程的生命周期。  
4 启动一个协程，其中执行 MM.startWatch(ctx, “dummy-LC”)，即调用 startWatch 函数开始监视。

5 主函数等待 10 秒后，通过调用 cancel() 方法取消 ctx，从而终止监视协程的执行。

MiraMenu 类包含一个方法 SetWatchStart()，该方法用于设置启动监视的函数。  
在本例中，它设置了 watcher.Start 作为启动监视的函数。

MiraWatcher 类包含两个通道 Rx 和 Tx，分别用于接收和发送消息。它还包含一个 Start() 方法，用于启动监视逻辑。

在 Start() 方法中，首先创建了两个定时器 ticker 和 after。然后调用 WatchDaemon() 方法启动守护进程。

WatchDaemon() 方法是监视逻辑的核心。它通过调用 GetWatcherWithTimeout() 方法获取 watcher，并进行重试，直到 watcher 建立成功或达到重试次数上限。如果 watcher 成功建立，则通过 Tx 通道发送一个 WatcherUpEvent 事件，表示监视已成功启动。最后，关闭 watcher。

GetWatcherWithTimeout() 方法使用 errgroup 控制并发执行两个协程。一个协程负责获取 watcher，另一个协程用于设置超时。在超时之前，如果成功获取 watcher，则返回 watcher 信息；否则，返回错误信息。

整个代码的目的是创建一个监视系统，通过启动协程并使用上下文控制，定期执行监视操作，并根据监视结果发送消息。这个例子中使用了定时器、协程、通道和上下文来实现监视逻辑的控制和通信。