.netcore3.1 RabbitMq Routing的应用

• .netcore3.1 RabbitMq Routing的应用
  • 创建水果交换机，打折队列
    • 安装依赖包
    • 代码
  • 创建篮子（消费端）并进行消费
    • 命令创建消费篮子
    • 消费代码
    • 运行消费端
    • 添加消息
    • 测试运行

创建水果交换机，打折队列

安装依赖包

dotnet add package RabbitMQ.Client --version 6.2.1

代码

var factory = new ConnectionFactory()
{
    HostName = "127.0.0.1",
    UserName = "bob",
    Password = "bob"
};
// 创建一个链接
using (var connection = factory.CreateConnection())
{
    // 创建一个通道
    using (var channel = connection.CreateModel())
    {
        // 创建水果交换机
        channel.ExchangeDeclare("FruitExchange", ExchangeType.Direct,true,false);
        // 声明一个七折队列
        channel.QueueDeclare(
            queue: "SevenQueue",
            durable: true,
            exclusive: false,
            autoDelete: false,
            arguments: null);
        // 将七折队列与水果交换机绑定在一起
        // 添加 苹果，香蕉，橘子
        string[] fruits = new string[] { "apple", "banana", "orange" };
        for (var i = 0; i < fruits.Length; i++)
        {
            channel.QueueBind("SevenQueue", "FruitExchange", fruits[i], null);
        }
        // 声明一个原价队列
        channel.QueueDeclare(
            queue: "OriginalQueue",
            durable: true,
            exclusive: false,
            autoDelete: false,
            arguments: null);
        // 将原价队列与水果交换机绑定在一起
        // 添加 蓝莓
        string[] originalfruits = new string[] { "Blueberry" };
        for (var i = 0; i < originalfruits.Length; i++)
        {
            channel.QueueBind("OriginalQueue", "FruitExchange", originalfruits[i], null);
        }
        Console.WriteLine("Finish");
        Console.ReadLine();
    }
}

创建篮子（消费端）并进行消费

命令创建消费篮子

dotnet new console -n BasketConsumer
cd BasketConsumer
dotnet add package RabbitMQ.Client --version 6.2.1
dotnet restore

消费代码

var factory = new ConnectionFactory()
{
    HostName = "127.0.0.1",
    UserName = "bob",
    Password = "bob"
};
// 获取篮子名称
var _BasketName = args[0];
// 创建一个链接
using (var connection = factory.CreateConnection())
{
    // 创建一个通道
    using (var channel = connection.CreateModel())
    {
        // one by one 处理
        channel.BasicQos(prefetchSize: 0, prefetchCount: 1, global: false);
        // 创建消费实例
        var consumer = new EventingBasicConsumer(channel);
        // 事件在交付到使用者时触发。（消费处理事件）
        consumer.Received += (model, ea) =>
        {
            try
            {
                string BasketName = _BasketName;
                var body = ea.Body.ToArray();
                var message = Encoding.UTF8.GetString(body);
                Thread.Sleep(1000);
                if (BasketName.ToLower() == "original")
                {
                    Console.BackgroundColor = ConsoleColor.Blue; //设置背景色
                    Console.ForegroundColor = ConsoleColor.White; //设置前景色，即字体颜色
                    Console.WriteLine(" Original : {0}", message);
                    // 手动确认
                    channel.BasicAck(deliveryTag: ea.DeliveryTag, multiple: false);
                }
                else
                {
                    Console.BackgroundColor = ConsoleColor.Green;
                    Console.ForegroundColor = ConsoleColor.DarkGreen;
                    Console.WriteLine(" Seven : {0}", message);
                    // 手动确认
                    channel.BasicAck(deliveryTag: ea.DeliveryTag, multiple: false);
                }
            }
            catch (Exception ex)
            {
                Console.WriteLine("【Error】:", ex.Message);
            }
        };
        // 通过判断绑定到不同队列中去
        if (_BasketName.ToLower() == "original")
        {
            channel.BasicConsume(queue: "OriginalQueue", autoAck: false, consumer: consumer);
        }
        else
        {
            channel.BasicConsume(queue: "SevenQueue", autoAck: false, consumer: consumer);
        }
        Console.ReadLine();
    }
}

运行消费端

dotnet run seven
dotnet run original

添加消息

var factory = new ConnectionFactory()
{
    HostName = "127.0.0.1",
    UserName = "bob",
    Password = "bob"
};
// 创建一个链接
using (var connection = factory.CreateConnection())
{
    // 创建一个通道
    using (var channel = connection.CreateModel())
    {
        // 创建水果交换机
        channel.ExchangeDeclare("FruitExchange", ExchangeType.Direct,true,false);
        // 声明一个七折队列
        channel.QueueDeclare(
            queue: "SevenQueue",
            durable: true,
            exclusive: false,
            autoDelete: false,
            arguments: null);
        // 将七折队列与水果交换机绑定在一起
        // 添加 苹果，香蕉，橘子
        string[] fruits = new string[] { "apple", "banana", "orange" };
        for (var i = 0; i < fruits.Length; i++)
        {
            channel.QueueBind("SevenQueue", "FruitExchange", fruits[i], null);
        }
        // 持久化操作，告诉Rabbitmq服务器将消息存储在磁盘上，但仍然有
        // 少部分是存储在缓存中的，所以仍然不能绝对保证
        var properties = channel.CreateBasicProperties();
        properties.Persistent = true;
        // 发布一个苹果，香蕉，与橘子
        channel.BasicPublish(
            exchange: "FruitExchange",
            routingKey: "apple",
            basicProperties: properties,
            body: Encoding.UTF8.GetBytes("Apple")
            );
        channel.BasicPublish(
            exchange: "FruitExchange",
            routingKey: "banana",
            basicProperties: properties,
            body: Encoding.UTF8.GetBytes("Banana")
            );
        channel.BasicPublish(
            exchange: "FruitExchange",
            routingKey: "orange",
            basicProperties: properties,
            body: Encoding.UTF8.GetBytes("Orange")
        );
        // 声明一个原价队列
        channel.QueueDeclare(
            queue: "OriginalQueue",
            durable: true,
            exclusive: false,
            autoDelete: false,
            arguments: null);
        // 将原价队列与水果交换机绑定在一起
        // 添加 蓝莓
        string[] originalfruits = new string[] { "Blueberry" };
        for (var i = 0; i < originalfruits.Length; i++)
        {
            channel.QueueBind("OriginalQueue", "FruitExchange", originalfruits[i], null)
        }
        // 创建一个蓝莓水果到原价队列进行消费
        channel.BasicPublish(
            exchange: "FruitExchange",
            routingKey: "Blueberry",
            basicProperties: properties,
            body: Encoding.UTF8.GetBytes("Blueberry")
        );
        Console.WriteLine("Finish");
        Console.ReadLine();
    }
}

测试运行