func getAllAvailableNSQDFromNSQLookupds(nsqlookupdAddrs []string) ([]nsqlookupdNodesProducer, error) {
	if nil == nsqlookupdAddrs ||
		len(nsqlookupdAddrs) == 0 {
		return nil, fmt.Errorf("nsqlookupd address must be non-empty")
	}
	var nodes []nsqlookupdNodesProducer
	var err error
	for _, v := range nsqlookupdAddrs {
		nodes, err = getAllAvailableNSQDFromNSQLookupd(v)
		if nil != err {
			continue
		}
		return nodes, nil
	}
	return nil, err
}
Get all available nodes from nsqlookupd
package nsqproducer
import (
	"encoding/json"
	"fmt"
	"github.com/nsqio/go-nsq"
)
var (
	maxRetryTimes = 1
)
func SetMaxRetryTimes(tm int) {
	maxRetryTimes = tm
}
type logger interface {
	Output(calldepth int, s string) error
}
type INSQProducer interface {
	Stop()
	Publish(topic string, body []byte) error
	SetLogger(l logger, lvl nsq.LogLevel)
}
type nsqlookupdNodesProducer struct {
	BroadcastAddress string   `json:"broadcast_address"`
	TCPPort          int      `json:"tcp_port"`
	Topics           []string `json:"topics"`
	available bool
	inuse     bool
}
type nsqlookupdNodesData struct {
	Producers []nsqlookupdNodesProducer `json:"producers"`
}
type nsqlookupdNodesResult struct {
	StatusCode int                 `json:"status_code"`
	StatusText string              `json:"status_txt"`
	Data       nsqlookupdNodesData `json:"data"`
}
type nsqProducer struct {
	lookupdAddrs   []string
	cfg            *nsq.Config
	availableNodes []nsqlookupdNodesProducer
	currentProducer *nsq.Producer
	inuseNode       *nsqlookupdNodesProducer
	loglvl nsq.LogLevel
	log    logger
}
func (n *nsqProducer) Stop() {
	if n.currentProducer != nil {
		n.currentProducer.Stop()
	}
	n.currentProducer = nil
	n.inuseNode = nil
	n.availableNodes = nil
}
func (n *nsqProducer) Publish(topic string, body []byte) error {
	producer, err := n.getProducerWithRetry()
	if nil != err {
		return err
	}
	err = producer.Publish(topic, body)
	if nil != err {
		producer, err = n.getProducerWithRetry()
		if nil != err {
			return err
		}
		return producer.Publish(topic, body)
	}
	return nil
}
func (n *nsqProducer) SetLogger(l logger, lvl nsq.LogLevel) {
	n.log = l
	n.loglvl = lvl
}
func (n *nsqProducer) getProducer() (*nsq.Producer, error) {
	if nil != n.currentProducer {
		if err := n.currentProducer.Ping(); nil == err {
			return n.currentProducer, nil
		}
		n.inuseNode.available = false
		n.inuseNode = nil
		n.currentProducer.Stop()
		n.currentProducer = nil
	}
	if nil == n.availableNodes ||
		len(n.availableNodes) == 0 {
		return nil, fmt.Errorf("No available nsqd node : node list is empty")
	}
	var producer *nsq.Producer
	var err error
	for i, v := range n.availableNodes {
		if !v.available {
		}
		producer, err = nsq.NewProducer(fmt.Sprintf("%s:%d", v.BroadcastAddress, v.TCPPort), n.cfg)
		if nil != err {
			continue
		}
		if nil != n.log {
			producer.SetLogger(n.log, n.loglvl)
		}
		err = producer.Ping()
		if nil != err {
			continue
		}
		n.availableNodes[i].inuse = true
		n.inuseNode = &n.availableNodes[i]
		n.currentProducer = producer
		break
	}
	return producer, err
}
func (n *nsqProducer) getProducerWithRetry() (*nsq.Producer, error) {
	producer, err := n.getProducer()
	if err != nil {
		n.Stop()
		for i := 0; i < maxRetryTimes; i++ {
			n.availableNodes, err = getAllAvailableNSQDFromNSQLookupds(n.lookupdAddrs)
			if nil != err {
				continue
			}
			producer, err = n.getProducer()
			if nil != err {
				break
			}
		}
		if nil != err {
			return nil, err
		}
	}
	return producer, err
}