Go 综合实战题详解（Web框架/RPC/K8s Controller）

type ShortLinkService struct {
	seq     *Snowflake
	cache   *TTLCache
	base62  *Base62Encoder
}

func (s *ShortLinkService) Generate(longURL string) (string, error) {
	// 1. 生成唯一 ID
	id := s.seq.NextID()
	
	// 2. 转换为 Base62
	shortCode := s.base62.Encode(id)
	
	// 3. 存储映射
	s.cache.Set(shortCode, longURL, 30*24*time.Hour)
	
	return shortCode, nil
}

func (s *ShortLinkService) Resolve(shortCode string) (string, error) {
	url, ok := s.cache.Get(shortCode)
	if !ok {
		return "", fmt.Errorf("short code not found")
	}
	return url.(string), nil
}

type Base62Encoder struct {
	charset string
}

func NewBase62Encoder() *Base62Encoder {
	return &Base62Encoder{charset: "0123456789ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz"}
}

func (b *Base62Encoder) Encode(n int64) string {
	if n == 0 {
		return string(b.charset[0])
	}
	var result []byte
	for n > 0 {
		result = append([]byte{b.charset[n%62]}, result...)
		n /= 62
	}
	return string(result)
}

func (b *Base62Encoder) Decode(s string) int64 {
	var result int64
	for _, c := range s {
		result = result*62 + int64(bytes.IndexByte([]byte(b.charset), byte(c)))
	}
	return result
}

type DelayQueue struct {
	client *redis.Client
	key    string
}

func NewDelayQueue(client *redis.Client, key string) *DelayQueue {
	return &DelayQueue{client: client, key: key}
}

func (dq *DelayQueue) Enqueue(item interface{}, delay time.Duration) error {
	score := float64(time.Now().Add(delay).UnixNano())
	data, _ := json.Marshal(item)
	return dq.client.ZAdd(dq.key, redis.Z{Score: score, Member: data}).Err()
}

func (dq *DelayQueue) Dequeue(timeout time.Duration) (interface{}, error) {
	ctx, cancel := context.WithTimeout(context.Background(), timeout)
	defer cancel()
	
	for {
		select {
		case <-ctx.Done():
			return nil, ctx.Err()
		default:
			items, err := dq.client.ZRangeByScoreWithScores(ctx, dq.key, 
				&redis.ZRangeBy{Min: "0", Max: fmt.Sprintf("%d", time.Now().UnixNano())})
			if err != nil || len(items) == 0 {
				time.Sleep(100 * time.Millisecond)
				continue
			}
			
			for _, item := range items {
				dq.client.ZRem(dq.key, item.Member)
				var result interface{}
				json.Unmarshal(item.Member.([]byte), &result)
				return result, nil
			}
		}
	}
}

type TimerWheel struct {
	wheels []*TickWheel
	size   int
}

type TickWheel struct {
	buckets [][]DelayedTask
	index   int
}

type DelayedTask struct {
	task     func()
	execTime int64
}

func NewTimerWheel(size int, ticks int) *TimerWheel {
	wheels := make([]*TickWheel, 0)
	prevTicks := 1
	for i := 0; i < size; i++ {
		wheels = append(wheels, &TickWheel{
			buckets: make([][]DelayedTask, ticks),
			ticks:   ticks,
		})
		prevTicks *= ticks
	}
	return &TimerWheel{wheels: wheels, size: size}
}

func (tw *TimerWheel) Schedule(task func(), delay time.Duration) {
	execTime := time.Now().Add(delay).UnixMilli()
	
	// 分配到对应的 wheel 和 bucket
	slot := execTime % int64(tw.size)
	bucket := execTime / int64(tw.size)
	
	tw.wheels[slot].buckets[bucket%int64(tw.wheels[slot].ticks)] = 
		append(tw.wheels[slot].buckets[bucket%int64(tw.wheels[slot].ticks)], 
			DelayedTask{task: task, execTime: execTime})
}

func (tw *TimerWheel) Start() {
	ticker := time.NewTicker(1 * time.Second)
	defer ticker.Stop()
	
	for range ticker.C {
		tw.tick()
	}
}

func (tw *TimerWheel) tick() {
	for _, wheel := range tw.wheels {
		idx := int(time.Now().UnixMilli()) % wheel.ticks
		for _, task := range wheel.buckets[idx] {
			task.task()
		}
		wheel.buckets[idx] = nil
	}
}

type Controller struct {
	informer cache.Informer
	workqueue workqueue.RateLimitingInterface
	handler  func(interface{}) error
}

func NewController(informer cache.Informer, handler func(interface{}) error) *Controller {
	c := &Controller{
		informer: informer,
		workqueue: workqueue.NewRateLimitingQueue(workqueue.DefaultControllerRateLimiter()),
		handler:  handler,
	}
	
	// 添加事件处理器
	c.informer.AddEventHandler(cache.ResourceEventHandlerFuncs{
		AddFunc:    c.enqueue,
		UpdateFunc: func(_, newObj interface{}) { c.enqueue(newObj) },
		DeleteFunc: c.enqueue,
	})
	
	return c
}

func (c *Controller) enqueue(obj interface{}) {
	key, _ := cache.MetaNamespaceKeyFunc(obj)
	c.workqueue.Add(key)
}

func (c *Controller) Run(workerCount int, stopCh <-chan struct{}) {
	defer c.workqueue.ShutDown()
	
	go c.informer.Run(stopCh)
	if !cache.WaitForCacheSync(stopCh, c.informer.HasSynced) {
		klog.ErrorS(nil, "Failed to wait for caches to sync")
		return
	}
	
	for i := 0; i < workerCount; i++ {
		go wait.Until(c.worker, time.Second, stopCh)
	}
	
	<-stopCh
}

func (c *Controller) worker() {
	for c.processNextItem() {
	}
}

func (c *Controller) processNextItem() bool {
	key, quit := c.workqueue.Get()
	if quit {
		return false
	}
	defer c.workqueue.Done(key)
	
	if err := c.handler(key); err != nil {
		c.workqueue.AddRateLimit(key)
		klog.ErrorS(err, "Failed to reconcile", "key", key)
		return true
	}
	
	c.workqueue.Forget(key)
	return true
}