func NewUnitedQueue(storage store.Storage, ip string, port int, etcdServers []string, etcdKey string) (*UnitedQueue, error) {
	topics := make(map[string]*topic)
	etcdStop := make(chan bool)
	uq := new(UnitedQueue)
	uq.topics = topics
	uq.storage = storage
	uq.etcdStop = etcdStop
	if len(etcdServers) > 0 {
		selfAddr := utils.Addrcat(ip, port)
		uq.selfAddr = selfAddr
		etcdClient := etcd.NewClient(etcdServers)
		uq.etcdClient = etcdClient
		uq.etcdKey = etcdKey
	}
	err := uq.loadQueue()
	if err != nil {
		return nil, err
	}
	go uq.etcdRun()
	return uq, nil
}
NewUnitedQueue returns a new UnitedQueue
package queue
import (
	"encoding/binary"
	"errors"
	"log"
	"strconv"
	"strings"
	"sync"
	"time"
	"github.com/buaazp/uq/store"
	"github.com/buaazp/uq/utils"
	"github.com/coreos/go-etcd/etcd"
)
const (
	storageKeyWord   string        = "UnitedQueueKey"
	bgBackupInterval time.Duration = 10 * time.Second
	bgCleanInterval  time.Duration = 20 * time.Second
	bgCleanTimeout   time.Duration = 5 * time.Second
	keyTopicStore    string        = ":store"
	keyTopicHead     string        = ":head"
	keyTopicTail     string        = ":tail"
	keyLineStore     string        = ":store"
	keyLineHead      string        = ":head"
	keyLineRecycle   string        = ":recycle"
	keyLineInflight  string        = ":inflight"
)
type UnitedQueue struct {
	topics     map[string]*topic
	topicsLock sync.RWMutex
	storage    store.Storage
	etcdLock   sync.RWMutex
	selfAddr   string
	etcdClient *etcd.Client
	etcdKey    string
	etcdStop   chan bool
	wg         sync.WaitGroup
}