Score
0
Watch 81 Star 551 Fork 157

张奇峰 / GoSkeletonGoMIT

Create your Gitee Account
Explore and code with more than 5 million developers,Free private repositories !:)
Sign up
Clone or download
rabbitmq.md 10.25 KB
Copy Edit Web IDE Raw Blame History

消息队列(RabbitMq)概要

1.本文档主要按照本人的理解介绍RabbitMq的功能、如何使用。
2.关于RabbitMq的各种使用场景以及与其他同类产品的横向、纵向对比请自行百度。
3.消息队列看起来貌似非常复杂,感觉很麻烦,其实通过本项目骨架封装之后,使用非常简单,开发者安装rabbitmq(类似安装mysql),配置好账号、密码、端口即可快速使用.
4.1分钟快速安装参考地址:https://www.jianshu.com/p/66551708ad15
5.消息队列的两个核心角色:生产者(通常是一次性投递消息),消费者(需要一直处于阻塞状态监听、接受、处理消息)。
6.关于消费者如何启动问题:
(a)开发完成消费者代码,在程序启动处(BootStrap/Init.go)通过导入包初始化形式启动(该模式相当于与本项目骨架捆绑启动)。
(b)程序cmd目录创建相关功能分类、入口文件,调用相关的消费者程序,独立编译、启动。
(c)本项目骨架引入了cobra包,同样可以做到独立编译启动。

RabbitMq常用的几种模式

全场景图

1.hello_world模式(最基本模式), 特点如下:

1 一个生产者(producer)、一个消费者(consumer)通过队列(queue)进行 一对一 的数据传输。
2.使用非常简单,适合简单业务场景使用,相关的场景模型图:
场景图

2.WorkQueue模式(在消费者之间按照竞争力分配任务), 特点如下:

1 生产者(producer)、多个消费者(consumer)通过队列(queue)进行一对多、多对多的数据传输。
2.生产者(producer)将消息发布到交换机(exchange)的某个队列(queue),多个消费者(consumer)其中只要有一个消费(取走)了消息,那么其他消费者(consumer)将不会重复获得。
3.消费者支持设置更多的参数,使配置强的消费者可以多处理消息,配置低的可以少处理消息,做到尽其所能,资源最大化利用。
场景图

3.publish/subscribe模式(同时向许多消费者发送消息), 特点如下:

1 生产者(producer)、多个消费者(consumer)通过队列(queue)进行一对多、多对多的数据传输。
2.生产者(producer)将消息发布到交换机(exchange)的某个队列(queue),多个消费者(consumer)处理消息。
3.该模式也叫作广播(broadcast)、扇形(fanout)、发布/订阅模式,消费者(consumer)可以通过配置,接收来自生产者(consumer)发送的全部消息;或者每种消费者只接收指定队列的消息,将生产者发送的消息进行分类(按照不同的队列)处理。
场景图

4.routing模式(有选择性地接收消息), 特点如下:

1 生产者(producer)、多个消费者(consumer)通过队列(queue)进行一对多、多对多的数据传输。
2.生产者(producer)将消息发布到交换机(exchange)已经绑定好路由键的某个队列(queue),多个消费者(consumer)可以通过绑定的路由键获取消息、处理消息。
3.该模式下,消息的分类应该应该明确、种类数量不是非常多,那么就可以指定路由键(key)、绑定的到交换器的队列实现消息精准投递。
场景图

5.topics模式(基于主题接收消息), 特点如下:

1 该模式就是routing模式的加强版,由原来的路由键精确匹配模式升级现在的模糊匹配模式。
2.语法层面主要表现为灵活的匹配规则: 2.1 # 表示匹配一个或多个任意字符; 2.2 *表示匹配一个字符; 2.3 .(点)本身无实际意义,不表示任何匹配规则,主要用于将关键词分隔开,它的左边或右边可以写匹配规则,例如:abc.# 表示匹配abc张三、abc你好等;#.abc.# 表示匹配路由键中含有abc的字符;
3.注意:匹配语法中如果没有 .(点),那么匹配规则是无效的,例如:orange#,可能本意是匹配orange任意字符,实际上除了匹配 orange#本身之外,什么也匹配不到。
场景图

6.RPC模式(请求、回复), 特点如下:

1 严格地说,该模式和消息队列没有什么关系,通常是微服务场景才会使用远程过程调用(RPC),本功能建议自行学习或者选择专业的微服务框架使用,解决实际问题,本文档不做介绍。
场景图

RabbitMq快速使用指南

1.建议使用docker 快速安装使用即可,安装步骤请自行搜索。
2.详细使用指南参见单元测试demo代: rabbitmq全量单元测试
3.六种场景模型我们封装了统一的使用规范。

1.hello_world、work_queue、publish_subscribe 场景模型使用:

相关配置参见:config/config.yaml, rbbitmq 部分

1.1 启动一个消费者,通过回调函数在阻塞模式进行消息处理
consumer, err := HelloWorld.CreateConsumer()
	if err != nil {
		fmt.Printf("HelloWorld单元测试未通过。%s\n", err.Error())
		os.Exit(1)
	}

    // 连接关闭的回调,主要是记录错误,进行后续更进一步处理,不要尝试在这里编写重连逻辑
    // 本项目已经封装了完善的消费者端重连逻辑,触发这里的代码说明重连已经超过了最大重试次数
	consumer.OnConnectionError(func(err *amqp.Error) {
		log.Fatal(MyErrors.ErrorsRabbitMqReconnectFail + "\n" + err.Error())
	})

    // 进入阻塞状态,处理消息
	consumer.Received(func(received_data string) {
		fmt.Printf("HelloWorld回调函数处理消息:--->%s\n", received_data)
	})
1.2 调用生产者投递一个或者多个消息,投递通常都是一次性的。
    // 这里创建场景模型的时候通过不同的模型名称创建即可,主要有:hello_world、work_queue、publish_subscribe 
	hello_producer, _ := hello_world.CreateProducer()
	var res bool
	for i := 0; i < 10; i++ {
		str := fmt.Sprintf("%d_hello_world开始发送消息测试", (i + 1))
		res = hello_producer.Send(str)
		//time.Sleep(time.Second * 1)
	}

	hello_producer.Close() // 消息投递结束,必须关闭连接
    // 简单判断一下最后一次发送结果
	if res {
		fmt.Printf("消息发送OK")
	} else {
		fmt.Printf("消息发送 失败")
	}
	

2.routing、topics 场景模型使用:

routing模式属于路由键的严格匹配模式。
topics模式比routing模式更灵活,两者使用、功能几乎完全一致。该模式完全可以代替routing模式,因此这里仅介绍 topics模式。
注意:生产者设置键的规则必须是:关键词A.关键词B.关键词C等,即关键词之间必须使用.(点)隔开,消费者端只需要将.(点)左边或右边的关键词使用#代替即可。

2.1 启动多个消费者,处于阻塞模式进行消息接受、处理。
    // 启动第一个消费者,这里使用协程的目的主要是保证第一个启动后不阻塞,否则就会导致第二个消费者无法启动
    go func(){
        consumer, err := Topics.CreateConsumer()
    
        if err != nil {
            t.Errorf("Routing单元测试未通过。%s\n", err.Error())
            os.Exit(1)
        }

    // 连接关闭的回调,主要是记录错误,进行后续更进一步处理,不要尝试在这里编写重连逻辑
    // 本项目已经封装了完善的消费者端重连逻辑,触发这里的代码说明重连已经超过了最大重试次数
        consumer.OnConnectionError(func(err *amqp.Error) {
            log.Fatal(MyErrors.ErrorsRabbitMqReconnectFail + "\n" + err.Error())
        })

        // 通过route_key 模糊匹配队列路由键的消息来处理
        consumer.Received("#.even", func(received_data string) {
            fmt.Printf("模糊匹配偶数键:--->%s\n", received_data)
        })
    }()

        // 启动第二个消费者,这里没有使用协程,在消息处理环节程序就会阻塞等待,处理消息
        consumer, err := Topics.CreateConsumer()
    
        if err != nil {
            t.Errorf("Routing单元测试未通过。%s\n", err.Error())
            os.Exit(1)
        }
    
        consumer.OnConnectionError(func(err *amqp.Error) {
        // 连接关闭的回调,主要是记录错误,进行后续更进一步处理,不要尝试在这里编写重连逻辑
        // 本项目已经封装了完善的消费者端重连逻辑,触发这里的代码说明重连已经超过了最大重试次数
            log.Fatal(MyErrors.ErrorsRabbitMqReconnectFail + "\n" + err.Error())
        })

        // 通过route_key 模糊匹配队列路由键的消息来处理
        consumer.Received("#.odd", func(received_data string) {
    
            fmt.Printf("模糊匹配奇数键:--->%s\n", received_data)
        })
2.2 调用生产者投递一个或者多个消息

	producer, _ := Topics.CreateProducer()
	var res bool
	var key string
	for i := 1; i <= 10; i++ {

		//  将 偶数 和  奇数 分发到不同的key,消费者端,启动两个也各自处理偶数和奇数
		if i%2 == 0 {
			key = "key.even" //  偶数键
		} else {
			key = "key.odd" //  奇数键
		}
		str_data := fmt.Sprintf("%d_Routing_%s, 开始发送消息测试", i, key)
		res = producer.Send(key, str_data)
		//time.Sleep(time.Second * 1)
	}

	producer.Close() // 消息投递结束,必须关闭连接

    // 简单判断一下最后一次发送结果
	if res {
		fmt.Printf("消息发送OK")
	} else {
		fmt.Printf("消息发送 失败")
	}
	//Output: 消息发送OK

Comment ( 0 )

Sign in for post a comment