func (r *Consumer) Stats() *ConsumerStats {
	return &ConsumerStats{
		MessagesReceived: atomic.LoadUint64(&r.messagesReceived),
		MessagesFinished: atomic.LoadUint64(&r.messagesFinished),
		MessagesRequeued: atomic.LoadUint64(&r.messagesRequeued),
		Connections:      len(r.conns()),
	}
}
Stats retrieves the current connection and message statistics for a Consumer
package nsq
import (
	"bytes"
	"errors"
	"fmt"
	"log"
	"math"
	"math/rand"
	"net"
	"net/url"
	"os"
	"strconv"
	"strings"
	"sync"
	"sync/atomic"
	"time"
)
type Handler interface {
	HandleMessage(message *Message) error
}
type HandlerFunc func(message *Message) error
func (h HandlerFunc) HandleMessage(m *Message) error {
	return h(m)
}
type DiscoveryFilter interface {
	Filter([]string) []string
}
type FailedMessageLogger interface {
	LogFailedMessage(message *Message)
}
type ConsumerStats struct {
	MessagesReceived uint64
	MessagesFinished uint64
	MessagesRequeued uint64
	Connections      int
}
var instCount int64
type backoffSignal int
const (
	backoffFlag backoffSignal = iota
	continueFlag
	resumeFlag
)
type Consumer struct {
	messagesReceived uint64
	messagesFinished uint64
	messagesRequeued uint64
	totalRdyCount    int64
	backoffDuration  int64
	backoffCounter   int32
	maxInFlight      int32
	mtx sync.RWMutex
	logger   logger
	logLvl   LogLevel
	logGuard sync.RWMutex
	behaviorDelegate interface{}
	id      int64
	topic   string
	channel string
	config  Config
	rngMtx sync.Mutex
	rng    *rand.Rand
	needRDYRedistributed int32
	backoffMtx sync.RWMutex
	incomingMessages chan *Message
	rdyRetryMtx    sync.RWMutex
	rdyRetryTimers map[string]*time.Timer
	pendingConnections map[string]*Conn
	connections        map[string]*Conn
	nsqdTCPAddrs []string
	lookupdRecheckChan chan int
	lookupdHTTPAddrs   []string
	lookupdQueryIndex  int
	wg              sync.WaitGroup
	runningHandlers int32
	stopFlag        int32
	connectedFlag   int32
	stopHandler     sync.Once
	exitHandler     sync.Once
	StopChan chan int
	exitChan chan int
}
func NewConsumer(topic string, channel string, config *Config) (*Consumer, error) {
	config.assertInitialized()
	if err := config.Validate(); err != nil {
		return nil, err
	}
	if !IsValidTopicName(topic) {
		return nil, errors.New("invalid topic name")
	}
	if !IsValidChannelName(channel) {
		return nil, errors.New("invalid channel name")
	}
	r := &Consumer{
		id: atomic.AddInt64(&instCount, 1),
		topic:   topic,
		channel: channel,
		config:  *config,
		logger:      log.New(os.Stderr, "", log.Flags()),
		logLvl:      LogLevelInfo,
		maxInFlight: int32(config.MaxInFlight),
		incomingMessages: make(chan *Message),
		rdyRetryTimers:     make(map[string]*time.Timer),
		pendingConnections: make(map[string]*Conn),
		connections:        make(map[string]*Conn),
		lookupdRecheckChan: make(chan int, 1),
		rng: rand.New(rand.NewSource(time.Now().UnixNano())),
		StopChan: make(chan int),
		exitChan: make(chan int),
	}
	r.wg.Add(1)
	go r.rdyLoop()
	return r, nil
}