.netcore3.1 RabbitMq Fanout交换机

• .netcore3.1 RabbitMq Fanout交换机
  • 创建相关队列(手机与邮箱)
    • 代码
    • 运行图片
  • 创建2种消费端(手机和邮箱)
    • 创建消费端
    • 运行消费端
    • 添加消息
    • 运行发布端

创建相关队列(手机与邮箱)

代码

var factory = new ConnectionFactory()
{
    HostName = "47.98.187.188",
    UserName = "bob",
    Password = "bob"
};
// 创建一个链接
using (var connection = factory.CreateConnection())
{
    // 创建一个通道
    using (var channel = connection.CreateModel())
    {
        // 创建交换机 普通类型
        channel.ExchangeDeclare("MyFanoutExchangeName", ExchangeType.Fanout,true,false);
        // 声明一个队列
        channel.QueueDeclare(
            queue: "MyFanoutPhone",
            durable: true,
            exclusive: false,
            autoDelete: false,
            arguments: null);
        // 声明一个队列
        channel.QueueDeclare(
            queue: "MyFanoutEmail",
            durable: true,
            exclusive: false,
            autoDelete: false,
            arguments: null);
        // 将队列与交换机绑定在一起
        channel.QueueBind("MyFanoutPhone", "MyFanoutExchangeName", string.Empty, null);
        channel.QueueBind("MyFanoutEmail", "MyFanoutExchangeName", string.Empty, null);
        Console.WriteLine("Finish");
        Console.ReadLine();
    }
}

运行图片

创建2种消费端(手机和邮箱)

创建消费端

var factory = new ConnectionFactory()
{
    HostName = "47.98.187.188",
    UserName = "bob",
    Password = "bob",
    VirtualHost = "/"
};
// 获取要取消息的队列的名称
var QueueName = args[0];
Console.WriteLine($"Queue Name: {QueueName}");
// 创建一个链接
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 _QueueName = QueueName;
                var body = ea.Body.ToArray();
                var message = Encoding.UTF8.GetString(body);
                Thread.Sleep(1000);
                if (QueueName == "MyFanoutPhone")
                {
                    Console.BackgroundColor = ConsoleColor.Blue; //设置背景色
                    Console.ForegroundColor = ConsoleColor.White; //设置前景色，即字体颜色
                    Console.WriteLine("From MyFanoutPhone Message: {0}", message);
                    // 否定确认,减少网络请求，提高性能
                    channel.BasicAck(ea.DeliveryTag,multiple: true);
                }
                else
                {
                    Console.BackgroundColor = ConsoleColor.Green;
                    Console.ForegroundColor = ConsoleColor.DarkGreen;
                    Console.WriteLine("From MyFanoutEmail Message: {0}", message);
                    // 重新排队，减少网络请求，提高性能
                    channel.BasicAck(ea.DeliveryTag, multiple: true);
                }
            }
            catch (Exception ex)
            {
                Console.WriteLine("【Error】:", ex.Message);
                channel.BasicNack(ea.DeliveryTag, multiple: true, requeue: true);
            }
        };
        // 绑定到队列中去
        channel.BasicConsume(queue: QueueName, autoAck: false, consumer: consumer);
        Console.ReadLine();
    }
}

运行消费端

./Worker.exe MyFanoutPhone
./Worker.exe MyFanoutEmail

添加消息

var factory = new ConnectionFactory()
{
    HostName = "47.98.187.188",
    UserName = "bob",
    Password = "bob"
};
// 创建一个链接
using (var connection = factory.CreateConnection())
{
    // 创建一个通道
    using (var channel = connection.CreateModel())
    {
        // 创建交换机 普通类型
        channel.ExchangeDeclare("MyFanoutExchangeName", ExchangeType.Fanout,true,false);
        // 声明一个队列
        channel.QueueDeclare(
            queue: "MyFanoutPhone",
            durable: true,
            exclusive: false,
            autoDelete: false,
            arguments: null);
        // 声明一个队列
        channel.QueueDeclare(
            queue: "MyFanoutEmail",
            durable: true,
            exclusive: false,
            autoDelete: false,
            arguments: null);
        // 将队列与交换机绑定在一起
        channel.QueueBind("MyFanoutPhone", "MyFanoutExchangeName", string.Empty, null);
        channel.QueueBind("MyFanoutEmail", "MyFanoutExchangeName", string.Empty, null);
        var properties = channel.CreateBasicProperties();
        properties.Persistent = true;
        channel.BasicPublish(
            exchange: "MyFanoutExchangeName",
            routingKey: string.Empty,
            basicProperties: properties,
            body: Encoding.UTF8.GetBytes("Happy New Year")
        );
        Console.WriteLine("Finish");
        Console.ReadLine();
    }
}

运行发布端