.NET 云原生架构师训练营(模块二 基础巩固 RabbitMQ 工作队列和交换机)--学习笔记

简介: WorkQueuePublish/SubscribeRoutingEmitLog

2.6.4 RabbitMQ -- 工作队列和交换机

  • WorkQueue
  • Publish/Subscribe
  • Routing
  • EmitLog

WorkQueue

WorkQueue:https://www.rabbitmq.com/tutorials/tutorial-two-dotnet.html

  • 一个消息生产者,多个消息消费者
  • exchange 交换机自动恢复
  • 对消息进行持久化
  • 手动确认消息

对消息进行持久化

var properties = channel.CreateBasicProperties();
properties.Persistent = true;
channel.BasicPublish(exchange: "",
                     routingKey: "task_queue",
                     basicProperties: properties,
                     body: body);

手动确认消息

autoAck: false

channel.BasicConsume(queue: "task_queue", autoAck: false, consumer: consumer);

手动调用 BasicAck

channel.BasicAck(deliveryTag: ea.DeliveryTag, multiple: false);

修改接收端为手动确认消息

autoAck: false

channel.BasicConsume(queue: "hello",
    autoAck: false,
    consumer: consumer);

BasicAck

consumer.Received += (model, ea) =>
{
    var body = ea.Body.ToArray();
    var message = Encoding.UTF8.GetString(body);
    Thread.Sleep(2000);// 演示多个接收端
    channel.BasicAck(ea.DeliveryTag, false);
    Console.WriteLine(" [x] Received {0}", message);
};

启动多个接收端

010.jpg

Publish/Subscribe

Publish/Subscribe:https://www.rabbitmq.com/tutorials/tutorial-three-dotnet.html

Fanout 交换机,每个队列都会收到

channel.ExchangeDeclare("logs", ExchangeType.Fanout);

Routing

Routing:https://www.rabbitmq.com/tutorials/tutorial-four-dotnet.html

Bindings

channel.QueueBind(queue: queueName,
                  exchange: "logs",
                  routingKey: "");

Direct exchange

channel.ExchangeDeclare(exchange: "direct_logs", type: "direct");

EmitLog

新建控制台项目 EmitLogDirect,ReceiveLogsDirect

发送端

namespace EmitLogDirect
{
    class EmitLogDirect
    {
        public static void Main(string[] args)
        {
            var factory = new ConnectionFactory() { HostName = "localhost" };
            using(var connection = factory.CreateConnection())
            using(var channel = connection.CreateModel())
            {
                channel.ExchangeDeclare(exchange: "direct_logs", type: ExchangeType.Direct);// 声明交换机
                var severity = (args.Length > 0) ? args[0] : "info";
                var message = (args.Length > 1)
                    ? string.Join(" ", args.Skip( 1 ).ToArray())
                    : "Hello World!";
                var body = Encoding.UTF8.GetBytes(message);
                channel.BasicPublish(exchange: "direct_logs",
                    routingKey: severity,// 路由 Key 自动带上严重级别
                    basicProperties: null,
                    body: body);
                Console.WriteLine(" [x] Sent '{0}':'{1}'", severity, message);
            }
            Console.WriteLine(" Press [enter] to exit.");
            Console.ReadLine();
        }
    }
}

011.jpg

error 级别单独发送到一个队列

接收端

namespace ReceiveLogsDirect
{
    class ReceiveLogsDirect
    {
        public static void Main(string[] args)
        {
            var factory = new ConnectionFactory() { HostName = "localhost" };
            using (var connection = factory.CreateConnection())
            using (var channel = connection.CreateModel())
            {
                channel.ExchangeDeclare(exchange: "direct_logs", type: ExchangeType.Direct);// 声明交换机
                var queueName = channel.QueueDeclare().QueueName;
                if (args.Length < 1)
                {
                    Console.Error.WriteLine("Usage: {0} [info] [warning] [error]",
                        Environment.GetCommandLineArgs()[0]);
                    Console.WriteLine(" Press [enter] to exit.");
                    Console.ReadLine();
                    Environment.ExitCode = 1;
                    return;
                }
                foreach (var severity in args)
                {
                    channel.QueueBind(queue: queueName,
                        exchange: "direct_logs",
                        routingKey: severity);// 路由 Key 自动带上严重级别
                }
                Console.WriteLine(" [*] Waiting for messages.");
                var consumer = new EventingBasicConsumer(channel);
                consumer.Received += (model, ea) =>
                {
                    var body = ea.Body.ToArray();
                    var message = Encoding.UTF8.GetString(body);
                    var routingKey = ea.RoutingKey;
                    Console.WriteLine(" [x] Received '{0}':'{1}'",
                        routingKey, message);
                };
                channel.BasicConsume(queue: queueName,
                    autoAck: true,
                    consumer: consumer);
                Console.WriteLine(" Press [enter] to exit.");
                Console.ReadLine();
            }
        }
    }
}

替换发送端,接收端的 localhost 为服务器地址

接收端控制台启动

dotnet run info waring error

发送端控制台启动

dotnet run info
dotnet run error
dotnet run waring test

接收端输出

[x] Received 'info':'Hello World!'
 [x] Received 'error':'Hello World!'
 [x] Received 'waring':'test'

GitHub源码链接:

https://github.com/MINGSON666/Personal-Learning-Library/tree/main/ArchitectTrainingCamp

相关实践学习
消息队列RocketMQ版:基础消息收发功能体验
本实验场景介绍消息队列RocketMQ版的基础消息收发功能,涵盖实例创建、Topic、Group资源创建以及消息收发体验等基础功能模块。
消息队列 MNS 入门课程
1、消息队列MNS简介 本节课介绍消息队列的MNS的基础概念 2、消息队列MNS特性 本节课介绍消息队列的MNS的主要特性 3、MNS的最佳实践及场景应用 本节课介绍消息队列的MNS的最佳实践及场景应用案例 4、手把手系列:消息队列MNS实操讲 本节课介绍消息队列的MNS的实际操作演示 5、动手实验:基于MNS,0基础轻松构建 Web Client 本节课带您一起基于MNS,0基础轻松构建 Web Client
目录
相关文章
|
2月前
|
消息中间件 Java 应用服务中间件
详解rocketMq通信模块&升级构想(下)
详解rocketMq通信模块&升级构想(下)
409 0
详解rocketMq通信模块&升级构想(下)
|
2月前
|
存储 开发框架 前端开发
前端框架EXT.NET Dotnet 3.5开发的实验室信息管理系统(LIMS)成品源码 B/S架构
发展历史:实验室信息管理系统(LIMS),就是指通过计算机网络技术对实验的各种信息进行管理的计算机软、硬件系统。也就是将计算机网络技术与现代的管理思想有机结合,利用数据处理技术、海量数据存储技术、宽带传输网络技术、自动化仪器分析技术,来对实验室的信息管理和质量控制等进行全方位管理的计算机软、硬件系统,以满足实验室管理上的各种目标(计划、控制、执行)。
41 1
|
2月前
|
消息中间件 Java 中间件
详解rocketMq通信模块&升级构想(上)
详解rocketMq通信模块&升级构想(上)
174 0
|
7天前
|
监控
交易平台---架构设计第一步拆分模块,拆分为7个模块
交易平台---架构设计第一步拆分模块,拆分为7个模块
|
8月前
|
消息中间件 编解码 网络协议
聊聊 RocketMQ 网络通讯模块
RocketMQ 的网络通讯模块负责生产者、消费者与 Broker 之间的网络通信。 笔者学习 RocketMQ 也是从通讯模块源码开始的,并且从源码里汲取了很多营养。
37051 3
聊聊 RocketMQ 网络通讯模块
|
8月前
|
消息中间件 Java Maven
消息中间件系列教程(08) -RabbitMQ -案例代码(工作队列模式)
消息中间件系列教程(08) -RabbitMQ -案例代码(工作队列模式)
42 0
|
2月前
|
运维 Linux Apache
LAMP架构调优(三)——模块的安装与调用
LAMP架构调优(三)——模块的安装与调用
16 0
|
2月前
|
数据安全/隐私保护 Windows
.net三层架构开发步骤
.net三层架构开发步骤
|
2月前
|
消息中间件 关系型数据库 MySQL
使用Nginx的stream模块实现MySQL反向代理与RabbitMQ负载均衡
使用Nginx的stream模块实现MySQL反向代理与RabbitMQ负载均衡
203 0
|
2月前
|
存储 消息中间件 对象存储
RocketMQ 中冷热分离的随机索引模块详解
本文主要介绍了RocketMQ 中冷热分离的随机索引特点、具体内容、与其他系统对比等内容。
102626 1