client-go的workqueue详解

Table of Contents

1. 章节介绍

在介绍完Informer机制后,可以发现如果想自定义控制器非常简单,我们直接注册handler就行。但是绝大部分k8s原生控制器中,handler并没有直接处理。而是统一遵守一套:

Add , update, Del -> queue -> run -> runWorker -> syncHandler 处理的模式。

例如 namespaces控制器中:

// 1.先是定义了一个限速队列
queue:                      workqueue.NewNamedRateLimitingQueue(nsControllerRateLimiter(), "namespace"),


// 2.然后add, update都是入队列
// configure the namespace informer event handlers
    namespaceInformer.Informer().AddEventHandlerWithResyncPeriod(
        cache.ResourceEventHandlerFuncs{
            AddFunc: func(obj interface{}) {
                namespace := obj.(*v1.Namespace)
                namespaceController.enqueueNamespace(namespace)
            },
            UpdateFunc: func(oldObj, newObj interface{}) {
                namespace := newObj.(*v1.Namespace)
                namespaceController.enqueueNamespace(namespace)
            },
        },
        resyncPeriod,
    )
    
// 3.然后controller.run,启动多个协程
// Run starts observing the system with the specified number of workers.
func (nm *NamespaceController) Run(workers int, stopCh <-chan struct{}) {
  
    for i := 0; i < workers; i++ {
        go wait.Until(nm.worker, time.Second, stopCh)
    }
    <-stopCh
}

// 4. worker处理一个个数据
func (nm *NamespaceController) worker() {

    // 得到对象
        key, quit := nm.queue.Get()
        
        // 处理完对象
        defer nm.queue.Done(key)

        err := nm.syncNamespaceFromKey(key.(string))
        if err == nil {
            // no error, forget this entry and return
            nm.queue.Forget(key)
            return false
        }
}

可以看出来这一套的一个好处:

(1)利用了Indexer本地缓存机制,queue里面只包括 key就行。数据indexer都有

(2)workqueue除了一个缓冲机制外,还有着错误重试的机制

因此这一节分析一下,client-go提供了哪些workqueue

2. workerqueue介绍

client-go 的 util/workqueue 包里主要有三个队列,分别是普通队列,延时队列,限速队列,后一个队列以前一个队列的实现为基础,层层添加新功能,我们按照 Queue、DelayingQueue、RateLimitingQueue 的顺序层层拨开来看限速队列是如何实现的。

2.1 queue

2.1.1 queue接口
type Interface interface {
   Add(item interface{})  // 添加一个元素
   Len() int              // 元素个数
   Get() (item interface{}, shutdown bool) // 获取一个元素,第二个返回值和 channel 类似,标记队列是否关闭了
   Done(item interface{}) // 标记一个元素已经处理完
   ShutDown()             // 关闭队列
   ShuttingDown() bool    // 是否正在关闭
}


type Type struct {
   queue []t            // 定义元素的处理顺序,里面所有元素都应该在 dirty set 中有,而不能出现在 processing set 中
   dirty set            // 标记所有需要被处理的元素
   processing set       // 当前正在被处理的元素,当处理完后需要检查该元素是否在 dirty set 中,如果有则添加到 queue 里

   cond *sync.Cond      // 条件锁
   shuttingDown bool    // 是否正在关闭
   metrics queueMetrics
   unfinishedWorkUpdatePeriod time.Duration
   clock                      clock.Clock
}

这个 Queue 的工作逻辑大致是这样,里面的三个属性 queue、dirty、processing 都保存 items,但是含义有所不同:

  • queue:这是一个 []t 类型,也就是一个切片,因为其有序,所以这里当作一个列表来存储 item 的处理顺序。
  • dirty:这是一个 set 类型,也就是一个集合,这个集合存储的是所有需要处理的 item,这些 item 也会保存在 queue 中,但是 set 里是无序的,set 的特性是唯一。可以认为dirty就是queue的不同实现, queue是为了有序,set是为了保证元素唯一。
  • processing:这也是一个 set,存放的是当前正在处理的 item,也就是说这个 item 来自 queue 出队的元素,同时这个元素会被从 dirty 中删除。

目前看这些还有些懵,直接看看queue的核心函数。

add

从这里就可以看出来,queue函数进行了过滤。比如我更新了pod1三次。

pod1 := &v1.Pod{ObjectMeta: metav1.ObjectMeta{Name: "one", Annotations: map[string]string{"users": "ernie,bert"}}}

informer的distrube函数会发送三个更新事件,queue也会收到三个更新事件,但是queue里面只会有一个 one(pod1的key)。

为什么只需要保留一个就行?

因为indexer已经更新了,indexer的数据是最新的。所以从这里也可以看出来,使用这一套逻辑,就没有update ,add, delete等区别了。

如果我想统计一下,每个Pod变化了多少次,那就不能使用 workqueue了,必须在handler那里直接实现。

// Add marks item as needing processing.
func (q *Type) Add(item interface{}) {
    q.cond.L.Lock()
    defer q.cond.L.Unlock()
    if q.shuttingDown {
        return
    }
    
    // dirty set 中已经有了该 item,则返回
    if q.dirty.has(item) {   
        return
    }

    q.metrics.add(item)
  
  
    q.dirty.insert(item)
    // 如果正在处理,也直接返回
    if q.processing.has(item) {
        return
    }
  
  // 否则就扔进queue队列
    q.queue = append(q.queue, item)
    q.cond.Signal()
}
get

get会将元素从queue队列去列,表示这个元素,正在处理中。

dirty和queue保持一致,也会删除这个元素。

// get是从 queue队列中取出一个元素(queue中删除,dirty中删除)
// 并且标记它正在处理,
func (q *Type) Get() (item interface{}, shutdown bool) {
    q.cond.L.Lock()
    defer q.cond.L.Unlock()
    for len(q.queue) == 0 && !q.shuttingDown {
        q.cond.Wait()
    }
    if len(q.queue) == 0 {
        // We must be shutting down.
        return nil, true
    }

    item, q.queue = q.queue[0], q.queue[1:]

    q.metrics.get(item)

    q.processing.insert(item)
    q.dirty.delete(item)

    return item, false
}
done

done表明这个元素被处理完了,从processing队列删除。这里加了一个判断,如果dirty中还存在,还要将其加入 queue

为什么需要这个判断呢?

原因在于有一种请求是 itemA 正在处理,但是还没done,这个时候又来了一次 itemA。

这个时候add 逻辑中,是直接返回的,不会添加itemA到queue的。所以这里要重新添加一次


// Done marks item as done processing, and if it has been marked as dirty again
// while it was being processed, it will be re-added to the queue for
// re-processing.
func (q *Type) Done(item interface{}) {
    q.cond.L.Lock()
    defer q.cond.L.Unlock()

    q.metrics.done(item)

    q.processing.delete(item)
    // 判断dirty是否有该元素
    if q.dirty.has(item) {
        q.queue = append(q.queue, item)
        q.cond.Signal()
    }
}


2.2 DelayingQueue-延迟队列

// delayingType wraps an Interface and provides delayed re-enquing
type delayingType struct {
    Interface                     //上面的通用队列
    clock clock.Clock             // 时钟,用于获取时间
    stopCh chan struct{}          // 延时就意味着异步,就要有另一个协程处理,所以需要退出信号
    stopOnce sync.Once            // 用来确保 ShutDown() 方法只执行一次
    heartbeat clock.Ticker        // 定时器,在没有任何数据操作时可以定时的唤醒处理协程
    waitingForAddCh chan *waitFor // 所有延迟添加的元素封装成waitFor放到chan中
    metrics retryMetrics
}

type DelayingInterface interface {
    Interface
    // AddAfter adds an item to the workqueue after the indicated duration has passed
    AddAfter(item interface{}, duration time.Duration)
}
2.2.1 waitFor
type waitFor struct {
   data    t          // 准备添加到队列中的数据
   readyAt time.Time  // 应该被加入队列的时间
   index int          // 在 heap 中的索引
}

waitForPriorityQueue是一个数组,实现了最小堆,对比的就是延迟的时间。

type waitForPriorityQueue []*waitFor
// heap需要实现的接口,告知队列长度
func (pq waitForPriorityQueue) Len() int {
    return len(pq)
}
// heap需要实现的接口,告知第i个元素是否比第j个元素小
func (pq waitForPriorityQueue) Less(i, j int) bool {
    return pq[i].readyAt.Before(pq[j].readyAt) // 此处对比的就是时间,所以排序按照时间排序
}
// heap需要实现的接口,实现第i和第j个元素换
func (pq waitForPriorityQueue) Swap(i, j int) {
    // 这种语法好牛逼,有没有,C/C++程序猿没法理解~
    pq[i], pq[j] = pq[j], pq[i]
    pq[i].index = i                            // 因为heap没有所以,所以需要自己记录索引,这也是为什么waitFor定义索引参数的原因
    pq[j].index = j
}
// heap需要实现的接口,用于向队列中添加数据
func (pq *waitForPriorityQueue) Push(x interface{}) {
    n := len(*pq)                       
    item := x.(*waitFor)
    item.index = n                             // 记录索引值
    *pq = append(*pq, item)                    // 放到了数组尾部
}
// heap需要实现的接口,用于从队列中弹出最后一个数据
func (pq *waitForPriorityQueue) Pop() interface{} {
    n := len(*pq)
    item := (*pq)[n-1]
    item.index = -1
    *pq = (*pq)[0:(n - 1)]                     // 缩小数组,去掉了最后一个元素
    return item
}
// 返回第一个元素
func (pq waitForPriorityQueue) Peek() interface{} {
    return pq[0]
}

到这里就可以大概猜出来延迟队列的实现了。

就是所有添加的元素,有一个延迟时间,根据延迟时间构造一个最小堆。然后每次时间一到,从堆里面拿出来当前应该加入队列的时间。


2.2. 2 NewNamedDelayingQueue
// 这里可以传递一个名字
func NewNamedDelayingQueue(name string) DelayingInterface {
   return NewDelayingQueueWithCustomClock(clock.RealClock{}, name)
}

// 上面一个函数只是调用当前函数,附带一个名字,这里加了一个指定 clock 的能力
func NewDelayingQueueWithCustomClock(clock clock.Clock, name string) DelayingInterface {
  return newDelayingQueue(clock, NewNamed(name), name) // 注意这里的 NewNamed() 函数
}

func newDelayingQueue(clock clock.Clock, q Interface, name string) *delayingType {
   ret := &delayingType{
      Interface:       q,
      clock:           clock,
      heartbeat:       clock.NewTicker(maxWait), // 10s 一次心跳
      stopCh:          make(chan struct{}),
      waitingForAddCh: make(chan *waitFor, 1000),
      metrics:         newRetryMetrics(name),
   }

   go ret.waitingLoop() // 核心就是运行 waitingLoop
   return ret
}
2.2.3 waitingLoop
func (q *delayingType) waitingLoop() {
   defer utilruntime.HandleCrash()
   // 队列里没有 item 时实现等待用的
   never := make(<-chan time.Time)
   var nextReadyAtTimer clock.Timer
   // 构造一个优先级队列
   waitingForQueue := &waitForPriorityQueue{}
   heap.Init(waitingForQueue) // 这一行其实是多余的,等下提个 pr 给它删掉

   // 这个 map 用来处理重复添加逻辑的,下面会讲到
   waitingEntryByData := map[t]*waitFor{}
   // 无限循环
   for {
      // 这个地方 Interface 是多余的,等下也提个 pr 把它删掉吧
      if q.Interface.ShuttingDown() {
         return
      }

      now := q.clock.Now()
      // 队列里有 item 就开始循环
      for waitingForQueue.Len() > 0 {
         // 获取第一个 item
         entry := waitingForQueue.Peek().(*waitFor)
         // 时间还没到,先不处理
         if entry.readyAt.After(now) {
            break
         }
        // 时间到了,pop 出第一个元素;注意 waitingForQueue.Pop() 是最后一个 item,heap.Pop() 是第一个元素
         entry = heap.Pop(waitingForQueue).(*waitFor)
         // 将数据加到延时队列里
         q.Add(entry.data)
         // map 里删除已经加到延时队列的 item
         delete(waitingEntryByData, entry.data)
      }

      // 如果队列中有 item,就用第一个 item 的等待时间初始化计时器,如果为空则一直等待
      nextReadyAt := never
      if waitingForQueue.Len() > 0 {
         if nextReadyAtTimer != nil {
            nextReadyAtTimer.Stop()
         }
         entry := waitingForQueue.Peek().(*waitFor)
         nextReadyAtTimer = q.clock.NewTimer(entry.readyAt.Sub(now))
         nextReadyAt = nextReadyAtTimer.C()
      }

      select {
      case <-q.stopCh:
         return
      case <-q.heartbeat.C(): // 心跳时间是 10s,到了就继续下一轮循环
      case <-nextReadyAt: // 第一个 item 的等到时间到了,继续下一轮循环
      case waitEntry := <-q.waitingForAddCh: // waitingForAddCh 收到新的 item
         // 如果时间没到,就加到优先级队列里,如果时间到了,就直接加到延时队列里
         if waitEntry.readyAt.After(q.clock.Now()) {
            insert(waitingForQueue, waitingEntryByData, waitEntry)
         } else {
            q.Add(waitEntry.data)
         }
         // 下面的逻辑就是将 waitingForAddCh 中的数据处理完
         drained := false
         for !drained {
            select {
            case waitEntry := <-q.waitingForAddCh:
               if waitEntry.readyAt.After(q.clock.Now()) {
                  insert(waitingForQueue, waitingEntryByData, waitEntry)
               } else {
                  q.Add(waitEntry.data)
               }
            default:
               drained = true
            }
         }
      }
   }
}
2.2.4

这个方法的作用是在指定的延时到达之后,在 work queue 中添加一个元素,源码如下:

func (q *delayingType) AddAfter(item interface{}, duration time.Duration) {
   if q.ShuttingDown() { // 已经在关闭中就直接返回
      return
   }

   q.metrics.retry()

   if duration <= 0 { // 如果时间到了,就直接添加
      q.Add(item)
      return
   }

   select {
   case <-q.stopCh:
     // 构造 waitFor{},丢到 waitingForAddCh
   case q.waitingForAddCh <- &waitFor{data: item, readyAt: q.clock.Now().Add(duration)}:
   }
}

其实就是一个往堆加入元素的过程
func insert(q *waitForPriorityQueue, knownEntries map[t]*waitFor, entry *waitFor) {
   // 这里的主要逻辑是看一个 entry 是否存在,如果已经存在,新的 entry 的 ready 时间更短,就更新时间
   existing, exists := knownEntries[entry.data]
   if exists {
      if existing.readyAt.After(entry.readyAt) {
         existing.readyAt = entry.readyAt // 如果存在就只更新时间
         heap.Fix(q, existing.index)
      }

      return
   }
   // 如果不存在就丢到 q 里,同时在 map 里记录一下,用于查重
   heap.Push(q, entry)
   knownEntries[entry.data] = entry
}


2.2.5 总结

(1)延迟队列的核心就是,根据加入队列的时间,构造一个最小堆,然后再到时间点后,将其加入queue中

(2)上诉判断是否到时间点,不仅仅是一个for循环,还利用了心跳,channel机制

(3)当某个对象处理的时候失败了,可以利用延迟队列的思想,等一会再重试,因为马上重试肯定是失败的

2.3 RateLimitingQueue-限速队列

2.3.1 RateLimiting结构体
type RateLimitingInterface interface {
    DelayingInterface     //延迟队列

    AddRateLimited(item interface{})     //已限速方式,往队列添加一个元素

    // 标记介绍重试
    Forget(item interface{})
  
  // 重试了几次
    NumRequeues(item interface{}) int
}


// rateLimitingType wraps an Interface and provides rateLimited re-enquing
type rateLimitingType struct {
    DelayingInterface    

    rateLimiter RateLimiter   //多了一个限速器
}
2.3.2 限速器类型

可以看出来,限速队列和 延迟队列是一模一样的。

延迟队列是自己决定 某个元素延迟多久。

而限速队列是 有限速器决定 某个元素延迟多久。

type RateLimiter interface {
    // 输入一个对象,判断延迟多久
    When(item interface{}) time.Duration
    
    // 标记介绍重试
    Forget(item interface{})
    
    // 重试了几次
    NumRequeues(item interface{}) int
}

这个接口有五个实现,分别为:

  1. BucketRateLimiter
  2. ItemExponentialFailureRateLimiter
  3. ItemFastSlowRateLimiter
  4. MaxOfRateLimiter
  5. WithMaxWaitRateLimiter
BucketRateLimiter

这个限速器可说的不多,用了 golang 标准库的 golang.org/x/time/rate.Limiter 实现。BucketRateLimiter 实例化的时候比如传递一个 rate.NewLimiter(rate.Limit(10), 100) 进去,表示令牌桶里最多有 100 个令牌,每秒发放 10 个令牌。

所有元素都是一样的,来几次都是一样,所以NumRequeues,Forget都没有意义。

type BucketRateLimiter struct {
   *rate.Limiter
}

var _ RateLimiter = &BucketRateLimiter{}

func (r *BucketRateLimiter) When(item interface{}) time.Duration {
   return r.Limiter.Reserve().Delay() // 过多久后给当前 item 发放一个令牌
}

func (r *BucketRateLimiter) NumRequeues(item interface{}) int {
   return 0
}

// 
func (r *BucketRateLimiter) Forget(item interface{}) {
}
ItemExponentialFailureRateLimiter

Exponential 是指数的意思,从这个限速器的名字大概能猜到是失败次数越多,限速越长而且是指数级增长的一种限速器。

结构体定义如下,属性含义基本可以望文生义

func (r *ItemExponentialFailureRateLimiter) When(item interface{}) time.Duration {
   r.failuresLock.Lock()
   defer r.failuresLock.Unlock()

   exp := r.failures[item]
   r.failures[item] = r.failures[item] + 1 // 失败次数加一

   // 每调用一次,exp 也就加了1,对应到这里时 2^n 指数爆炸
   backoff := float64(r.baseDelay.Nanoseconds()) * math.Pow(2, float64(exp))
   if backoff > math.MaxInt64 { // 如果超过了最大整型,就返回最大延时,不然后面时间转换溢出了
      return r.maxDelay
   }

   calculated := time.Duration(backoff)
   if calculated > r.maxDelay { // 如果超过最大延时,则返回最大延时
      return r.maxDelay
   }

   return calculated
}

func (r *ItemExponentialFailureRateLimiter) NumRequeues(item interface{}) int {
   r.failuresLock.Lock()
   defer r.failuresLock.Unlock()

   return r.failures[item]
}

func (r *ItemExponentialFailureRateLimiter) Forget(item interface{}) {
   r.failuresLock.Lock()
   defer r.failuresLock.Unlock()

   delete(r.failures, item)
}
ItemFastSlowRateLimiter

快慢限速器,也就是先快后慢,定义一个阈值,超过了就慢慢重试。先看类型定义:

type ItemFastSlowRateLimiter struct {
   failuresLock sync.Mutex
   failures     map[interface{}]int

   maxFastAttempts int            // 快速重试的次数
   fastDelay       time.Duration  // 快重试间隔
   slowDelay       time.Duration  // 慢重试间隔
}

func (r *ItemFastSlowRateLimiter) When(item interface{}) time.Duration {
   r.failuresLock.Lock()
   defer r.failuresLock.Unlock()

   r.failures[item] = r.failures[item] + 1 // 标识重试次数 + 1

   if r.failures[item] <= r.maxFastAttempts { // 如果快重试次数没有用完,则返回 fastDelay
      return r.fastDelay
   }

   return r.slowDelay // 反之返回 slowDelay
}

func (r *ItemFastSlowRateLimiter) NumRequeues(item interface{}) int {
   r.failuresLock.Lock()
   defer r.failuresLock.Unlock()

   return r.failures[item]
}

func (r *ItemFastSlowRateLimiter) Forget(item interface{}) {
   r.failuresLock.Lock()
   defer r.failuresLock.Unlock()

   delete(r.failures, item)
}
MaxOfRateLimiter

组合限速器,内部放多个限速器,然后返回限速最慢的一个延时:

type MaxOfRateLimiter struct {
   limiters []RateLimiter
}

func (r *MaxOfRateLimiter) When(item interface{}) time.Duration {
   ret := time.Duration(0)
   for _, limiter := range r.limiters {
      curr := limiter.When(item)
      if curr > ret {
         ret = curr
      }
   }

   return ret
}


WithMaxWaitRateLimiter

这个限速器也很简单,就是在其他限速器上包装一个最大延迟的属性,如果到了最大延时,则直接返回。这样就能避免延迟时间不可控,万一一个对象失败了多次,那以后的时间会越来越大。

type WithMaxWaitRateLimiter struct {
   limiter  RateLimiter   // 其他限速器
   maxDelay time.Duration // 最大延时
}

func NewWithMaxWaitRateLimiter(limiter RateLimiter, maxDelay time.Duration) RateLimiter {
   return &WithMaxWaitRateLimiter{limiter: limiter, maxDelay: maxDelay}
}

func (w WithMaxWaitRateLimiter) When(item interface{}) time.Duration {
   delay := w.limiter.When(item)
   if delay > w.maxDelay {
      return w.maxDelay // 已经超过了最大延时,直接返回最大延时
   }

   return delay
}

3.总结

(1)workerqueue使用于只关注结果的处理方式。 比如统计一个Pod update了多少次这种关乎 过程的 处理。不能用,因为workerqueue进行了合并

(2)workerqueue实现了很多限速机制,可以更加情况酌情使用

4. 参考文档

https://blog.csdn.net/weixin_42663840/article/details/81482553

https://www.danielhu.cn/post/k8s/client-go-workqueue/

©著作权归作者所有,转载或内容合作请联系作者
  • 序言:七十年代末,一起剥皮案震惊了整个滨河市,随后出现的几起案子,更是在滨河造成了极大的恐慌,老刑警刘岩,带你破解...
    沈念sama阅读 206,839评论 6 482
  • 序言:滨河连续发生了三起死亡事件,死亡现场离奇诡异,居然都是意外死亡,警方通过查阅死者的电脑和手机,发现死者居然都...
    沈念sama阅读 88,543评论 2 382
  • 文/潘晓璐 我一进店门,熙熙楼的掌柜王于贵愁眉苦脸地迎上来,“玉大人,你说我怎么就摊上这事。” “怎么了?”我有些...
    开封第一讲书人阅读 153,116评论 0 344
  • 文/不坏的土叔 我叫张陵,是天一观的道长。 经常有香客问我,道长,这世上最难降的妖魔是什么? 我笑而不...
    开封第一讲书人阅读 55,371评论 1 279
  • 正文 为了忘掉前任,我火速办了婚礼,结果婚礼上,老公的妹妹穿的比我还像新娘。我一直安慰自己,他们只是感情好,可当我...
    茶点故事阅读 64,384评论 5 374
  • 文/花漫 我一把揭开白布。 她就那样静静地躺着,像睡着了一般。 火红的嫁衣衬着肌肤如雪。 梳的纹丝不乱的头发上,一...
    开封第一讲书人阅读 49,111评论 1 285
  • 那天,我揣着相机与录音,去河边找鬼。 笑死,一个胖子当着我的面吹牛,可吹牛的内容都是我干的。 我是一名探鬼主播,决...
    沈念sama阅读 38,416评论 3 400
  • 文/苍兰香墨 我猛地睁开眼,长吁一口气:“原来是场噩梦啊……” “哼!你这毒妇竟也来了?” 一声冷哼从身侧响起,我...
    开封第一讲书人阅读 37,053评论 0 259
  • 序言:老挝万荣一对情侣失踪,失踪者是张志新(化名)和其女友刘颖,没想到半个月后,有当地人在树林里发现了一具尸体,经...
    沈念sama阅读 43,558评论 1 300
  • 正文 独居荒郊野岭守林人离奇死亡,尸身上长有42处带血的脓包…… 初始之章·张勋 以下内容为张勋视角 年9月15日...
    茶点故事阅读 36,007评论 2 325
  • 正文 我和宋清朗相恋三年,在试婚纱的时候发现自己被绿了。 大学时的朋友给我发了我未婚夫和他白月光在一起吃饭的照片。...
    茶点故事阅读 38,117评论 1 334
  • 序言:一个原本活蹦乱跳的男人离奇死亡,死状恐怖,灵堂内的尸体忽然破棺而出,到底是诈尸还是另有隐情,我是刑警宁泽,带...
    沈念sama阅读 33,756评论 4 324
  • 正文 年R本政府宣布,位于F岛的核电站,受9级特大地震影响,放射性物质发生泄漏。R本人自食恶果不足惜,却给世界环境...
    茶点故事阅读 39,324评论 3 307
  • 文/蒙蒙 一、第九天 我趴在偏房一处隐蔽的房顶上张望。 院中可真热闹,春花似锦、人声如沸。这庄子的主人今日做“春日...
    开封第一讲书人阅读 30,315评论 0 19
  • 文/苍兰香墨 我抬头看了看天上的太阳。三九已至,却和暖如春,着一层夹袄步出监牢的瞬间,已是汗流浃背。 一阵脚步声响...
    开封第一讲书人阅读 31,539评论 1 262
  • 我被黑心中介骗来泰国打工, 没想到刚下飞机就差点儿被人妖公主榨干…… 1. 我叫王不留,地道东北人。 一个月前我还...
    沈念sama阅读 45,578评论 2 355
  • 正文 我出身青楼,却偏偏与公主长得像,于是被迫代替她去往敌国和亲。 传闻我的和亲对象是个残疾皇子,可洞房花烛夜当晚...
    茶点故事阅读 42,877评论 2 345

推荐阅读更多精彩内容