func (queue *PackQueue) Close() error {
	close(queue.writeChan)
	close(queue.readChan)
	close(queue.errorChan)
	return nil
}
Close the all of queue's channels
package main
import (
	"bufio"
	"fmt"
	"net"
	"time"
	"github.com/Alienero/quick-know/mqtt"
	"github.com/golang/glog"
)
type PackQueue struct {
	writeError error
	errorChan chan error
	writeChan chan *pakcAdnType
	readChan  chan *packAndErr
	r *bufio.Reader
	w *bufio.Writer
	conn net.Conn
	alive int
}
type packAndErr struct {
	pack *mqtt.Pack
	err  error
}
const (
	NO_DELAY = iota
	DELAY
	FLUSH
)
type pakcAdnType struct {
	pack *mqtt.Pack
	typ  byte
}
func NewPackQueue(r *bufio.Reader, w *bufio.Writer, conn net.Conn, alive int) *PackQueue {
	if alive < 1 {
		alive = Conf.ReadTimeout
	}
	alive = int(float32(alive)*1.5 + 1)
	return &PackQueue{
		alive:     alive,
		r:         r,
		w:         w,
		conn:      conn,
		writeChan: make(chan *pakcAdnType, Conf.WirteLoopChanNum),
		readChan:  make(chan *packAndErr, 1),
		errorChan: make(chan error, 1),
	}
}
func (queue *PackQueue) writeLoop() {
	var err error
loop:
	for {
		select {
		case pt := <-queue.writeChan:
			if pt == nil {
				break loop
			}
			if Conf.WriteTimeout > 0 {
				queue.conn.SetWriteDeadline(time.Now().Add(time.Second * time.Duration(Conf.WriteTimeout)))
			}
			switch pt.typ {
			case NO_DELAY:
				err = mqtt.WritePack(pt.pack, queue.w)
			case DELAY:
				err = mqtt.DelayWritePack(pt.pack, queue.w)
			case FLUSH:
				err = queue.w.Flush()
			}
			if err != nil {
				queue.writeError = err
				break loop
			}
		}
	}
	if err != nil {
		queue.errorChan <- err
	}
}
func (queue *PackQueue) WritePack(pack *mqtt.Pack) error {
	if queue.writeError != nil {
		return queue.writeError
	}
	queue.writeChan <- &pakcAdnType{pack: pack}
	return nil
}
func (queue *PackQueue) WriteDelayPack(pack *mqtt.Pack) error {
	if queue.writeError != nil {
		return queue.writeError
	}
	queue.writeChan <- &pakcAdnType{
		pack: pack,
		typ:  DELAY,
	}
	return nil
}
func (queue *PackQueue) Flush() error {
	if queue.writeError != nil {
		return queue.writeError
	}
	queue.writeChan <- &pakcAdnType{typ: FLUSH}
	return nil
}
func (queue *PackQueue) ReadPack() (pack *mqtt.Pack, err error) {
	go func() {
		p := new(packAndErr)
		if Conf.ReadTimeout > 0 {
			queue.conn.SetReadDeadline(time.Now().Add(time.Second * time.Duration(Conf.ReadTimeout)))
		}
		p.pack, p.err = mqtt.ReadPack(queue.r)
		queue.readChan <- p
	}()
	select {
	case err = <-queue.errorChan:
	case pAndErr := <-queue.readChan:
		pack = pAndErr.pack
		err = pAndErr.err
	}
	return
}
func (queue *PackQueue) ReadPackInLoop(fin <-chan byte) <-chan *packAndErr {
	ch := make(chan *packAndErr, Conf.ReadPackLoop)
	go func() {
		is_continue := true
		p := new(packAndErr)
	loop:
		for {
			if queue.alive > 0 {
				queue.conn.SetReadDeadline(time.Now().Add(time.Second * time.Duration(queue.alive)))
			}
			if is_continue {
				p.pack, p.err = mqtt.ReadPack(queue.r)
				if p.err != nil {
					is_continue = false
				}
				select {
				case ch <- p:
				case <-fin:
					glog.Info("Queue FIN")
					break loop
				}
			} else {
				<-fin
				glog.Info("Queue FIN")
				break loop
			}
			p = new(packAndErr)
		}
		close(ch)
	}()
	return ch
}