【代码篇】从零开始一步步搭建自己的golang框架(七)

这个框架的目标是作为一个通用的框架,我希望它大而全,在日后可以直接作为我其他项目的基础模板,所以我还想继续给他添加一些功能,就当写一些demo进去吧。这篇文章,我将会添加一个队列的功能。

nsq

队列有很多种,我选择nsq。使用nsq需要知道以下几个概念:

  1. nsqd:负责维护队列的组件,接受消息排队和投递;
  2. nsqlookupd:管理nsq集群的组件;
  3. nsqadmin:nsq的web管理组件;
  4. topic:消息的集合。产生消息,需要指定该消息属于哪个topic的;
  5. channel:队列消息的副本。nsq消费者与nsqd或者nsqlookupd建立一个channel,并监听channel中的消息,从而实现消费。

了解了一些nsq的基础概念之后,我们先把nsq环境搭建起来,这里我们还是用docker-compose好了,以下是nsq的docker-compose.yaml。

version: ‘3‘
services:
  nsqlookupd:
    image: nsqio/nsq
    command: /nsqlookupd
    networks:
      - nsq-network
    hostname: nsqlookupd
    ports:
      - "4161:4161"
      - "4160:4160"
  nsqd:
    image: nsqio/nsq
    command: /nsqd --lookupd-tcp-address=nsqlookupd:4160 --broadcast-address=nsqd
    depends_on:
      - nsqlookupd
    hostname: nsqd
    networks:
      - nsq-network
    ports:
      - "4151:4151"
      - "4150:4150"
  nsqadmin:
    image: nsqio/nsq
    command: /nsqadmin --lookupd-http-address=nsqlookupd:4161
    depends_on:
      - nsqlookupd
    hostname: nsqadmin
    ports:
      - "4171:4171"
    networks:
      - nsq-network

networks:
  nsq-network:
    driver: bridge

启动了nsq服务之后,我们简单地验证一下,发布一条消息到nsq中:

curl -d ‘hello awesome‘ ‘http://127.0.0.1:4151/pub?topic=awesome‘

浏览器访问:http://localhost:4171/,我们也可以看到有相应的topic和产生。
【代码篇】从零开始一步步搭建自己的golang框架(七)

接着我们修改一下代码,老规矩,先改配置:

"nsq_config":
  "topic": "awesome"
  "channel": "ch"
  "nsqlookupd_addr": "127.0.0.1:4161"
type NsqConfig struct {
    Topic          string `yaml:"topic"`
    Channel        string `yaml:"channel"`
    NsqLookupdAddr string `yaml:"nsqlookupd_addr"`
}

然后在mq.go中添加一个启动mq消费者的函数:

func StartMqServer() {
    conf := nsq.NewConfig()
    q, _ := nsq.NewConsumer(config.GetConfig().NsqConfig.Topic, config.GetConfig().NsqConfig.Channel, conf)
    q.AddHandler(nsq.HandlerFunc(func(message *nsq.Message) error {
        //do something when you receive a message
        logger.GetLogger().Info("receive", zap.String(config.GetConfig().NsqConfig.Topic, string(message.Body)))
        return nil
    }))
    err := q.ConnectToNSQLookupd(config.GetConfig().NsqConfig.NsqLookupdAddr)
    if err != nil {
        logger.GetLogger().Error("connect to nsqlookupd failed.", zap.Error(err))
        os.Exit(-1)
    }
}

同样在入口文件,启动它:

go mq.StartMqServer()

最后,我们写个生产者,生成一条消息,测试一下:

func TestProducer(t *testing.T)  {
    config := nsq.NewConfig()
    p, err := nsq.NewProducer("127.0.0.1:4150", config)

    if err != nil {
        log.Panic(err)
    }

    err = p.Publish("awesome", []byte("hello awesome"))
    if err != nil {
        log.Panic(err)
    }
}

完整的代码请见:https://github.com/TomatoMr/awesomeframework


欢迎关注我的公众号:onepunchgo,给我留言。

【代码篇】从零开始一步步搭建自己的golang框架(七)

相关推荐