go使用RabbitMQ完成消息异步通信实践问题并解决记录_rabbitmq go

前言

背景：因区块链项目需求，在同时间高并发大数据环境没法实时同步上链数据结果返回给业务后端（java实现的业务后端），所以考虑采用异步通信中间件，研发阶段选择了自己搭建rabbitMQ完成我们的需求替换到API请求返回的方式，做项目技术改造。涉及到跨语言通信，消息中间件，模式选择，消息体接收转换联调等一系列问题，仅此记录。
需求：
业务后端（java）提供上链信息结构体【生产者–发送消息】–>到go后端调用区块链上链方法【消费者–消费消息】–>调用上链智能合约返回结果后，根据不同情况返回上链结果【生产者–发送消息给java后端】

技术选择

• 消息中间件：rabbitMQ(不多赘述)
  模式选择：Routing路由模式（更改调整版本，后续踩坑原因之一 ）
  交换机类型：direct（直连）
  确保消息可靠性：选择手动确认ack方式（又一个坑）
• 业务后端：java
• 上链业务处理：go
• 区块链底层：fisco-bcos

RabbitMQ

RabbitMQ是采用Erlang编程语言实现了高级消息队列协议AMQP (Advanced Message Queuing Protocol)的开源消息代理软件（消息队列中间件）消息队列中间件的作用：解耦、削峰、异步处理、缓存存储、消息通信、提高系统拓展性。

RabbitMQ特点

• 可靠性：通过一些机制例如,持久化,传输确认等来确保消息传递的可靠性
• 拓展性：多个RabbitMQ节点可以组成集群
• 高可用性：队列可以在RabbitMQ集群中设置镜像,如此一来即使部分节点挂掉了,但是队列仍然可以使用
• 多种协议：原生的支持AMQP,也能支持STOMP,MQTT等协议
• 丰富的客户端：我们常用的编程语言都支持RabbitMQ
• 管理界面：自带提供一个WEB管理界面
• 插件机制：RabbitMQ 自己提供了很多插件,可以按需要进行拓展 Plugins

问题及其解决

问题一、Routing路由模式下历史消息不能消费

因项目需求，我们选择了routing路由模式，但是在测试的时候发现go后端不能拿到历史队列里的历史数据（即在消费协程未开启时由java后端发起的消息），只能拿到消费协程开启后再往队列里新加入的消息体。

问题二、Routing路由模式下手动ack失效

上一个问题的衍生，当消费协程开启后可以拿到的消息体，当选择自动ack确认时，消费还是没有确认消费成功，改成手动ack任然失效，苦恼

问题一、问题二错误原因

最终排查的原因竟然是因为我们对路由模式的一个小改动造成的后续连锁反应。
在官网还是其他参考博文中，路由模式的创建生产者只需要创建mq实例的时候给交换机名称exchangeName和routingkey即可，这时候消息都会放到临时队列里，即需要用到就当即创建，不用到的时候就自动删除。临时队列的声明很简单，只需要在队列声明的时候队列名字写成空串即可，这样该临时队列就会被分配一个随机的名字，如：amq.gen-JzTY20BRgKO-HjmUJj0wLg。在连接关闭的时候该临时队列会被删除

var (
	mqURL = viper.GetString("RabbitMQ.Url") //连接信息读取配置文件 amqp:// 账号 密码@地址:端口号/vhost
)

//RabbitMQ结构体
type RabbitMQ struct {
	//连接
	conn    *amqp.Connection
	channel *amqp.Channel
	//队列
	QueueName string
	//交换机名称
	ExChange string
	//绑定的key名称
	Key string
	//连接的信息，上面已经定义好了
	MqUrl string
}
//rabbitmq的路由模式。
//主要特点不仅一个消息可以被多个消费者消费还可以由生产端指定消费者。
//这里相对比订阅模式就多了一个routingkey的设计，也是通过这个来指定消费者的。
//创建exchange的kind需要是"direct",不然就不是roting模式了。
//创建rabbitmq实例，这里有了routingkey为参数了。
func NewRabbitMqRouting(exchangeName string, routingKey string) *RabbitMQ {
	return NewRabbitMQ("", exchangeName, routingKey)
}

//创建RabbitMQ结构体实例
func NewRabbitMQ(queueName string, exchange string, key string) *RabbitMQ {
	rabbitMQ := &RabbitMQ{QueueName: queueName, ExChange: exchange, Key: key, MqUrl: mqURL}
	var err error
	//创建rabbitMQ连接
	rabbitMQ.conn, err = amqp.Dial(rabbitMQ.MqUrl)
	rabbitMQ.failOnErr(err, "创建rabbit的路由实例的时候连接出现问题！")
	rabbitMQ.channel, err = rabbitMQ.conn.Channel()
	rabbitMQ.failOnErr(err, "创建rabbitmq的路由实例时获取channel出错")
	return rabbitMQ
}
//错误处理函数
func (r *RabbitMQ) failOnErr(err error, message string) {
	if err != nil {
		log.Fatalf("%s:%s", message, err)
		panic(fmt.Sprintf("%s:%s", message, err))
	}
}
//断开channel和connection
func (r *RabbitMQ) Destory() {
	r.channel.Close()
	r.conn.Close()
}
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

但是为了便于后续定位问题且众多java关于消息中间件的博文，java后端开发的同事选择创建指定队列名称并与交换机绑定。生产者发送消息都没有问题，但是在我接收消息消费时候缺遇到了问题一和问题二描述的情况

解决方式

在消费者方法里与官方的稍微不同，注释掉创建交换机、创建队列、绑定队列和交换机这段代码，直接从mq实例的通道里获取消费信息即可（前提：执行前需要先创建好队列、交换机、绑定交换机和队列关系，可以在控制面板创建或者代码创建，总之一定要先有才行）

//路由模式接收信息-测试
func (r *RabbitMQ) ReceiveRouting() {
	尝试创建交换机，不存在创建
	//err := r.channel.ExchangeDeclare(
	//	//交换机名称
	//	r.ExChange,
	//	//交换机类型 广播类型
	//	"direct",
	//	//是否持久化
	//	true,
	//	//是否字段删除
	//	false,
	//	//true表示这个exchange不可以被client用来推送消息，仅用来进行exchange和exchange之间的绑定
	//	false,
	//	//是否阻塞 true表示要等待服务器的响应
	//	false,
	//	nil,
	//)
	//r.failOnErr(err, "failed to declare an excha"+"nge")
	2试探性创建队列，创建队列
	//q, err := r.channel.QueueDeclare(
	//	"", //随机生产队列名称
	//	false,
	//	false,
	//	true,
	//	false,
	//	nil,
	//)
	//r.failOnErr(err, "Failed to declare a queue")
	绑定队列到exchange中
	//err = r.channel.QueueBind(
	//	q.Name,
	//	//在pub/sub模式下，这里的key要为空
	//	r.Key,
	//	r.ExChange,
	//	false,
	//	nil,
	//)
	//消费消息
	message, err := r.channel.Consume(
		"queue_data_chain_reply",
		"",
		false, // 是否自动确认消费
		false,
		false,
		false,
		nil,
	)
	forever := make(chan bool)
	go func() {
		for d := range message {
			//chainDataRepBody := new(model.ChainDataRep)
			//err = json.Unmarshal(d.Body, &chainDataRepBody)
			d.Ack(true) // 更改为手动确认ack
			result := make(map[string]interface{})
			err = json.Unmarshal(d.Body, &result)
			if err != nil {
				fmt.Println(err)
			}
			for k, v := range result {
				fmt.Println(k)
				fmt.Println(v)
			}
			//fmt.Printf("消费的id是%s", chainDataRepBody.ID)
			fmt.Printf("消费的数据是%s", d.Body)
			//log.Printf("消费消息:%s,", d.Body)
		}
	}()
	fmt.Println("退出请按 Ctrl+C")
	<-forever
}

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
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72

如上述测试代码一样即可解决历史消费不能消费，且保存手动确认ack

问题三、如何修改rabbitMQ的手动ack确认

问：消息确认问题，假设有一个消费者正在处理消息，消息还没处理完毕突然就中断了（可能该消费者进程被杀死了），这时候该消费者正在处理的消息就丢失了，应该怎么解决这种消息丢失的问题？

解决方式

答：这时候不关生产者的事了，主要是消费者这边的设置。在消费者端的代码中的Consume方法下的autoAck自动确认字段设置为false，表示我们不做自动确认。而是在消息处理完毕之后手动确认，这样如果消息没处理完就中断了，RabbitMQ服务器没有收到消息的确认，那么队列中的该消息将会再次被重新分配。消费者端修改如下（代码上述相同）

问题四、传输信息类型在java\go中转换

因为go的amqp消息体是[]byte类型，所以我们的对象或者结构体在go之间传递都需要将其dataBytes, err := json.Marshal(data)转为字节类型存储，接收后用json.Unmarshal()反序列化解析即可使用

result := make(map[string]interface{})
err = json.Unmarshal(d.Body, &result)
1
2

但是我使用的go语言需要和java后端进行通信，java byte 的范围是 -128-127，而golang byte 是uint8的别名，范围是0-255，所以会有类型不匹配转换的问题

解决方式

java生产者发送信息–>go消费者消费消费

java后端作为生产者的时候发送消息仍然用json类型放入消息队列，从go中拿出的消息体json.Unmarshal()即可转化为指定的对象结构了

go生产者发送消息–>java消费者消费消息

go作为生产者的时候发送消息将对象json.Marshal()序列化，但需要修改路由模式发送消息amqp.Publishing的ContentType消息类型为application/json即可（官方案例或者其他博文都是text/plain类型）

路由模式发送消息

//路由模式发送信息
func (r *RabbitMQ) PublishRouting(message []byte) {
	//第一步，尝试创建交换机，与pub/sub模式不同的是这里的kind需要是direct
	err := r.channel.ExchangeDeclare(
		//交换机名称
		r.ExChange,
		//交换机类型 广播类型
		"direct",
		//是否持久化
		true,
		//是否字段删除
		false,
		//true表示这个exchange不可以被client用来推送消息，仅用来进行exchange和exchange之间的绑定
		false,
		//是否阻塞 true表示要等待服务器的响应
		false,
		nil,
	)
	r.failOnErr(err, "路由模式，尝试创建交换机失败")
	//发送信息
	err = r.channel.Publish(
		r.ExChange,
		//要设置
		r.Key,
		false,
		false,
		amqp.Publishing{
			//类型
			//ContentType: "text/plain", // byte类型
			ContentType: "application/json", // 更改为json类型
			//消息
			Body: message,
		})
}
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

问题五、协程消费消息逻辑代码中再往MQ中发送消息失效

因业务需要在开启协程处理队列的消息时，需要再将处理结果再发送到另外一个队列中（即在承担MQ消费者处理业务逻辑中还要作为另个MQ实例的生产者发送消息），因在消费中是手动ack模式，我之前的处理顺序是

1. 协程中作为消费者接收消息体
2. 处理业务逻辑
3. 封装返回结构体
4. 作为生产者发送返回消息到另外一个队列中
5. 手动ack确认消息
   这时候就遇到问题，消息已经被确认消费了，但是没有往另外一个队列发送消息。

解决方式

更换步骤4、5的代码，先手动ack确认消费再发送消费问题解决，虽然我也不懂这是什么逻辑，仅作为问题记录，如有知晓的欢迎交流。

问题六、接收mq消息体内部分字段缺失

一个低级错误，但记录提醒一下自己，在与java后端沟通的对象结构体交互中有int类型的两个字段，作为上链状态判断，但在go作为生产者发送到队列后，java后端作为消费者接收时发现这两个int类型的字段缺失了，key和value都没有。

解决方式

本来还以为又是语言不一致对int类型传输造成的奇奇怪怪问题，后来发现是我在结构体的json命令忘了更改（举例如下图这样），两个结构体的json是一样的，导致传递到mq队列中都缺失了，低级错误，但写了迷糊了容易犯这种低级错误还困扰自己，记录提醒自己！！

问题七、压测导致消息通道堵塞死循环，消息体丢失，手动ack失效

原本以为解决上述重重问题基本可以完成业务需求，但奈何没迈过压测这关，当我们压测关键业务消息进队–消费–返回这个逻辑时候，用tps2000的强度压测的时候，出现消费到1000条左右的时候拿到一条没有消息体的信息，然后手动ack还失效，导致通道堵塞死循环卡住了

解决方式：设置qos参数 prefetch 解决消息堵塞问题

// 启用QoS，每次预取5条消息，避免消息处理不过来，全部堆积在本地缓存里
r.channel.Qos(
		//每次队列只消费一个消息 这个消息处理不完服务器不会发送第二个消息过来
		//当前消费者一次能接受的最大消息数量-自定义设置
		1,
		//服务器传递的最大容量
		0,
		//如果为true 对channel可用 false则只对当前队列可用
		false,
	)
1
2
3
4
5
6
7
8
9
10

开启QoS，当RabbitMQ的队列达到5条Unacked消息时，不会再推送消息给Consumer；解决没有及时ack
prefetch是指单一消费者最多能消费的unacked messages数目

q为每一个 consumer设置一个缓冲区，大小就是prefetch。每次收到一条消息，MQ会把消息推送到缓存区中，然后再推送给客户端。当收到一个ack消息时(consumer 发出baseack指令)，mq会从缓冲区中空出一个位置，然后加入新的消息。但是这时候如果缓冲区是满的，MQ将进入堵塞状态。
更具体点描述，假设prefetch值设为10，共有两个consumer。也就是说每个consumer每次会从queue中预抓取 10 条消息到本地缓存着等待消费。同时该channel的unacked数变为20。而Rabbit投递的顺序是，先为consumer1投递满10个message，再往consumer2投递10个message。如果这时有新message需要投递，先判断channel的unacked数是否等于20，如果是则不会将消息投递到consumer中，message继续呆在queue中。之后其中consumer对一条消息进行ack，unacked此时等于19，Rabbit就判断哪个consumer的unacked少于10，就投递到哪个consumer中。

设置手动ack之后再配合qos就能解决我的问题啦，压测也不会再出类似问题。

优秀轮子

在我解决上述所有问题之后发现，如果客户端大数据量同时上链，因为复杂的处理mq的消费速率不够理想，2000条消息居然要半小时至一小时才消费完成，虽然不会有其他异常错误，但是性能还是太差，应用起来体验难免不够好，于是在我的检索之后，发现一个优秀的轮子，大佬已经封装的非常好，且考虑到连接池channel复用、错误重试、死性队列、集群轮询等其他各方面的处理。不得不说上述也得确是自己根据官网官方案例和有限的go关于rabbitmq的资料整合的可用版本，但性能得确不够优秀，参考大佬的代码之后又学到蛮多，就记录分享一波好用的轮子给大家。
tym_hmm / rabbitmq-pool-go
我是将代码拉倒本地，再进行根据项目需求调整过，进过自己压测，性能直接迈n个等级，大佬也有写是应用到生产环境，5200W请求 qbs 3000 时， 连接池显示无压力，也做了集群部署的，亲测好用呀。（ps:使用的前提也得消化好官方的基础写法和一些参数的意义才行）

参考文章

一些文章参考，包括一些模式举例都很详细
go操作RabbitMQ
rabbitMQ官网路由模式go语言
【go-zero】go-zero 与 amqp go整合 Rabbitmq 实现消息推送 go 消息队列 （best practice）
golang实现rabbitmq消息队列
rabbitmq系列文章
amqp
rabbitmq使用
Golang之消息队列——RabbitMQ的使用
RabbitMQ常见问题
RabbitMq qos prefetch 消息堵塞问题
RabbitMQ清除积压消息/管理界面出现 Unacked 消息