type T interface{}
type UnlimitSizeChan struct {
bufCount int64    // 統(tǒng)計(jì)元素?cái)?shù)量，原子操作
In       chan<- T // 寫入channle
Out      <-chan T    // 讀取channle
buffer   *RingBuffer // 自適應(yīng)擴(kuò)縮容Buf
}

type cell struct {
	Data     []T   // 數(shù)據(jù)部分
	fullFlag bool  // cell滿的標(biāo)志
	next     *cell // 指向后一個(gè)cellBuffer
	pre      *cell // 指向前一個(gè)cellBuffer
	r int // 下一個(gè)要讀的指針
	w int // 下一個(gè)要下的指針
}
type RingBuffer struct {
	cellCount int // cell 數(shù)量統(tǒng)計(jì)
	readCell  *cell // 下一個(gè)要讀的cell
	writeCell *cell // 下一個(gè)要寫的cell
}

package unlimitSizeChan
import (
	"errors"
	"fmt"
)
var ErrRingIsEmpty = errors.New("ringbuffer is empty")
// CellInitialSize cell的初始容量
var CellInitialSize = 1024
// CellInitialCount 初始化cell數(shù)量
var CellInitialCount = 2
type cell struct {
	Data     []T   // 數(shù)據(jù)部分
	fullFlag bool  // cell滿的標(biāo)志
	next     *cell // 指向后一個(gè)cellBuffer
	pre      *cell // 指向前一個(gè)cellBuffer
	r int // 下一個(gè)要讀的指針
	w int // 下一個(gè)要下的指針
}
type RingBuffer struct {
	cellCount int // cell 數(shù)量統(tǒng)計(jì)
	readCell  *cell // 下一個(gè)要讀的cell
	writeCell *cell // 下一個(gè)要寫的cell
}
// NewRingBuffer 新建一個(gè)ringbuffe，包含兩個(gè)cell
func NewRingBuffer() *RingBuffer {
	rootCell := &cell{
		Data: make([]T, CellInitialSize),
	}
	lastCell := &cell{
		Data: make([]T, CellInitialSize),
	}
	rootCell.pre = lastCell
	lastCell.pre = rootCell
	rootCell.next = lastCell
	lastCell.next = rootCell
	return &RingBuffer{
		cellCount: CellInitialCount,
		readCell:  rootCell,
		writeCell: rootCell,
	}
}
// Read 讀取數(shù)據(jù)
func (r *RingBuffer) Read() (T, error) {
	// 無數(shù)據(jù)
	if r.IsEmpty() {
		return nil, ErrRingIsEmpty
	}
	// 讀取數(shù)據(jù)，并將讀指針向右移動(dòng)一位
	value := r.readCell.Data[r.readCell.r]
	r.readCell.r++
	// 此cell已經(jīng)讀完
	if r.readCell.r == CellInitialSize {
		// 讀指針歸零，并將該cell狀態(tài)置為非滿
		r.readCell.r = 0
		r.readCell.fullFlag = false
		// 將readCell指向下一個(gè)cell
		r.readCell = r.readCell.next
	}
	return value, nil
}
// Pop 讀一個(gè)元素，讀完后移動(dòng)指針
func (r *RingBuffer) Pop() T {
	value, err := r.Read()
	if err != nil {
		panic(err.Error())
	}
	return value
}
// Peek 窺視 讀一個(gè)元素，僅讀但不移動(dòng)指針
func (r *RingBuffer) Peek() T {
	if r.IsEmpty() {
		panic(ErrRingIsEmpty.Error())
	}
	// 僅讀
	value := r.readCell.Data[r.readCell.r]
	return value
}
// Write 寫入數(shù)據(jù)
func (r *RingBuffer) Write(value T) {
	// 在 r.writeCell.w 位置寫入數(shù)據(jù)，指針向右移動(dòng)一位
	r.writeCell.Data[r.writeCell.w] = value
	r.writeCell.w++
	// 當(dāng)前cell寫滿了
	if r.writeCell.w == CellInitialSize {
		// 指針置0，將該cell標(biāo)記為已滿，并指向下一個(gè)cell
		r.writeCell.w = 0
		r.writeCell.fullFlag = true
		r.writeCell = r.writeCell.next
	}
	// 下一個(gè)cell也已滿，擴(kuò)容
	if r.writeCell.fullFlag == true {
		r.grow()
	}
}
// grow 擴(kuò)容
func (r *RingBuffer) grow() {
	// 新建一個(gè)cell
	newCell := &cell{
		Data: make([]T, CellInitialSize),
	}
	// 總共三個(gè)cell，writeCell，preCell，newCell
	// 本來關(guān)系： preCell <===> writeCell
	// 現(xiàn)在將newcell插入：preCell <===> newCell <===> writeCell
	pre := r.writeCell.pre
	pre.next = newCell
	newCell.pre = pre
	newCell.next = r.writeCell
	r.writeCell.pre = newCell
	// 將writeCell指向新建的cell
	r.writeCell = r.writeCell.pre
	// cell 數(shù)量加一
	r.cellCount++
}
// IsEmpty 判斷ringbuffer是否為空
func (r *RingBuffer) IsEmpty() bool {
	// readCell和writeCell指向同一個(gè)cell，并且該cell的讀寫指針也指向同一個(gè)位置，并且cell狀態(tài)為非滿
	if r.readCell == r.writeCell && r.readCell.r == r.readCell.w && r.readCell.fullFlag == false {
		return true
	}
	return false
}
// Capacity ringBuffer容量
func (r *RingBuffer) Capacity() int {
	return r.cellCount * CellInitialSize
}
// Reset 重置為僅指向兩個(gè)cell的ring
func (r *RingBuffer) Reset() {
	lastCell := r.readCell.next
	lastCell.w = 0
	lastCell.r = 0
	r.readCell.r = 0
	r.readCell.w = 0
	r.cellCount = CellInitialCount
	lastCell.next = r.readCell
}

package unlimitSizeChan
import "sync/atomic"
type T interface{}
// UnlimitSizeChan 無限緩存的Channle
type UnlimitSizeChan struct {
	bufCount int64       // 統(tǒng)計(jì)元素?cái)?shù)量，原子操作
	In       chan<- T    // 寫入channle
	Out      <-chan T    // 讀取channle
	buffer   *RingBuffer // 自適應(yīng)擴(kuò)縮容Buf
}
// Len uc中總共的元素?cái)?shù)量
func (uc UnlimitSizeChan) Len() int {
	return len(uc.In) + uc.BufLen() + len(uc.Out)
}
// BufLen uc的buf中的元素?cái)?shù)量
func (uc UnlimitSizeChan) BufLen() int {
	return int(atomic.LoadInt64(&uc.bufCount))
}
// NewUnlimitSizeChan 新建一個(gè)無限緩存的Channle，并指定In和Out大小(In和Out設(shè)置得一樣大)
func NewUnlimitSizeChan(initCapacity int) *UnlimitSizeChan {
	return NewUnlitSizeChanSize(initCapacity, initCapacity)
}
// NewUnlitSizeChanSize 新建一個(gè)無限緩存的Channle，并指定In和Out大小(In和Out設(shè)置得不一樣大)
func NewUnlitSizeChanSize(initInCapacity, initOutCapacity int) *UnlimitSizeChan {
	in := make(chan T, initInCapacity)
	out := make(chan T, initOutCapacity)
	ch := UnlimitSizeChan{In: in, Out: out, buffer: NewRingBuffer()}
	go process(in, out, &ch)
	return &ch
}
// 內(nèi)部Worker Groutine實(shí)現(xiàn)
func process(in, out chan T, ch *UnlimitSizeChan) {
	defer close(out) // in 關(guān)閉，數(shù)據(jù)讀取后也把out關(guān)閉
	// 不斷從in中讀取數(shù)據(jù)放入到out或者ringbuf中
loop:
	for {
		// 第一步：從in中讀取數(shù)據(jù)
		value, ok := <-in
		if !ok {
			// in 關(guān)閉了，退出loop
			break loop
		}
		// 第二步：將數(shù)據(jù)存儲(chǔ)到out或者buf中
		if atomic.LoadInt64(&ch.bufCount) > 0 {
			// 當(dāng)buf中有數(shù)據(jù)時(shí)，新數(shù)據(jù)優(yōu)先存放到buf中，確保數(shù)據(jù)FIFO原則
			ch.buffer.Write(value)
			atomic.AddInt64(&ch.bufCount, 1)
		} else {
			// out 沒有滿,數(shù)據(jù)放入out中
			select {
			case out <- value:
				continue
			default:
			}
			// out 滿了，數(shù)據(jù)放入buf中
			ch.buffer.Write(value)
			atomic.AddInt64(&ch.bufCount, 1)
		}
		// 第三步：處理buf，一直嘗試把buf中的數(shù)據(jù)放入到out中，直到buf中沒有數(shù)據(jù)
		for !ch.buffer.IsEmpty() {
			select {
			// 為了避免阻塞in，還要嘗試從in中讀取數(shù)據(jù)
			case val, ok := <-in:
				if !ok {
					// in 關(guān)閉了，退出loop
					break loop
				}
				// 因?yàn)檫@個(gè)時(shí)候out是滿的，新數(shù)據(jù)直接放入buf中
				ch.buffer.Write(val)
				atomic.AddInt64(&ch.bufCount, 1)
			// 將buf中數(shù)據(jù)放入out
			case out <- ch.buffer.Peek():
				ch.buffer.Pop()
				atomic.AddInt64(&ch.bufCount, -1)
				if ch.buffer.IsEmpty() { // 避免內(nèi)存泄露
					ch.buffer.Reset()
					atomic.StoreInt64(&ch.bufCount, 0)
				}
			}
		}
	}
	// in被關(guān)閉退出loop后，buf中還有可能有未處理的數(shù)據(jù)，將他們?nèi)雘ut中，并重置buf
	for !ch.buffer.IsEmpty() {
		out <- ch.buffer.Pop()
		atomic.AddInt64(&ch.bufCount, -1)
	}
	ch.buffer.Reset()
	atomic.StoreInt64(&ch.bufCount, 0)
}

ch := NewUnlimitSizeChan(1000)
// or ch := NewUnlitSizeChanSize(100,200)
go func() {
? ? for ...... {
? ? ? ? ...
? ? ? ? ch.In <- ... // send values
? ? ? ? ...
? ? }
? ? close(ch.In) // close In channel
}()
for v := range ch.Out { // read values
? ? fmt.Println(v)
}