6赞

C#使用RabbitMq队列(Sample,Work,Fanout,Direct等模式的简单使用)

作者：夏晶阳--艺术 | 2021-10-22 21:57

这篇文章主要介绍了C#使用RabbitMq队列(Sample,Work,Fanout,Direct等模式的简单使用),本文给大家介绍的非常详细，对大家的学习或工作具有一定的参考借鉴价值，需要的朋友可以参考下

1：RabbitMQ是个啥？（专业术语参考自网络）

　RabbitMQ是实现了高级消息队列协议（AMQP）的开源消息代理软件（亦称面向消息的中间件）。

　　RabbitMQ服务器是用Erlang语言编写的，Erlang是专门为高并发而生的语言，而集群和故障转移是构建在开发电信平台框架上的。所有主要的编程语言均有与代理接口通讯的客户端库

2：使用RabbitMQ有啥好处？

RabbitMQ是使用Erlang语言开发的开源消息队列系统，基于AMQP协议来实现。
AMQP的主要特征是面向消息、队列、路由（包括点对点和发布/订阅）、可靠性、安全。
AMQP协议更多用在企业系统内，对数据一致性、稳定性和可靠性要求很高的场景，对性能和吞吐量的要求还在其次。
RabbitMQ的可靠性是非常好的，数据能够保证百分之百的不丢失。可以使用镜像队列，它的稳定性非常好。所以说在我们互联网的金融行业。

对数据的稳定性和可靠性要求都非常高的情况下，我们都会选择RabbitMQ。当然没有kafka性能好，但是要比AvtiveMQ性能要好很多。也可以自己做一些性能的优化。

RabbitMQ可以构建异地双活架构，包括每一个节点存储方式可以采用磁盘或者内存的方式，

3:RabbitMq的安装以及环境搭建等：

网络上有很多关于怎么搭建配置RabbitMq服务环境的详细文章，也比较简单，这里不再说明，本人是Docker上面的pull RabbitMq 镜像来安装的！

3.1：运行容器的命令如下：

docker run -d --hostname Log --restart=always --name rabbitmq -p 5672:5672 -p 15672:15672 -e RABBITMQ_DEFAULT_USER=log_user -e RABBITMQ_DEFAULT_PASS=331QQFEG123 rabbitmq:3-management

4：RabbitMq的使用场景主要有哪些，啥时候用或者不用？

4.1什么时候使用MQ？

1）数据驱动的任务依赖

2）上游不关心多下游执行结果

3）异步返回执行时间长

4.2什么时候不使用MQ？

需要实时关注执行结果（eg：同步调用）

5：具体C#怎么使用RabbitMq?下面直接上code和测试截图了（Demo环境是.NetCore3.1控制台+Docker上的RabbitMQ容器来进行的）

6：sample模式，就是简单地队列模式，一进一出的效果差不多，测试截图：

Code:

//简单生产端 ui调用者

using System;
namespace RabbitMqPublishDemo
{
  using MyRabbitMqService;
  using System.Runtime.CompilerServices;

  class Program
  {
    static void Main(string[] args)
    {
        //就是简单的队列，生产者
        Console.WriteLine("====RabbitMqPublishDemo====");
        for (int i = 0; i < 500; i++)
        {
          ZrfRabbitMqHelper.PublishSampleMsg("smapleMsg", $"nihaifengge:{i}");
        }
        Console.WriteLine("生成完毕！");
        Console.ReadLine();
    }
  }
}

/// 
/// 简单生产者 逻辑
/// 
/// 
/// 
public static void PublishSampleMsg(string queueName, string msg)
{

  using (IConnection conn = connectionFactory.CreateConnection())
  {
    using (IModel channel = conn.CreateModel())
    {
      channel.QueueDeclare(queue: queueName, durable: false, exclusive: false, autoDelete: false, arguments: null);
      var msgBody = Encoding.UTF8.GetBytes(msg);
      channel.BasicPublish(exchange: "", routingKey: queueName, basicProperties: null, body: msgBody);
    }
  }
}


//简单消费端
using System;

namespace RabbitMqConsumerDemo
{
  using MyRabbitMqService;
  using System.Runtime.InteropServices;

  class Program
  {
    static void Main(string[] args)
    {
      Console.WriteLine("====RabbitMqConsumerDemo====");
      ZrfRabbitMqHelper.ConsumeSampleMsg("smapleMsg", isBasicNack: true, handleMsgStr: handleMsgStr =>
      {
        Console.WriteLine($"订阅到消息:{DateTime.Now}:{handleMsgStr}");
      });
      Console.ReadLine();
    }
  }
}

   #region 简单生产者后端逻辑
    /// 
    /// 简单消费者
    /// 
    /// 队列名称
    /// 失败后是否自动放到队列
    /// 有就自己对字符串的处理,如果要存储到数据库请自行扩展
    public static void ConsumeSampleMsg(string queueName, bool isBasicNack = false, Action handleMsgStr = null)// bool ifBasicReject = false,
    {
      Console.WriteLine("ConsumeSampleMsg Waiting for messages....");
      IConnection conn = connectionFactory.CreateConnection();
      IModel channel = conn.CreateModel();
      channel.QueueDeclare(queue: queueName, durable: false, exclusive: false, autoDelete: false, arguments: null);
      var consumer = new EventingBasicConsumer(channel);
      consumer.Received += (sender, ea) =>
      {
        byte[] bymsg = ea.Body.ToArray();
        string msg = Encoding.UTF8.GetString(bymsg);
        if (handleMsgStr != null)
        {
          handleMsgStr.Invoke(msg);
        }
        else
        {
          Console.WriteLine($"{DateTime.Now}->收到消息:{msg}");
        }
      };
      channel.BasicConsume(queueName, autoAck: true, consumer);
    }
    #endregion

7：Work模式

//简单生产端 ui调用者

using System;
namespace RabbitMqPublishDemo
{
  using MyRabbitMqService;
  using System.Runtime.CompilerServices;

  class Program
  {
    static void Main(string[] args)
    {
        //就是简单的队列，生产者
        Console.WriteLine("====RabbitMqPublishDemo====");
        for (int i = 0; i < 500; i++)
        {
          ZrfRabbitMqHelper.PublishSampleMsg("smapleMsg", $"nihaifengge:{i}");
        }
        Console.WriteLine("生成完毕！");
        Console.ReadLine();
    }
  }
}

/// 
/// 简单生产者 逻辑
/// 
/// 
/// 
public static void PublishSampleMsg(string queueName, string msg)
{

  using (IConnection conn = connectionFactory.CreateConnection())
  {
    using (IModel channel = conn.CreateModel())
    {
      channel.QueueDeclare(queue: queueName, durable: false, exclusive: false, autoDelete: false, arguments: null);
      var msgBody = Encoding.UTF8.GetBytes(msg);
      channel.BasicPublish(exchange: "", routingKey: queueName, basicProperties: null, body: msgBody);
    }
  }
}


//简单消费端
using System;

namespace RabbitMqConsumerDemo
{
  using MyRabbitMqService;
  using System.Runtime.InteropServices;

  class Program
  {
    static void Main(string[] args)
    {
      Console.WriteLine("====RabbitMqConsumerDemo====");
      ZrfRabbitMqHelper.ConsumeSampleMsg("smapleMsg", isBasicNack: true, handleMsgStr: handleMsgStr =>
      {
        Console.WriteLine($"订阅到消息:{DateTime.Now}:{handleMsgStr}");
      });
      Console.ReadLine();
    }
  }
}

   #region 简单生产者后端逻辑
    /// 
    /// 简单消费者
    /// 
    /// 队列名称
    /// 失败后是否自动放到队列
    /// 有就自己对字符串的处理,如果要存储到数据库请自行扩展
    public static void ConsumeSampleMsg(string queueName, bool isBasicNack = false, Action handleMsgStr = null)// bool ifBasicReject = false,
    {
      Console.WriteLine("ConsumeSampleMsg Waiting for messages....");
      IConnection conn = connectionFactory.CreateConnection();
      IModel channel = conn.CreateModel();
      channel.QueueDeclare(queue: queueName, durable: false, exclusive: false, autoDelete: false, arguments: null);
      var consumer = new EventingBasicConsumer(channel);
      consumer.Received += (sender, ea) =>
      {
        byte[] bymsg = ea.Body.ToArray();
        string msg = Encoding.UTF8.GetString(bymsg);
        if (handleMsgStr != null)
        {
          handleMsgStr.Invoke(msg);
        }
        else
        {
          Console.WriteLine($"{DateTime.Now}->收到消息:{msg}");
        }
      };
      channel.BasicConsume(queueName, autoAck: true, consumer);
    }
    #endregion

8：Fanout

Code:

//就如下的code, 多次生产，3个消费者都可以自动开始消费

//生产者
using System;
namespace RabbitMqPublishDemo
{
  using MyRabbitMqService;
  using System.Runtime.CompilerServices;
  class Program
  {
    static void Main(string[] args)
    {
      for (int i = 0; i < 500; i++)
      {
        ZrfRabbitMqHelper.PublishWorkQueueModel("workqueue", $" :发布消息成功{i}");
      }
      Console.WriteLine("工作队列模式 生成完毕......！");
      Console.ReadLine();
    }
  }
}

//生产者后端逻辑
public static void PublishWorkQueueModel(string queueName, string msg)
    {
      using (var connection = connectionFactory.CreateConnection())
      using (var channel = connection.CreateModel())
      {
        channel.QueueDeclare(queue: queueName, durable: true, exclusive: false, autoDelete: false, arguments: null);
        var body = Encoding.UTF8.GetBytes(msg);
        var properties = channel.CreateBasicProperties();
        properties.Persistent = true;

        channel.BasicPublish(exchange: "", routingKey: queueName, basicProperties: properties, body: body);
        Console.WriteLine($"{DateTime.Now},SentMsg: {msg}");
      }
    }

//work消费端
using System;

namespace RabbitMqConsumerDemo
{
  using MyRabbitMqService;
  using System.Runtime.InteropServices;
  class Program
  {
    static void Main(string[] args)
    {
      Console.WriteLine("====Work模式开启了====");
      ZrfRabbitMqHelper.ConsumeWorkQueueModel("workqueue", handserMsg: msg =>
      {
        Console.WriteLine($"work模式获取到消息{msg}");
      });
      Console.ReadLine();
    }
  }
}

//work后端逻辑
    public static void ConsumeWorkQueueModel(string queueName, int sleepHmao = 90, Action handserMsg = null)
    {
      var connection = connectionFactory.CreateConnection();
      var channel = connection.CreateModel();

      channel.QueueDeclare(queue: queueName, durable: true, exclusive: false, autoDelete: false, arguments: null);
      channel.BasicQos(prefetchSize: 0, prefetchCount: 1, global: false);

      var consumer = new EventingBasicConsumer(channel);
      Console.WriteLine(" ConsumeWorkQueueModel Waiting for messages....");

      consumer.Received += (sender, ea) =>
      {
        var body = ea.Body.ToArray();
        var message = Encoding.UTF8.GetString(body);
        if (handserMsg != null)
        {
          if (!string.IsNullOrEmpty(message))
          {
            handserMsg.Invoke(message);
          }
        }
        channel.BasicAck(deliveryTag: ea.DeliveryTag, multiple: false);
      };
      channel.BasicConsume(queue: queueName, autoAck: false, consumer: consumer);
    }

9：Direct

Code:

//同一个消息会被多个订阅者消费

//发布者
using System;

namespace RabbitMqPublishDemo
{
  using MyRabbitMqService;
  using System.Runtime.CompilerServices;

  class Program
  {
    static void Main(string[] args)
    {

      #region 发布订阅模式，带上了exchange
      for (int i = 0; i < 500; i++)
      {
        ZrfRabbitMqHelper.PublishExchangeModel("exchangemodel", $"发布的消息是：{i}");
      }
      Console.WriteLine("发布ok！");
      #endregion
      Console.ReadLine();
    }
  }
}
//发布者的后端逻辑 我在这里选择了扇形: ExchangeType.Fanout
  public static void PublishExchangeModel(string exchangeName, string message)
    {
      using (var connection = connectionFactory.CreateConnection())
      using (var channel = connection.CreateModel())
      {
        channel.ExchangeDeclare(exchange: exchangeName, type: ExchangeType.Fanout);
        var body = Encoding.UTF8.GetBytes(message);
        channel.BasicPublish(exchange: exchangeName, routingKey: "", basicProperties: null, body: body);
        Console.WriteLine($" Sent {message}");
      }
    }


//订阅者
using System;
namespace RabbitMqConsumerDemo
{
  using MyRabbitMqService;
  using System.Runtime.InteropServices;
  class Program
  {
    static void Main(string[] args)
    {

      #region 发布订阅模式 Exchange
      ZrfRabbitMqHelper.SubscriberExchangeModel("exchangemodel", msg =>
      {
        Console.WriteLine($"订阅到消息：{msg}");
      });
      #endregion
      Console.ReadLine();
    }
  }
}

//订阅者后端的逻辑
 public static void SubscriberExchangeModel(string exchangeName, Action handlerMsg = null)
    {
      var connection = connectionFactory.CreateConnection();
      var channel = connection.CreateModel();

      channel.ExchangeDeclare(exchange: exchangeName, type: ExchangeType.Fanout);//Fanout 扇形分叉

      var queueName = channel.QueueDeclare().QueueName;
      channel.QueueBind(queue: queueName,
               exchange: exchangeName,
               routingKey: "");

      Console.WriteLine(" Waiting for msg....");

      var consumer = new EventingBasicConsumer(channel);
      consumer.Received += (model, ea) =>
      {
        var body = ea.Body.ToArray();
        var message = Encoding.UTF8.GetString(body);
        if (handlerMsg != null)
        {
          if (!string.IsNullOrEmpty(message))
          {
            handlerMsg.Invoke(message);
          }
        }
        else
        {
          Console.WriteLine($"订阅到消息:{message}");
        }
      };
      channel.BasicConsume(queue: queueName, autoAck: true, consumer: consumer);
    }

到此这篇关于C#使用RabbitMq队列(Sample,Work,Fanout,Direct等模式的简单使用)的文章就介绍到这了,更多相关C#使用RabbitMq队列内容请搜索以前的文章或继续浏览下面的相关文章希望大家以后多多支持！

推荐阅读

程序员
ng-click功能影响所有ng-repeat元素

如何解决《ng-click功能影响所有ng-repeat元素》经验，为你挑选了0个好方法。 ... [详细]
程序员
离子条件文本颜色

如何解决《离子条件文本颜色》经验，为你挑选了1个好方法。 ... [详细]
程序员
c ++迭代器卡在第一个值

如何解决《c++迭代器卡在第一个值》经验，为你挑选了1个好方法。 ... [详细]
程序员
在ASP.NET Core中限制到控制器命名空间的路由

如何解决《在ASP.NETCore中限制到控制器命名空间的路由》经验，为你挑选了1个好方法。 ... [详细]
程序员
展开segue - 当多个视图调用它时返回调用视图

如何解决《展开segue-当多个视图调用它时返回调用视图》经验，为你挑选了1个好方法。 ... [详细]
程序员
如何在SumoSelect下拉列表中获取选定的值？

如何解决《如何在SumoSelect下拉列表中获取选定的值？》经验，为你挑选了1个好方法。 ... [详细]
程序员
使用未对齐缓冲区进行矢量化:使用VMASKMOVPS:根据未对齐计数生成掩码？或者根本不使用那个insn

如何解决《使用未对齐缓冲区进行矢量化:使用VMASKMOVPS:根据未对齐计数生成掩码？或者根本不使用那个insn》经验，为你挑选了1个好方法。 ... [详细]
程序员
AX 2012R2:查找查询花费的时间太长,查找永远不会打开

如何解决《AX2012R2:查找查询花费的时间太长,查找永远不会打开》经验，为你挑选了1个好方法。 ... [详细]
程序员
(Android)我如何获得SmartPhone ScreenWidth和高度？

如何解决《(Android)我如何获得SmartPhoneScreenWidth和高度？》经验，为你挑选了0个好方法。 ... [详细]
程序员
alias_method:堆栈级别太深

如何解决《alias_method:堆栈级别太深》经验，为你挑选了1个好方法。 ... [详细]
程序员
Chomp地图的所有键(使用更少的代码)

如何解决《Chomp地图的所有键(使用更少的代码)》经验，为你挑选了1个好方法。 ... [详细]
程序员
两个类有共同的方法和属性

如何解决《两个类有共同的方法和属性》经验，为你挑选了1个好方法。 ... [详细]
程序员
SQL查询缓存

如何解决《SQL查询缓存》经验，为你挑选了0个好方法。 ... [详细]
程序员
H.264编码的NAL /切片分割

如何解决《H.264编码的NAL/切片分割》经验，为你挑选了0个好方法。 ... [详细]
程序员
将recyclerview焦点转移到列表中的最后一个元素

如何解决《将recyclerview焦点转移到列表中的最后一个元素》经验，为你挑选了3个好方法。 ... [详细]
程序员
Swagger PHP:如何声明属性以使用模式定义？

如何解决《SwaggerPHP:如何声明属性以使用模式定义？》经验，为你挑选了1个好方法。 ... [详细]
程序员
从.json url中提取数据导致Jquery出错

如何解决《从.jsonurl中提取数据导致Jquery出错》经验，为你挑选了1个好方法。 ... [详细]
程序员
如何使用jQuery选择data-title中的元素？

如何解决《如何使用jQuery选择data-title中的元素？》经验，为你挑选了0个好方法。 ... [详细]
程序员
SQL Server表列中的默认值为GUID

如何解决《SQLServer表列中的默认值为GUID》经验，为你挑选了1个好方法。 ... [详细]
程序员
覆盖iPhone中所有设备的屏幕所需的图像大小

如何解决《覆盖iPhone中所有设备的屏幕所需的图像大小》经验，为你挑选了0个好方法。 ... [详细]

夏晶阳--艺术

这个屌丝很懒，什么也没留下！

关注作者

Tags | 热门标签

RankList | 热门文章