NSQ实现消息延迟执行
消息队列
13 分钟 读完 (大约 1924 个字) 0次访问

NSQ实现消息延迟执行

最近我给自己做了一个记事微信小程序,主要是怕自己把事情给忘了，虽然现在市面上有很多成熟的应用，但是作为程序猿，而且最近工作也不是那么忙，就想着自己折腾一个。既然是备忘，当然得有一个消息提醒，这就需要涉及到延迟执行。现在针对延迟执行也有许多方案，比如:定时扫描、消息队列等。定时扫描在时间上有延迟而且扫描频率影响着数据库的性能，另外还有定时轮(TimingWheel)算法，不过这里我还是选择了消息队列的方式。

支持延迟消息投递的消息队列也比较多,如:rabbitmq、rocketmq、kafka等都可以实现延迟消息。不过 NSQ具有分布式、无单点故障、故障容错、高可用性以及高可靠等特征,已有一些公司用于生产。毕竟我只有2台1核2G的服务器，所以我还是更加青睐于它。

NSQ部署

在部署NSQ之前，我们首先得了解它的三个组件nsqd、nsqlookupd、nsqadmin:

  • nsqd主要负责接收、存储和转发消息发给客户端
  • nsqlookupd主要负责nsqd的注册以及心跳、状态监控。
  • nsqadmin是消息的web管理界面

Windows部署NSQ

我们这里先实践一下Windows的安装方式，首先需要去NSQ下载Windows版本并解压.

  1. 首先启动nsqlookupd:

    1
    2
    3
    4
    E:\tool\nsq-1.2.0\bin>nsqlookupd
    [nsqlookupd] 2020/01/14 09:26:37.816513 INFO: nsqlookupd v1.2.0 (built w/go1.12.9)
    [nsqlookupd] 2020/01/14 09:26:37.882337 INFO: HTTP: listening on [::]:4161
    [nsqlookupd] 2020/01/14 09:26:37.882337 INFO: TCP: listening on [::]:4160

    默认端口为http:4161、tcp:4160,可以通过分别指定-http-address和-tcp-address两个参数来修改。

  2. 启动nsqd实例:

    1
    2
    3
    4
    5
    6
    7
    8
    9
    E:\tool\nsq-1.2.0\bin>nsqd --lookupd-tcp-address=0.0.0.0:4160 -data-path="D:/nsqdata" --broadcast-address=0.0.0.0
    [nsqd] 2020/01/14 09:38:00.365419 INFO: nsqd v1.2.0 (built w/go1.12.9)
    [nsqd] 2020/01/14 09:38:00.403363 INFO: ID: 632
    [nsqd] 2020/01/14 09:38:00.408304 INFO: NSQ: persisting topic/channel metadata to D:/nsqdata/nsqd.dat
    [nsqd] 2020/01/14 09:38:00.453186 INFO: TCP: listening on [::]:4150
    [nsqd] 2020/01/14 09:38:00.453186 INFO: HTTP: listening on [::]:4151
    [nsqd] 2020/01/14 09:38:00.453186 INFO: LOOKUP(0.0.0.0:4160): adding peer
    [nsqd] 2020/01/14 09:38:00.500058 INFO: LOOKUP connecting to 0.0.0.0:4160
    [nsqd] 2020/01/14 09:38:00.523995 INFO: LOOKUPD(0.0.0.0:4160): peer info {TCPPort:4160 HTTPPort:4161 Version:1.2.0 BroadcastAddress:2CNU7X5OLAUE004}
  • -lookupd-tcp-address:nsqlookupd的IP和tcp端口
  • -broadcast-address:会注册到nsqlookupd，填写自己主机IP
  • -data-path:消息持久化的存储路径

    NSQ支持延时消息的默认最长时间为3600000(60分钟),可以通过-max-req-timeout参数修改

  1. 最后启动我们的nsqadmin实例:
    1
    2
    3
    E:\tool\nsq-1.2.0\bin>nsqadmin --lookupd-http-address=0.0.0.0:4161
    [nsqadmin] 2020/01/14 09:38:34.709978 INFO: nsqadmin v1.2.0 (built w/go1.12.9)
    [nsqadmin] 2020/01/14 09:38:34.745941 INFO: HTTP: listening on [::]:4171
  • -lookupd-http-address:nsqlookupd的IP和http端口

上面我们演示的均是单个实例，仅适合于开发环境，以后有机会我们再来探究如何部署多实例搭建高可用NSQ。

Docker部署NSQ

NSQ支持跨平台，如果是部署在Linux的话，建议还是使用Docker部署，在NSQ官网有详细的Docker安装说明，这里我们参照官网实践一下。编辑我们的docker-compose文件,这里我命名为nsq.yml:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
version: '3'
services:
  nsqlookupd:
    image: nsqio/nsq
    command: /nsqlookupd
    ports:
      - "4160"
      - "4161"
  nsqd:
    image: nsqio/nsq
    command: /nsqd --lookupd-tcp-address=nsqlookupd:4160 –data-path=/data/nsqdata
    depends_on:
      - nsqlookupd
    ports:
      - "4150"
      - "4151"
    volumes:
      - /root/nsqdata:/data/nsqdata
  nsqadmin:
    image: nsqio/nsq
    command: /nsqadmin --lookupd-http-address=nsqlookupd:4161
    depends_on:
      - nsqlookupd  
    ports:
      - "4171"

我们稍作修改，将NSQD数据目录挂载到物理机。创建好yml配置文件后，就运行命令创建我们的容器:

1
2
3
4
5
6
7
8
9
10
11
12
[root@instance-p0a4erj8 ~]# docker-compose -f nsq.yml up -d
Creating network "root_default" with the default driver
Pulling nsqlookupd (nsqio/nsq:)...
latest: Pulling from nsqio/nsq
5d20c808ce19: Pull complete
e43ee7addbdb: Pull complete
cbc99497dda7: Pull complete
Digest: sha256:78b986254986c4ae1237b32219a83c5a23354a6c30c18817597f776a4edcac41
Status: Downloaded newer image for nsqio/nsq:latest
Creating root_nsqlookupd_1 ... done
Creating root_nsqd_1       ... done
Creating root_nsqadmin_1   ... done

执行docker-compose ps查看我们的容器:

1
2
3
4
5
6
[root@instance-p0a4erj8 ~]# docker-compose -f nsq.yml ps
      Name                     Command               State                                            Ports                                          
-----------------------------------------------------------------------------------------------------------------------------------------------------
root_nsqadmin_1     /nsqadmin --lookupd-http-a ...   Up      4150/tcp, 4151/tcp, 4160/tcp, 4161/tcp, 4170/tcp, 0.0.0.0:32772->4171/tcp               
root_nsqd_1         /nsqd --lookupd-tcp-addres ...   Up      0.0.0.0:32771->4150/tcp, 0.0.0.0:32770->4151/tcp, 4160/tcp, 4161/tcp, 4170/tcp, 4171/tcp
root_nsqlookupd_1   /nsqlookupd                      Up      4150/tcp, 4151/tcp, 0.0.0.0:32769->4160/tcp, 0.0.0.0:32768->4161/tcp, 4170/tcp, 4171/tcp

如果需要删除,需要先stop,在rm:

1
2
3
4
5
6
7
8
9
10
[root@instance-p0a4erj8 ~]# docker-compose -f nsq.yml stop
Stopping root_nsqadmin_1   ... done
Stopping root_nsqd_1       ... done
Stopping root_nsqlookupd_1 ... done
[root@instance-p0a4erj8 ~]# docker-compose -f nsq.yml rm
Going to remove root_nsqadmin_1, root_nsqd_1, root_nsqlookupd_1
Are you sure? [yN] y
Removing root_nsqadmin_1   ... done
Removing root_nsqd_1       ... done
Removing root_nsqlookupd_1 ... done

刚才看到，如果不指定端口，容器启动时随机分配绑定的主机端口号，这样就会比较乱，所以要指定端口:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
version: '3'
services:
  nsqlookupd:
    image: nsqio/nsq
    command: /nsqlookupd
    ports:
      - "14160:4160"
      - "14161:4161"
  nsqd:
    image: nsqio/nsq
    command: /nsqd --lookupd-tcp-address=nsqlookupd:4160 –data-path=/data/nsqdata
    depends_on:
      - nsqlookupd
    ports:
      - "14150:4150"
      - "14151:4151"
    volumes:
      - /root/nsqdata:/data/nsqdata
  nsqadmin:
    image: nsqio/nsq
    command: /nsqadmin --lookupd-http-address=nsqlookupd:4161
    depends_on:
      - nsqlookupd  
    ports:
      - "14171:4171"

我们在重新创建NSQ容器:

1
2
3
4
5
6
7
8
9
10
[root@instance-p0a4erj8 ~]# docker-compose -f nsq.yml up -d
Creating root_nsqlookupd_1 ... done
Creating root_nsqadmin_1   ... done
Creating root_nsqd_1       ... done
[root@instance-p0a4erj8 ~]# docker-compose -f nsq.yml ps
      Name                     Command               State                                            Ports                                          
-----------------------------------------------------------------------------------------------------------------------------------------------------
root_nsqadmin_1     /nsqadmin --lookupd-http-a ...   Up      4150/tcp, 4151/tcp, 4160/tcp, 4161/tcp, 4170/tcp, 0.0.0.0:14171->4171/tcp               
root_nsqd_1         /nsqd --lookupd-tcp-addres ...   Up      0.0.0.0:14150->4150/tcp, 0.0.0.0:14151->4151/tcp, 4160/tcp, 4161/tcp, 4170/tcp, 4171/tcp
root_nsqlookupd_1   /nsqlookupd                      Up      4150/tcp, 4151/tcp, 0.0.0.0:14160->4160/tcp, 0.0.0.0:14161->4161/tcp, 4170/tcp, 4171/tcp

现在我们就可以查看NSQ的状态:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
[root@instance-p0a4erj8 ~]# curl http://127.0.0.1:14151/stats
nsqd v1.2.0 (built w/go1.12.9)
start_time 2020-01-14T08:50:35Z
uptime 1m9.272087092s

Health: OK

Memory:
   heap_objects                 1863
   heap_idle_bytes              65486848
   heap_in_use_bytes            1130496
   heap_released_bytes          0
   gc_pause_usec_100            0
   gc_pause_usec_99             0
   gc_pause_usec_95             0
   next_gc_bytes                4473924
   gc_total_runs                0

Topics: None

Producers: None

关于NSQ AUTH方面的知识点，大家可以看看https://nsq.io/components/nsqd.html#auth。

代码示例

在NSQ官网我们可以看到支持多种语言，这里我还是使用Golang,Producer:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
func main()  {
	config := nsq.NewConfig()
	topicName := "nsq_demo"
	msgCount := 10
	producer, _ := nsq.NewProducer("127.0.0.1:4150", config)

	err := producer.Ping()
	if err != nil {
		log.Fatal("should not be able to ping after Stop()")
		return
	}
	defer producer.Stop()
	for i := 1; i < msgCount; i++ {
        fmt.Println(fmt.Sprintf("test nsq message index:%d", i))
		err1 := producer.Publish(topicName, []byte(fmt.Sprintf("test nsq message index:%d", i)))
		if err1 != nil {
			log.Fatal("error",err1)
		}
	}
}

Comsumer:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
type ComsumerHandler struct {

}

func (h *ComsumerHandler) HandleMessage(message *nsq.Message) error {
	if string(message.Body) == "TOBEFAILED" {
		return errors.New("fail this message")
	}
	fmt.Println("receive", message.NSQDAddress, "message:", string(message.Body))
	return nil
}

func main()  {

	waiter := sync.WaitGroup{}
	waiter.Add(1)

	go func() {
		defer waiter.Done()

		config := nsq.NewConfig()
		laddr := "127.0.0.1"
		// so that the test can simulate binding consumer to specified address
		config.LocalAddr, _ = net.ResolveTCPAddr("tcp", laddr+":0")
		// so that the test can simulate reaching max requeues and a call to LogFailedMessage
		config.DefaultRequeueDelay = 0
		// so that the test wont timeout from backing off
		config.MaxBackoffDuration = time.Millisecond * 50

		topicName := "nsq_demo"
		if config.Deflate {
			topicName = topicName + "_deflate"
		} else if config.Snappy {
			topicName = topicName + "_snappy"
		}
		if config.TlsV1 {
			topicName = topicName + "_tls"
		}
		consumer, _ := nsq.NewConsumer(topicName, "ch", config)


		handler := &ComsumerHandler{
		}
		consumer.AddHandler(handler)
		err := consumer.ConnectToNSQD("127.0.0.1:4150")
		if nil != err {
			fmt.Println("err", err)
			return
		}
		select{}

	}()
	waiter.Wait()
}

,
我主要的目的是实现延迟执行，所以这里需要尝试NSQ延迟消息投递功能,我们稍作修改:
Producer:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
unc main() {
	config := nsq.NewConfig()
	topicName := "nsq_demo"
	msgCount := 10
	producer, _ := nsq.NewProducer("127.0.0.1:4150", config)

	err := producer.Ping()
	if err != nil {
		log.Fatal("should not be able to ping after Stop()")
		return
	}
	defer producer.Stop()
	for i := 1; i < msgCount; i++ {
		fmt.Println(fmt.Sprintf("test nsq message index:%d", i))
		//err1 := producer.Publish(topicName, []byte(fmt.Sprintf("test nsq message index:%d", i)))
		delay := rand.Intn(10)
        //延时消息，delay为延迟多少秒投递。
		err1 := producer.DeferredPublish(topicName, time.Second * time.Duration(delay), []byte(fmt.Sprintf("test nsq message index:%d,time:%s,delay:%d", i, time.Now().Format("2020-01-14 15:04:05"), delay)))
		if err1 != nil {
			log.Fatal("error", err1)
		}
	}
}

Comsumer可以不用修改，但是我还是加上时间打印:

1
fmt.Println("time",time.Now().Format("2020-01-14 15:04:05"),"receive", message.NSQDAddress, "message:", string(message.Body))

我们再来看看效果:

1
2
3
4
5
6
7
8
9
time 14140-01-119 16:19:44 receive 127.0.0.1:4150 message: test nsq message index:8,time:14140-01-119 16:19:44,delay:0
time 14140-01-119 16:19:45 receive 127.0.0.1:4150 message: test nsq message index:1,time:14140-01-119 16:19:44,delay:1
time 14140-01-119 16:19:45 receive 127.0.0.1:4150 message: test nsq message index:5,time:14140-01-119 16:19:44,delay:1
time 14140-01-119 16:19:49 receive 127.0.0.1:4150 message: test nsq message index:7,time:14140-01-119 16:19:44,delay:5
time 14140-01-119 16:19:50 receive 127.0.0.1:4150 message: test nsq message index:9,time:14140-01-119 16:19:44,delay:6
time 14140-01-119 16:19:51 receive 127.0.0.1:4150 message: test nsq message index:2,time:14140-01-119 16:19:44,delay:7
time 14140-01-119 16:19:51 receive 127.0.0.1:4150 message: test nsq message index:3,time:14140-01-119 16:19:44,delay:7
time 14140-01-119 16:19:52 receive 127.0.0.1:4150 message: test nsq message index:6,time:14140-01-119 16:19:44,delay:8
time 14140-01-119 16:19:53 receive 127.0.0.1:4150 message: test nsq message index:4,time:14140-01-119 16:19:44,delay:9

# ,

喜欢这篇文章？打赏一下作者吧

You forgot to set the qrcode for Alipay. Please set it in _config.yml.
You forgot to set the qrcode for Wechat. Please set it in _config.yml.
You forgot to set the business and currency_code for Paypal. Please set it in _config.yml.
You forgot to set the url Patreon. Please set it in _config.yml.
NSQ的可用性和可靠性
.Net Core 3.X WebApi 自宿主并注册成Windows服务
关注我

链接

分类

标签云

.net .net core ABP Datax Docker Docker命令 ECMAScript Gis Git Github Go Golang Go基础 Hexo Java JavaScript Prometheus Redis Spring Boot Swagger Tool Vue ack aop autofac css docker ioc k8s mysql explain nginx other python redash shell swagger tool tools 微信 微信公众号 总结 数据同步 文件存储 消息队列

最新文章

归档

标签

.net 1
.net core 13
ABP 1
Datax 1
Docker 2
Docker命令 2
ECMAScript 1
Gis 1
Git 4
Github 4
Go 2
Golang 4
Go基础 4
Hexo 3
Java 20
JavaScript 2
Prometheus 1
Redis 2
Spring Boot 15
Swagger 2
Tool 10
Vue 4
ack 1
aop 1
autofac 2
css 1
docker 2
ioc 1
k8s 1
mysql explain 1
nginx 5
other 1
python 1
redash 1
shell 1
swagger 2
tool 1
tools 1
微信 2
微信公众号 1
总结 1
数据同步 1
文件存储 1
消息队列 2

最新文章

归档

标签

.net 1
.net core 13
ABP 1
Datax 1
Docker 2
Docker命令 2
ECMAScript 1
Gis 1
Git 4
Github 4
Go 2
Golang 4
Go基础 4
Hexo 3
Java 20
JavaScript 2
Prometheus 1
Redis 2
Spring Boot 15
Swagger 2
Tool 10
Vue 4
ack 1
aop 1
autofac 2
css 1
docker 2
ioc 1
k8s 1
mysql explain 1
nginx 5
other 1
python 1
redash 1
shell 1
swagger 2
tool 1
tools 1
微信 2
微信公众号 1
总结 1
数据同步 1
文件存储 1
消息队列 2
Your browser is out-of-date!

Update your browser to view this website correctly. Update my browser now

×