发表更新消息队列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
13
[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
7
[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
11
12
[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
11
[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
22
[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

NSQ实现消息延迟执行

https://blogs.52fx.biz/posts/3853429767.html

作者

eyiadmin

发布于

2020-01-13

更新于

2024-05-31

许可协议

#

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

评论

关注我

目录

链接

标签

.net1
.net core13
ABP1
ASP.NET Core 6.014
C#3
CSS,前端3
DB1
DBeaver1
DBeaver下载驱动失败1
DBeaver设置Maven镜像仓库1
Datax2
Docker14
Docker命令2
ECMAScript1
EF Core2
ES61
Gis1
Git5
Github4
Go2
Golang11
Go基础7
Hexo3
Hutool1
IDEA插件1
Java26
Java81
JavaScript2
Js1
Linux3
Linux,Java1
MariaDB1
Meaven仓库地址1
MyBatis1
MySQL5
Mybatis4
Nginx4
Nginx安装1
Node常见问题1
NuGet1
Prometheus1
Redis3
Redisson1
Redis分布式锁1
RocketMQ1
Spring Boot37
Spring Boot,Java1
Spring Boot异步1
Spring Cloud Alibaba1
SpringBoot1
Swagger3
Tool10
Uniapp1
Vue5
WinDbg1
ack1
aop1
autofac2
docker2
ioc1
k8s1
mysql explain1
nginx5
other1
paddleocr1
python1
redash1
shell1
swagger2
tool1
免费工具1
其它1
前端1
前端，Html51
发版1
开源52
开源,消息队列1
开源工具5
微信1
微信公众号1
数据同步1
文件存储2
易企签1
消息队列3
第三方组件1
给Spring Boot的Jar包瘦身1
自动化测试1

订阅更新

follow.it

最新文章

归档

分类

×