nsq 在 golang 中使用

森林生灵 2020/07/01 217 0

文章目录 (?) [+]

    nsqlookupd

    https://nsq.io/components/nsqlookupd.html

    nsqlookupd 是管理拓扑信息的守护程序。客户端通过查询 nsqlookupd 以发现指定 topic 的 nsqd 生产者，并且 nsqd 节点广播 topic 和 channel。

    命令参数

    # lookupd 节点的外部地址，默认为主机名
-broadcast-address=<host>

# 配置文件路径
-config=<path>

# 用于 http 客户端通信的监听地址，默认为 0.0.0.0:4161
-http-address=<addr>:<port>

# 生产者自上次 ping 保留在活动列表中的持续时间，默认为 5m0s
-inactive-producer-timeout=5m0s

# 日志消息前缀，默认为 '[nsqlookupd] '
-log-prefix='[nsqlookupd] '

# 用于 tcp 客户端通信的监听地址，默认为 0.0.0.0:4160
-tcp-address=<addr>:<port>

# 生产者离线后保持注册名状态的持续时间
-tombstone-lifetime=45s

# 开启详细日志
-verbose

# 输出版本号
-version

    启动

    nsqlookupd

# 测试
curl http://127.0.0.1:4161/ping


    nsqd

    https://nsq.io/components/nsqd.html

    nsqd 是接收、排队并将消息转发到客户端的守护程序。

    一个 nsqd 可以创建多个 topic，每个 topic 有多个 channel，每个 channel 都能获取该 topic 的所有消息的副本。topic 和 channel 都相互独立地缓冲数据，从而防止缓慢的消费者造成 channel 的积压（在主题级别也是如此）。一个 channel 可以连接多个消费者客户端，但每条消息会随机发送其中一个消费者。

    命令参数

    # 认证服务器地址，可以为多个
-auth-http-address=<addr>:<port>

# 向 lookupd 注册当前 nsqd 节点的外部地址，默认为主机名
-broadcast-address=<host>

# 配置文件路径
-config=<config-file>

# 消息备份目录
-data-path=<path>

# 是否启用客户端协商压缩，默认为 true
-deflate=true

# 跟踪消息处理时间百分比，范围 (0, 1.0]，多次指定使用 , 分隔，默认为 none
-e2e-processing-latency-percentile='1.0,0.99,0.95'

# 计算时间段内点对点时间延迟，默认为 10m0s
-e2e-processing-latency-window-time=60s

# http 客户端监听地址，默认为 0.0.0.0:4151
-http-address=<addr>:<port>

# http 客户端连接超时时间，默认为 2s
-http-client-connect-timeout=2s

# http 客户端请求超时时间，默认为 5s
-http-client-request-timeout=5s

# https 客户端监听地址，默认为 0.0.0.0:4152
-https-address=<addr>:<port>

# 日志级别，可选级别有 debug、info、warn、error、fatal，默认为 info
-log-level=info

# 日志消息前缀，默认为 '[nsqd] '
-log-prefix='[nsqd] '

# lookupd 的 tcp 地址，可以为多个
-lookupd-tcp-address=<addr>:<port>

# 单条命令最大的正文大小，默认为 5242880
-max-body-size=5242880

# 每个磁盘队列文件的最大字节数，默认为 104857600
-max-bytes-per-file=104857600

# 每个 nsqd 实例的最大 channel 消费者连接数，默认为 0，即无限制
-max-channel-consumers=0

# 客户端最大协商压缩等级，值越大 CPU 占用越高，默认为 6
-max-deflate-level=6

# 客户端间最大心跳时间间隔，默认为 1m0s
-max-heartbeat-interval=1m0s

# 单条消息最大字节数，默认为 1048576
-max-msg-size=1048576

# 消息的最大超时时间，默认为 15m0s
-max-msg-timeout=15m0s

# 客户端输出缓冲区的最大字节数，默认为 65536
-max-output-buffer-size=65536

#  缓存数据到达客户端的最大时间段，默认为 30s
-max-output-buffer-timeout=30s

#  缓存数据到达客户端的最小时间段，默认为 25ms
-min-output-buffer-timeout=25ms

# 缓存数据到达客户端的时间段，默认为 250ms
-output-buffer-timeout=250ms

# 客户端最大 RDY 数量，默认为 2500
-max-rdy-count=2500

# 消息重新排队的最大超时时间，默认为 1h0m0s
-max-req-timeout=1h0m0s

# 保留在内存中的消息数（每个主题/通道），默认为 10000
-mem-queue-size=10000

# 自动请求消息需要等待时间，默认为 1m0s
-msg-timeout=1m0s

# 消息 ID 的唯一部分，范围为 [0,1024)
-node-id=559

# 启用客户端快速协商压缩功能，默认为 true
-snappy=true

# 用于推送统计信息的 statsd 守护程序的 UDP 地址
-statsd-address=<addr>:<port>

# 推送到 statsd 时间间隔，默认为 1m0s
-statsd-interval=1m0s

# 开启发送内存和 GC 统计信息到 statsd，默认为 true
-statsd-mem-stats=true

# 用于发送给 statsd 的密钥的前缀，使用 %s 代表主机，默认为 'nsq.%s'
-statsd-prefix='nsq.%s'

# statsd UDP 数据包字节大小，默认为 508
-statsd-udp-packet-size=508

# 每个磁盘队列 fsync 的消息数 ，默认为 2500
-sync-every=2500

# 每个磁盘队列 fsync 的持续时间，默认为 2s
-sync-timeout=2s

# 用于和 tcp 客户端通信的监听地址，默认为 0.0.0.0:4150
-tcp-address=<addr>:<port>

# 证书 cer 文件路径
-tls-cert=<path>

# 客户端证书授权策略，'require' 或 'require-verify'
-tls-client-auth-policy=

# 证书私钥 key 文件路径
-tls-key=<path>

# 可接受的最低 SSL/TLS 版本，'ssl3.0'、'tls1.0'、'tls1.1' 或 'tls1.2'
-tls-min-version=

# 是否要求 TLS 进行客户端连接，true、false 或 tcp-https
-tls-required=

# ca 文件路径
-tls-root-ca-file=<path>

# 已弃用，使用 --log-level 代替
-verbose

# 输出版本号
-version

# 已弃用，使用 --node-id 代替
-worker-id

    启动

    nsqd --broadcast-address=127.0.0.1 --lookupd-tcp-address=127.0.0.1:4160

# 测试
curl -d 'hello world' 'http://127.0.0.1:4151/pub?topic=test'


    nsq_to_file

    从 nsqlookupd 备份指定 topic 的消息。

    nsq_to_file --topic=test --output-dir=/tmp --lookupd-http-address=127.0.0.1:4161


    nsqadmin

    https://nsq.io/components/nsqadmin.html

    nsqadmin 用来收集集群实时统计信息和管理任务的 WEB UI。

    命令参数

    # 允许 http 请求访问 /config 节点的 CIDR，默认为 127.0.0.1/8
-allow-config-from-cidr='127.0.0.1/8'

# 配置文件路径
-config=<path>

# graphite 的 http 地址
-graphite-url=

# 用于和 http 客户端通信的监听地址，默认为 "0.0.0.0:4171
-http-address=<addr>:<port>

# http 客户端连接超时时间，默认为 2s
-http-client-connect-timeout=2s

# http 客户端请求超时时间，默认为 5s
-http-client-request-timeout=5s

# 用于 http 客户端的证书 cer 文件路径
-http-client-tls-cert=<path>

# 用于 http 客户端的证书私钥 key 文件路径
-http-client-tls-key=<path>

# 用于 http 客户端的 ca 证书文件路径
-http-client-tls-root-ca-file=<path>

# 为 http 客户端配置跳过验证 tls 证书
-http-client-tls-insecure-skip-verify

# 日志消息前缀，默认为 
-log-prefix='[nsqadmin] '

# lookupd 的 http 地址，可以为多个
-lookupd-http-address=

# 发送管理员操作的 POST 通知到 http 端（完全限定）
-notification-http-endpoint=

# nsqd 的 http 地址，可以为多个
-nsqd-http-address=<addr>:<port>

# 代理HTTP请求到 graphite
-proxy-graphite

# 格式化 statsd-counter，如果不需要格式化请将其设置为空，默认为 stats.counters.%s.count
-statsd-counter-format='stats.counters.%s.count'

# 格式化 statsd-gauge，如果不需要格式化请将其设置为空，默认为 stats.gauges.%s
-statsd-gauge-format='stats.gauges.%s'

# 推送到 statsd 时间间隔，必须与 nsqd 配置一致，默认为 1m0s
-statsd-interval=1m0s

# 用于发送给 statsd 的密钥的前缀，必须与 nsqd 配置一致，使用 %s 代替主机，默认为 nsq.%s
-statsd-prefix='nsq.%s'

# 输出版本号
-version

    启动

    nsqadmin --lookupd-http-address=127.0.0.1:4161


    部署

    使用 docker-compose 部署

    https://nsq.io/deployment/docker.html

    version: '3'
services:
  nsqlookupd:
    image: nsqio/nsq
    command: /nsqlookupd
    ports:
      - "4160:4160"
      - "4161:4161"
    networks:
      nsq:
        ipv4_address: 10.0.10.2
    container_name: nsqlookupd
  nsqd:
    image: nsqio/nsq
    command: /nsqd --broadcast-address=10.0.10.3 --lookupd-tcp-address=nsqlookupd:4160
    depends_on:
      - nsqlookupd
    ports:
      - "4150:4150"
      - "4151:4151"
    networks:
      nsq:
        ipv4_address: 10.0.10.3
    container_name: nsqd
  nsq_to_file:
    image: nsqio/nsq
    command: /nsq_to_file --topic=test --lookupd-http-address=nsqlookupd:4161
    depends_on:
      - nsqlookupd
    container_name: nsq_to_file
    networks:
      nsq:
        ipv4_address: 10.0.10.4
    container_name: nsq_to_file
  nsqadmin:
    image: nsqio/nsq
    command: /nsqadmin --lookupd-http-address=nsqlookupd:4161
    depends_on:
      - nsqlookupd  
    ports:
      - "4171:4171"
    networks:
      nsq:
        ipv4_address: 10.0.10.5
    container_name: nsqadmin

networks:
  nsq:
    driver: bridge
    ipam:
      config:
        - subnet: 10.0.10.0/16


    使用

    生产者

    package main

import (
    "github.com/nsqio/go-nsq"
    "log"
    "net"
    "os"
)

var logger = log.New(os.Stdout, "[producer] ", log.LstdFlags)

func main() {
    var err error
    config := nsq.NewConfig()
    config.LocalAddr, err = net.ResolveTCPAddr("tcp", "127.0.0.1:0")
    if err != nil {
        log.Fatalln(err)
    }

    w, err := nsq.NewProducer("127.0.0.1:4150", config)
    if err != nil {
        log.Fatalln(err)
    }
    w.SetLogger(logger, nsq.LogLevelInfo)
    err = w.Ping()
    if err != nil {
        log.Fatalln(err)
    }

    err = w.Publish("news_push", []byte("hello"))
    if err != nil {
        log.Fatalln(err)
    }
    w.Stop()
}

    消费者

    如果使用 docker 搭建测试环境，且使用 lookupd 来查询 nsqd 很可能会出现如下错误：

    error connecting to nsqd - dial tcp: lookup 4d846ab34d0d: Temporary failure in name resolution

    出现这个问题原因如下：

    启动 nsqd 时未使用 -broadcast-address 指定地址，默认为容器主机名，lookupd 查询时返回该主机名，然而在宿主机里并不能解析出对应 ip，需要在宿主机修改 hosts 记录或指定 -broadcast-address 为 nsqd 的主机 ip，注意不要修改为 127.0.0.1 ，这将导致处于不同容器 nsqadmin 去根据此 ip 访问容器本地，会使相关页面失效；另一种方法是使用 docker-compose 指定每个容器的 ip。

    package main

import (
    "errors"
    "github.com/nsqio/go-nsq"
    "log"
    "os"
    "time"
)

var logger = log.New(os.Stdout, "[consumer] ", log.LstdFlags)

type Handler struct{}

func (h *Handler) HandleMessage(msg *nsq.Message) error {
    if string(msg.Body) == "TOBEFAILED" {
        log.Println("failed")

        return errors.New("failed")
    }

    log.Println(msg.NSQDAddress, string(msg.Body))

    return nil
}

func main() {
    var err error
    config := nsq.NewConfig()
    config.DefaultRequeueDelay = 0
    config.MaxBackoffDuration = time.Microsecond * 50

    cs, err := nsq.NewConsumer("news_push", "ch", config)
    if err != nil {
        log.Fatalln(err)
    }
    cs.SetLogger(logger, nsq.LogLevelInfo)
    cs.AddHandler(&Handler{})

    // 直接连接 nsqd
    err = cs.ConnectToNSQD("127.0.0.1:4150")

    // 通过查询获取 nsqd 地址并连接
    // err = cs.ConnectToNSQLookupd("127.0.0.1:4161")
    if err != nil {
        log.Fatalln(err)
    }

    stat := cs.Stats()
    if stat.Connections == 0 {
        log.Fatalln("0 connection")
    }

    // err = cs.DisconnectFromNSQD("127.0.0.1:4150")
    // err = cs.DisconnectFromNSQLookupd("127.0.0.1:4161")
    // if err != nil {
    //     log.Fatalln(err)
    // }

    <-cs.StopChan
}


    Golang golang mq
    本文标题：nsq 在 golang 中使用
    本文链接：https://www.lanseyujie.com/post/application-of-nsq-in-golang.html
    作者授权：除特别说明外，本文由 森林生灵 原创编译并授权 蓝色域界 刊载发布。
    版权声明：本文使用「署名-非商业性使用-相同方式共享」创作共享协议，转载或使用请遵守署名协议。
    点赞 0 赏 分享 0

    相关文章

    上一篇：grpc 在 golang 使用
    下一篇：根据背景图片色调动态调整文字颜色