nsq基础知识与简单demo

// producer
package nsq

import (
	"context"
	"fmt"
	"time"

	"github.com/nsqio/go-nsq"
	"github.com/spf13/cast"
)

// 主函数
func Send(ctx context.Context, cancel context.CancelFunc, topic string) {
	defer cancel()
	str := "127.0.0.1:4150"
	fmt.Println("address: ", str)
	producer, err := nsq.NewProducer(str, nsq.NewConfig())
	if err != nil {
		panic(err)
	}

	producer.SetLogger(nil, 0)

	for i := 0; i < 5; i++ {
		msg := "puresai, " + cast.ToString(i)
		fmt.Println("publish", msg, producer.Publish(topic, []byte(msg)))
		time.Sleep(time.Second * 1)
	}

	<-ctx.Done()
	producer.Stop()
	fmt.Println("producer exit")
}

// consumer
package nsq

import (
	"context"
	"fmt"
	"time"

	"github.com/nsqio/go-nsq"
)

// 消费者
type Consumer struct{}

// 主函数
func Receive(ctx context.Context, cancel context.CancelFunc, topic string) {
	defer cancel()
	// address := "127.0.0.1:4161"

	channel := topic + "-channel"
	cfg := nsq.NewConfig()
	cfg.LookupdPollInterval = time.Second * 2
	c, err := nsq.NewConsumer(topic, channel, cfg)
	if err != nil {
		panic(err)
	}
	c.SetLogger(nil, 0) //屏蔽系统日志
	c.AddConcurrentHandlers(&Consumer{}, 3)

	//建立NSQLookupd连接
	// if err := c.ConnectToNSQLookupd(address); err != nil {
	// 	panic(err)
	// }

	//建立多个nsqd连接
	if err := c.ConnectToNSQDs([]string{"127.0.0.1:4150"}); err != nil {
		panic(err)
	}
	<-ctx.Done()
	c.Stop()
	fmt.Println("consumer exit")
}

// 处理消息
func (*Consumer) HandleMessage(msg *nsq.Message) error {
	fmt.Println("receive", msg.NSQDAddress, "message:", string(msg.Body))
	return nil
}

// nsq_test
package nsq

import (
	"context"
	"os"
	"os/signal"
	"syscall"
	"testing"
	"time"
)

func TestReceive(t *testing.T) {
	topic := "sai0556"
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()
	go Send(ctx, cancel, topic)
	go Receive(ctx, cancel, topic)

	sig := make(chan os.Signal)
	signal.Notify(sig, syscall.SIGTERM, syscall.SIGINT)
	t.Log("开始监听")

	select {
	case <-ctx.Done():
		t.Log("ctx done")
		return
	case <-sig:
		t.Log("signal exit...")
		cancel()
		time.Sleep(2 * time.Second)
		return
	}

	// send(topic)

}