C# 对RabbitMQ使用

这篇具有很好参考价值的文章主要介绍了C# 对RabbitMQ使用。希望对大家有所帮助。如果存在错误或未考虑完全的地方,请大家不吝赐教,您也可以点击"举报违法"按钮提交疑问。

1、安装NuGet包RabbitMQ.Client

 2    生产者-确认机制   

(1). 含义:就是应答模式,生产者发送一条消息之后,Rabbitmq服务器做了个响应,表示收到了。

(2). 特点:异步模式,在响应之前,可以继续发送消息,单条消息、批量消息均可继续发送。

(3). 核心代码:单条消息确认: channel.waitForConfirms()

         批量消息确认: channel.waitForConfirmsOrDie()

大致流程:channel.ConfirmSelect();  开启确认模式→发送消息→提供一个回执方法WaitForConfirms();   返回一个bool 值

using RabbitMQ.Client;
using RabbitMQ.Client.Events;
using System.Text;

var factory = new ConnectionFactory();
factory.HostName = "127.0.0.1";//RabbitMQ服务器地址
factory.DispatchConsumersAsync = true;//支持异步发送消息
string exchangeName = "exchange1";//交换机的名字
string eventName = "myEvent";// routingKey的值
using var conn = factory.CreateConnection();
while (true)
{
    string msg = DateTime.Now.TimeOfDay.ToString();//待发送消息
    using (var channel = conn.CreateModel())//创建信道
    {
        try
        {
            var properties = channel.CreateBasicProperties();
            properties.DeliveryMode = 2;//1非持久化、2是持久化


            //交换机 Name:交换机名称。
            // Type: 交换机类型——direct、topic、fanout、headers、sharding
            //Durable:消息代理重启后,交换机是否还存在。
            //Auto-delete :当所有与之绑定的消息队列都完成了对此交换机的使用后,删掉它。
            //Arguments:依赖代理本身。
            channel.ExchangeDeclare(exchange: exchangeName, type: "direct", true, false);//声明交换机

            byte[] body = Encoding.UTF8.GetBytes(msg);


            channel.ConfirmSelect();//开启消息确认模式

            //发布消息    
            //exchange:交换机名称
            //mandatory为true时,表示如果消息没有被正确路由,消息将退回消息的生产者 如果设置为false,那么broker端自动删除该消息。
            //routingKey:路由键
            //props:消息属性字段,比如消息头部信息等等
            //body:消息主体部分
            channel.BasicPublish(exchange: exchangeName, routingKey: eventName,
            mandatory: true, basicProperties: properties, body: body);
            Console.WriteLine("发布了消息:" + msg);



            /*首先开启Confirm模式,通知消息生产者成功推送到RabbitMQ中*/
            if (channel.WaitForConfirms())  //单条消息确认
            {
                //表示消息发送成功(已经存入队列)
                Console.WriteLine($"【{msg}】成功发送到RabbitMQ!");
            }
            else
            {
                Console.WriteLine($"【{msg}】发送到RabbitMQ失败!");
            }
            //channel.WaitForConfirmsOrDie();//如果所有消息发送成功 就正常执行, 如果有消息发送失败;就抛出异常;

        }
        catch (Exception ex)
        {
            //表示消息发送失败
            Console.WriteLine($"【{msg}】发送到RabbitMQ失败!");
        }
    }

    Thread.Sleep(1000);
}

运行结果:

c#使用rabbitmq,C#,rabbitmq,c#

2. TX事务模式

(1). 含义:基于AMPQ协议;可以让信道设置成一个带事务的信道,分为三步:开启事务、提交事务、事务回滚

(2). 特点:同步模式,在事务提交之前不能继续发送消息,该模式相比Confirm模式效率差一点。

(3). 核心代码:channel.TxSelect();   开启一个事务

         channel.TxCommit();  提交事务, 这一步成功后,消息才真正的写入队列

       channel.TxRollback();   事务回滚

using RabbitMQ.Client;
using RabbitMQ.Client.Events;
using System.Text;

var factory = new ConnectionFactory();
factory.HostName = "127.0.0.1";
factory.DispatchConsumersAsync = true;
string exchangeName = "exchange1";
string eventName = "myEvent";
using var conn = factory.CreateConnection();
while (true)
{
    string msg = DateTime.Now.TimeOfDay.ToString();
    using (var channel = conn.CreateModel())
    {
        try
        {
            var properties = channel.CreateBasicProperties();
            properties.DeliveryMode = 2;
            channel.ExchangeDeclare(exchange: exchangeName, type: "direct", true, false);
            byte[] body = Encoding.UTF8.GetBytes(msg);
            channel.TxSelect(); //开启事务
            channel.BasicPublish(exchange: exchangeName, routingKey: eventName,
            mandatory: true, basicProperties: properties, body: body);
            Console.WriteLine($"【{msg}】成功发送到RabbitMQ!");
            channel.TxCommit(); //只有事务提交成功以后,才会真正的写入到队列里面去
        }
        catch (Exception ex)
        {
            //表示消息发送失败
            Console.WriteLine($"【{msg}】发送到RabbitMQ失败!");
            channel.TxRollback();
        }
    }

    Thread.Sleep(1000);
}

3、消费者(手动确认)

(1) 含义:消费者消费一条,回执给RabbitMq一条消息,Rabbitmq 只删除当前这一条消息;相当于是一条消费了,删除一条消息,性能稍微低一些;

(2) 特点:消费1条应答一次,可以告诉RabbitMq消费成功or失败,消费成功,服务器删除该条消息,消费失败,可以删除也可以重新写入。

(3) 核心代码:autoAck: false,表示不自动确认

 然后:channel.BasicAck(deliveryTag: ea.DeliveryTag, multiple: false);    表示消费成功

 channel.BasicReject(deliveryTag: ea.DeliveryTag, requeue: false);         表示消费失败, 可以配置:requeue: true:重新写入到队列里去; false: 删除消息

using RabbitMQ.Client;
using RabbitMQ.Client.Events;
using System.Text;

var factory = new ConnectionFactory();
factory.HostName = "127.0.0.1";//服务器地址
factory.DispatchConsumersAsync = true;//支持异步接受
string exchangeName = "exchange1";//交换机的名称
string eventName = "myEvent";//路由键
using var conn = factory.CreateConnection();
using var channel = conn.CreateModel();//创建信道
string queueName = "queue1";//队列名称

//声明了交换机
channel.ExchangeDeclare(exchange: exchangeName, type: "direct",true,false);

//声明一个队列
//queuename: 队列的名称
//durable 是否持久化。true消息会持久化到本地,保证重启服务后消息不会丢失
//exclusive :表示独占方式,设置为true 在某些情景下有必要,例如:顺序消费。表示只有一个channel可以去监听,其他channel都不能够监听。目的就是为了保证顺序消费。
//autoDelete:队列如果与Exchange未绑定,则自动删除
//arguments:扩展参数
channel.QueueDeclare(queue: queueName, durable: true,
        exclusive: false, autoDelete: false, arguments: null);

//绑定队列
channel.QueueBind(queue: queueName,
    exchange: exchangeName, routingKey: eventName);


//创建一个消费者
var consumer = new AsyncEventingBasicConsumer(channel);
consumer.Received += Consumer_Received;
//注册消费者订阅
//autoAck 是否自动确认消息, true自动确认,false 不自动要手动调用, 建立设置为false
channel.BasicConsume(queue: queueName, autoAck: false, consumer: consumer);
Console.ReadLine();


async Task Consumer_Received(object sender, BasicDeliverEventArgs args)
{
    try
    {
        var bytes = args.Body.ToArray();
        string msg = Encoding.UTF8.GetString(bytes);
        Console.WriteLine(DateTime.Now + "收到了消息" + msg);
         
        //DeliveryTag: 唯一的编号
        //multiple:是否批量确认.true:将一次性ack所有小于deliveryTag的消息。
        channel.BasicAck(args.DeliveryTag, multiple: false);手动确认 
        await Task.Delay(800);
    }
    catch (Exception ex)
    {
        //异常重试
        //DeliveryTag: 唯一的编号
        //requeue:如果 requeue 参数设置为 true,则 RabbitMQ 会重新将这条消息存入队列,以便发送给下一个订阅的消费者; 如果 requeue 参数设置为 false,则 RabbitMQ 立即会还把消息从队列中移除,而不会把它发送给新的消费者。
        channel.BasicReject(args.DeliveryTag, true);
        Console.WriteLine("处理收到的消息出错" + ex);
    }
}

运行结果

c#使用rabbitmq,C#,rabbitmq,c#

参考

【Windows安装RabbitMQ详细教程】_慕之寒的博客-CSDN博客_rabbitmq安装windows

第四节:RabbitMq剖析生产者、消费者的几种消息确认机制(Confirm、事务、自动、手动) - Yaopengfei - 博客园文章来源地址https://www.toymoban.com/news/detail-699530.html

到了这里,关于C# 对RabbitMQ使用的文章就介绍完了。如果您还想了解更多内容,请在右上角搜索TOY模板网以前的文章或继续浏览下面的相关文章,希望大家以后多多支持TOY模板网!

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处: 如若内容造成侵权/违法违规/事实不符,请点击违法举报进行投诉反馈,一经查实,立即删除!

领支付宝红包 赞助服务器费用

相关文章

  • C#调用RabbitMQ实现消息队列

    前言 前几天在做日志收集,用到了RabbitMQ,它作为一种中间件,需要对其进行下载,安装,和配置。 消息队列 什么是消息队列?,我们这样想一下,用户访问网站,最终是要将数据以HTTP的协议的方式,通过网络传输到主机的某个端口上的。 那么,接收数据的方式是什么呢?

    2024年02月05日
    浏览(36)
  • c# MES 对接之二(MQTT、Restful、RabbitMQ)

    //以下是C# MES对接的示例代码: //1. MQTT //使用M2Mqtt库实现C#与MQTT的通信,示例代码如下: //csharp using System; using System.Text; using uPLibrary.Networking.M2Mqtt; using uPLibrary.Networking.M2Mqtt.Messages; class Program {     static void Main(string[] args)     {         // 创建MQTT客户端实例         MqttCl

    2024年02月09日
    浏览(41)
  • Linux安装RabbitMQ,怎么使用RabbitMQ

    MQ(消息队列),典型的生产者消费者模式,生产者不断向消息队列发送消息,消费者不断从消息队列中获取消息。因为消息的生产和消费都是异步的,而且只关心消息的发送和接收,没有业务逻辑的侵入,轻松实现了系统之间的解耦合。通过高效的可靠的消息传递机制进行系统

    2024年02月12日
    浏览(39)
  • 【RabbitMQ】RabbitMQ控制台的使用

    如果在本机上装了RabbitMQ则在浏览器访问127.0.0.1:15672,如果在服务器装了RabbitMQ则通过在浏览器输入urlip:15762来访问 登录后进入主页   进入主页后选择Admin,对应找到添加用户选择输入信息后即可完成添加 还是在Admin界面,与添加用户类似   我们给刚创建的用户授权使用刚刚创

    2024年02月12日
    浏览(49)
  • rabbitmq笔记-rabbitmq客户端开发使用

    1.创建ConnectionFactory,给定参数ip地址,端口号,用户名和密码等 2.创建ConnectionFactory,使用uri方式实现,创建channel。 注意: Connection可以用来创建多个channel实例,但channel实例不能在线程间共享,应用程序为每个线程开辟一个channel。多线程间共享channel实例是非线程安全的。

    2024年02月11日
    浏览(46)
  • RabbitMQ实现延迟消息,RabbitMQ使用死信队列实现延迟消息,RabbitMQ延时队列插件

    假设有一个业务场景:超过30分钟未付款的订单自动关闭,这个功能应该怎么实现? RabbitMQ使用死信队列,可以实现消息的延迟接收。 队列有一个消息过期属性。就像丰巢超过24小时就收费一样,通过设置这个属性,超过了指定事件的消息将会被丢弃。 这个属性交:x-message

    2024年02月13日
    浏览(79)
  • 【RabbitMQ】RabbitMQ安装与使用详解以及Spring集成

    🎉🎉欢迎来到我的CSDN主页!🎉🎉 🏅我是Java方文山,一个在CSDN分享笔记的博主。📚📚 🌟推荐给大家我的专栏《RabbitMQ实战》。🎯🎯 👉点击这里,就可以查看我的主页啦!👇👇 Java方文山的个人主页 🎁如果感觉还不错的话请给我点赞吧!🎁🎁 💖期待你的加入,一

    2024年01月20日
    浏览(43)
  • RabbitMQ系列(5)--使用Java实现RabbitMQ的消费者接收消息

    前言:先简单了解RabbitMQ的工作过程,方便后续开发理清思路 简略: 详细: 1、新建消费者类 效果图: 2、编写消费者消费消息的代码 例: 3、查看代码运行结果 运行代码后如果有输出生产者发送的”Hello World”信息,则证明消费者消费消息成功 4、在web页面上查看队列的消

    2024年02月06日
    浏览(43)
  • 【RabbitMQ】RabbitMQ 消息的堆积问题 —— 使用惰性队列解决消息的堆积问题

    消息的堆积问题是指在消息队列系统中,当生产者以较快的速度发送消息,而消费者处理消息的速度较慢,导致消息在队列中积累并达到队列的存储上限。在这种情况下,最早被发送的消息可能会在队列中滞留较长时间,直到超过队列的容量上限。当队列已满且没有更多的可

    2024年02月05日
    浏览(50)
  • 【RabbitMQ】| Lion带你 (超详细) 从0到1使用SpringBoot操作RabbitMQ

    Spring Boot操作RabbitMQ是一种非常有用的技术,它可以提高应用程序的性能、可靠性和可伸缩性,为开发人员提供更好的开发和维护体验。下面是它的一些优点: 提高应用程序的可靠性和稳定性。RabbitMQ作为一种基于消息传递的中间件,可以将消息传递给应用程序,从而避免了高

    2024年02月01日
    浏览(35)

觉得文章有用就打赏一下文章作者

支付宝扫一扫打赏

博客赞助

微信扫一扫打赏

请作者喝杯咖啡吧~博客赞助

支付宝扫一扫领取红包,优惠每天领

二维码1

领取红包

二维码2

领红包