В предыдущей статье говорилось оchannelВ этой статье основное внимание уделяется основному использованию и некоторым вопросам, требующим внимания при его использовании.channelДве структуры данных в:круговая очередьиДвусторонний связанный список.

Требование к описанию канала

Чтобы понять, какие проблемы решают эти структуры данных, давайте вкратце рассмотрим, зачем нужны эти две структуры данных и какие проблемы они решают. мы знаемgoroutineэто поток пользовательского режима, отличающийсяgoroutineСуществует потребность в передаче сообщений между ними. В исходном программировании процессов и потоков (системных потоков) мы будем использовать конвейер, аchannelЭто реализация конвейера для потока пользовательского режима для передачи сообщений, и он безопасен для типов.

теперь, когдаchannelэто конвейер, используемый для удовлетворения различныхgoroutineобмен сообщениями. Итак, как реализовать такой конвейер?

Давайте посмотрим на наши ежедневные потребности в сообщениях:

1. может иметь несколькоgoroutineВ направленииchannelЧитайте и пишите, и убедитесь, что нет проблем с конкуренцией, вам нужно использоватьочередьуправлять блокировкойgoroutine, решать проблемы конкуренции;
2. нужно управлять отправкойchannelсообщения (буферизованные, небуферизованные), для буферизованныхchannelможет быть использованкруговая очередьдля управления несколькими сообщениями.

Конечно, приведенные выше требования упрощены, напримерchannelТакже нужно иметь блокировку, пробуждениеgoroutineОднако, чтобы больше сосредоточиться на теме этой статьи, мы сосредоточимся только на двух вышеупомянутых вопросах.

Структура данных канала

Далее мы анализируемchannelВ реальной эксплуатации, какова его структура. Конечно, он делится на два типа: буферизованный и небуферизованный. Давайте сначала посмотрим на небуферизованный случай.

небуферизованный

Сначала опубликуйте пример кода. только два прочиталgoroutineЗаблокирован на небуферизованном канале.

func main() {
 ch := make(chan int) // 无缓冲

 go goRoutineA(ch)
 go goRoutineB(ch)
 ch <- 1

 time.Sleep(time.Second * 1)
}

func goRoutineA(ch chan int) {
 v := <-ch
 fmt.Printf("A received data: %d\n", v)
}

func goRoutineB(ch chan int) {
 v := <-ch
 fmt.Printf("B received data: %d\n", v)
}

Посмотрим, когда код выполнится доch <- 1после этой строкиchannelЧем наполнена структура!

Обратите внимание, чтоbufСохраняемая длина поля равна 0, потому чтонебуферизованный каналЦиклическая очередь не используется для хранения данных. Он должен дождаться готовности горутин чтения и записи, а затем передать данные непосредственно другой стороне. Давайте посмотрим на процесс обмена небуферизованными данными с картинкой.

На приведенном выше рисунке описан процесс обмена данными, а затем посмотрите на блок-схему структуры горутины чтения. Заблокированная горутина будет смонтирована в соответствующей очереди, которая является двусторонней очередью.

В приведенном выше примере, поскольку запись не готова, когда запущены две горутины чтения, все чтение приостанавливается в очереди; когда горутина записи готова, запись не будет выполнена, потому что в это время готово чтение. заблокировать, приостановитьsendqсередина. Вы можете изменить приведенный выше код и убедиться, что запись блокируется, а чтение выполняется немедленно.

буферизованный

Давайте изменим приведенный выше код на буферизованный канал, а затем посмотрим на ситуацию с буферизацией.

func main() {
 ch := make(chan int, 3) // 有缓冲

 // 都不会阻塞
 ch <- 1
 ch <- 2
    ch <- 3
    // 会阻塞，被挂起到 sendq 中
 go func() {
  ch <- 4
 }()

 // 只是为了debug
 var a int
 fmt.Println(a)

 go goRoutineA(ch)
 go goRoutineA(ch)
 go goRoutineB(ch)
 go goRoutineB(ch) // 猜猜这里会被挂起吗？

 time.Sleep(time.Second * 2)
}

func goRoutineA(ch chan int) {
 v := <-ch
 fmt.Printf("A received data: %d\n", v)
}

func goRoutineB(ch chan int) {
 v := <-ch
 fmt.Printf("B received data: %d\n", v)
}

Вставьте выполнение в первую строкуgo goRoutineA(ch)час,hchanнаполнение конструкции.

Здесь вы можете видеть, что размер буфера равен 3. Из-за увеличенного буфера, пока горутина записи не заполнит буфер, это не приведет к блокировке сопрограммы. Но как только в буфере больше нет места, горутина записи будет приостановлена ​​доsendq, пока не появится возможность его разбудить (есть и другие сцены пробуждения, пропустите эту).

Фактически буферизованный канал превращает синхронную связь в асинхронную. Каналу записи не нужно обращать внимание на канал чтения, пока есть место, он будет писать; и то же самое верно и для чтения, пока есть данные, их можно нормально прочитать, если нет, то это будет приостановлено в очереди, ожидая пробуждения. На следующем рисунке графически показано, как буферизованные каналы обмениваются данными.

Давайте посмотрим на структуру на этот раз в виде диаграммы.Схема здесь немного ленивая, но описание круговой части очереди добавлено на основе приведенной выше диаграммы.На самом деле, в этом примере, Горутина чтения не будет заблокирована, на это нужно обратить внимание при просмотре.

круговая очередь

Самое главное сегодня — понять две ключевые структуры данных в каналах. Готовясь к чтению исходного кода для следующей лекции, я абстрагировал код части канала с циклической очередью.

// 队列满了
var ErrQFull = errors.New("circular is full")

// 没有值
var ErrQEmpty = errors.New("circular is empty")

// 定义循环队列
// 如何确定队空，还是队满？q.sendx = (q.sendx+1) % q.dataqsiz
type queue struct {
 buf      []int // 队列元素存储
 dataqsiz uint  // circular 队列长度
 qcount   uint  // 有多少元素在buf中 qcount = len(buf)
 sendx    uint  // 可以理解为队尾指针，向队列写入数据
 recvx    uint  // 可以理解为队头指针，从队列读取数据
}

func makeQ(size int) *queue {
 q := &queue{
  dataqsiz: uint(size),
  buf:      nil,
 }

 q.buf = make([]int, q.dataqsiz)

 return q
}

// 向buf中写入数据
// 请看 chansend 函数
func (c *queue) insert(ele int) error {
 // 检查队列是否有空间
 if c.dataqsiz > 0 && c.qcount == c.dataqsiz {
  return ErrQFull
 }

 // 存入数据
 c.buf[c.sendx] = ele
 c.sendx++                  // 尾指针后移
 if c.sendx == c.dataqsiz { // 如果相等，说明队列写满了，sendx放到开始位置
  c.sendx = 0
 }
 c.qcount++

 return nil
}

// 从buf中读取数据
func (c *queue) read() (int, error) {
 // 队列中没有数据了
 if c.dataqsiz > 0 && c.qcount == 0 {
  return 0, ErrQEmpty
 }

 ret := c.buf[c.recvx] // 取出元素
 c.buf[c.recvx] = 0
 c.recvx++
 if c.recvx == c.dataqsiz { // 如果相等，说明写到末尾了，recvx放到开始位置
  c.recvx = 0
 }
 c.qcount--

 return ret, nil
}

Вышеуказанный код в основномchannelРеализация части круговой очереди . Реализация этой очереди немного отличается от циклической очереди, которую мы обычно реализуем. Как правило, чтобы облегчить пустое суждение, мы будем тратить буферное пространство, чтобы облегчить пустое суждение.Формула такова:(tail+1)%n=head; Однако в циклической очереди канала за счет подсчета элементов циклической очереди это гарантирует, что это пространство не будет потрачено впустую, и в то же время также может удовлетворять временной сложности O(1) вычисления количество буферизованных элементов канала.

Суммировать

Обобщите сегодняшние ключевые сообщения.

1. В канале используются две структуры данных:круговая очередьиДвусторонний связанный список;
2. круговая очередьОн используется только в буферизованном канале, в основном используется в качестве буфера для сообщений, чтобы обеспечить упорядоченность сообщений;
3. Двусторонний связанный списокОн используется для приостановки заблокированных горутин чтения и записи и будет уведомлен справедливо в соответствии с порядком очереди при его пробуждении;
4. Небуферизованные каналы не используютсякруговая очередьСвязанная структура, она должна быть готова к обмену сообщениями после горутин чтения и записи;
5. как буферизованное сообщениекруговая очередьЧтобы не тратить пространство данных впустую, отметьте поле текущего номера элемента.

В следующей главе мы попробуем прочитатьchannelИсходный код, я хочу попробовать записать видео, чтобы рассказать об этой части исходного кода!

использованная литература

• [1] Diving Deep Into The Golang Channels.
• [2] Understanding Channels
• [3] The Nature Of Channels In Go