之前有写过.net core集成使用rabbitmq的博文,见.net core使用rabbitmq消息队列,但是里面的使用很简单,而且还有几个bug,想改下,但是后来想了想,还是算了,之前使用的是.net core 2.x,还是重新整理一遍吧!
由于代码比较多,我把代码传到gitee上了,地址见:https://gitee.com/shanfeng1000/dotnetcore-demo/tree/master/Rabbitmq
这是一个Demo项目,介绍.net core集成使用rabbitmq消息队列,使用的.net core 3.1,这里简单介绍:
生产者(AspNetCore.WebApi.Producer)
在Startup的ConfigureServices方法中添加相关rabbitmq的服务:
public void ConfigureServices(IServiceCollection services) { string[] hosts = new string[] { "192.168.209.133", "192.168.209.134", "192.168.209.135" }; int port = 5672; string userName = "admin"; string password = "123456"; string virtualHost = "/"; var arguments = new Dictionary<string, object>() { { "x-queue-type", "classic" } }; #region 日志记录 services.AddRabbitLogger(options => { options.Hosts = hosts; options.Password = password; options.Port = port; options.UserName = userName; options.VirtualHost = virtualHost; options.Arguments = arguments; options.Durable = true; options.AutoDelete = true; options.Category = "Home"; options.MinLevel = LogLevel.Warning; //队列模式 options.Queue = "queue.logger"; //交换机模式 //options.Exchange = "exchange.logger"; //options.RouteQueues = new RouteQueue[] { new RouteQueue() { Queue = "queue.logger", Route = "#" } }; //options.Type = RabbitExchangeType.Topic; }); #endregion #region 普通模式 services.AddRabbitProducer("SimplePattern", options => { options.Hosts = hosts; options.Password = password; options.Port = port; options.UserName = userName; options.VirtualHost = virtualHost; options.Arguments = arguments; options.Durable = true; options.AutoDelete = true; options.InitializeCount = 3; options.Queues = new string[] { "queue.simple" }; }); #endregion #region 工作模式 services.AddRabbitProducer("WorkerPattern", options => { options.Hosts = hosts; options.Password = password; options.Port = port; options.UserName = userName; options.VirtualHost = virtualHost; options.Arguments = arguments; options.Durable = true; options.AutoDelete = true; options.InitializeCount = 3; options.Queues = new string[] { "queue.worker" }; }); #endregion #region 发布订阅模式 services.AddRabbitProducer("FanoutPattern", options => { options.Hosts = hosts; options.Password = password; options.Port = port; options.UserName = userName; options.VirtualHost = virtualHost; options.Arguments = arguments; options.Durable = true; options.AutoDelete = true; options.InitializeCount = 3; options.RouteQueues = new RouteQueue[] { new RouteQueue() { Queue = "queue.fanout1" }, new RouteQueue() { Queue = "queue.fanout2" } }; options.Type = RabbitExchangeType.Fanout; options.Exchange = "exchange.fanout"; }); #endregion #region 路由模式 services.AddRabbitProducer("DirectPattern", options => { options.Hosts = hosts; options.Password = password; options.Port = port; options.UserName = userName; options.VirtualHost = virtualHost; options.Arguments = arguments; options.Durable = true; options.AutoDelete = true; options.InitializeCount = 5; options.Exchange = "exchange.direct"; options.Type = RabbitExchangeType.Direct; options.RouteQueues = new RouteQueue[] { new RouteQueue() { Queue = "queue.direct1", Route = "direct1" }, new RouteQueue() { Queue = "queue.direct2", Route = "direct2" } }; }); #endregion #region 主题模式 services.AddRabbitProducer("TopicPattern", options => { options.Hosts = hosts; options.Password = password; options.Port = port; options.UserName = userName; options.VirtualHost = virtualHost; options.Arguments = arguments; options.Durable = true; options.AutoDelete = true; options.InitializeCount = 5; options.RouteQueues = new RouteQueue[] { new RouteQueue() { Queue = "queue.topic1", Route = "topic1.#" }, new RouteQueue() { Queue = "queue.topic2", Route = "topic2.#" } }; options.Type = RabbitExchangeType.Topic; options.Exchange = "exchange.topic"; }); #endregion services.AddControllers(); }
里面介绍了6中集成方式:
使用AddRabbitLogger方法添加日志相关的服务,需要注意的是,数据是以json格式发送到rabbitmq中去的,具体可以参考RabbitLoggerMessage<T>类,最好是自己发布测试就可以了,当然读者可以安装自己的需求修改RabbitLogger类中的发布逻辑。
使用AddRabbitProducer方法添加一个发布者,可以指定名称,这个名称是获取发布者对象时使用。这个方法添加的发布者可以满足rabbitmq的五种使用方式(普通模式,工作模式,发布订阅模式,路由模式,主题模式),具体由RabbitProducerOptions配置指定。
服务配置好,具体使用可以参考HomeController,日志记录可以注入ILogger<T>对象,或者注入ILoggerFactory对象,然后获取ILogger<T>对象,直接使用ILogger<T>对象的方法就是发布消息了:
/// <summary> /// 日志 /// </summary> /// <returns></returns> [HttpGet] public string Get() { logger.LogDebug("Debug"); logger.LogInformation("Information"); logger.LogWarning("Warning"); logger.LogError("Error"); return "success"; }
至于另外五种模式,我们需要注入IRabbitProducerFactory对象,然后使用Create方法创建指定名称的发布者,然后调用Publish或者PublishAsync方法发布消息,而且他们都有几个重载。
需要注意的是,不同类型的生产者应该使用不同的Publish或者PublishAsync方法,比如普通模式和工作模式,因为他们没有路由参数,因此需要使用无路由参数的Publish方法,如:
/// <summary> /// Simple /// </summary> /// <returns></returns> [HttpGet("Simple")] public string Simple(string message = "Simple") { var producer = rabbitProducerFactory.Create("SimplePattern"); producer.Publish(message); return "success"; } /// <summary> /// Worker /// </summary> /// <param name="message"></param> /// <returns></returns> [HttpGet("Worker")] public string Worker(string message = "Worker") { var producer = rabbitProducerFactory.Create("WorkerPattern"); int count = 10; while (count-- > 0) { producer.Publish(message); } return "success"; }
而发布订阅模式、路由模式、主题模式都是有路由的(发布订阅模式的路由可以认为是空值),因此需要使用带路由参数的Publish方法:
/// <summary> /// Direct /// </summary> /// <param name="route"></param> /// <param name="message"></param> /// <returns></returns> [HttpGet("Direct")] public string Direct(string route = "direct1", string message = "Direct") { var producer = rabbitProducerFactory.Create("DirectPattern"); producer.Publish(route, message); return "success"; } /// <summary> /// Fanout /// </summary> /// <param name="message"></param> /// <returns></returns> [HttpGet("Fanout")] public string Fanout(string message = "Fanout") { var producer = rabbitProducerFactory.Create("FanoutPattern"); producer.Publish("", message);//fanout模式路由为空值 return "success"; } /// <summary> /// Topic /// </summary> /// <param name="route"></param> /// <param name="message"></param> /// <returns></returns> [HttpGet("Topic")] public string Topic(string route = "topic1.a", string message = "Topic") { var producer = rabbitProducerFactory.Create("TopicPattern"); producer.Publish(route, message); return "success"; }
消费者(AspNetCore.WebApi.Consumer)
生产者和消费者不在同一个项目中,同样的,需要先在Startup的ConfigureServices方法中添加服务:
public void ConfigureServices(IServiceCollection services) { string[] hosts = new string[] { "192.168.209.133", "192.168.209.134", "192.168.209.135" }; int port = 5672; string userName = "admin"; string password = "123456"; string virtualHost = "/"; var arguments = new Dictionary<string, object>() { { "x-queue-type", "classic" } }; #region 日志记录 services.AddRabbitConsumer(options => { options.Hosts = hosts; options.Password = password; options.Port = port; options.UserName = userName; options.VirtualHost = virtualHost; options.Arguments = arguments; options.Durable = true; options.AutoDelete = true; //options.AutoAck = true; //options.FetchCount = 10; //options.RouteQueues = new RouteQueue[] { new RouteQueue() { Queue = "queue.logger", Route = "#" } };//交换机模式 //options.Type = RabbitExchangeType.Topic;//交换机模式 }) //.AddListener("queue.logger", result => //{ // Console.WriteLine("Message From queue.logger:" + result.Body); //}); .AddListener<RabbitConsumerListener>("queue.logger"); //.AddListener("exchange.logger", "queue.logger", result => //{ // Console.WriteLine("Message From queue.logger:" + result.Body); //});//交换机模式 #endregion #region 普通模式 services.AddRabbitConsumer(options => { options.Hosts = hosts; options.Password = password; options.Port = port; options.UserName = userName; options.VirtualHost = virtualHost; options.Arguments = arguments; options.Durable = true; options.AutoDelete = true; //options.FetchCount = 1; options.AutoAck = false; }).AddListener("queue.simple", result => { Console.WriteLine("Message From queue.simple:" + result.Body); result.Commit(); }); #endregion #region 工作模式 services.AddRabbitConsumer(options => { options.Hosts = hosts; options.Password = password; options.Port = port; options.UserName = userName; options.VirtualHost = virtualHost; options.Arguments = arguments; options.Durable = true; options.AutoDelete = true; //options.FetchCount = 2; options.AutoAck = false; }).AddListener("queue.worker", result => { Console.WriteLine("Message From queue.worker1:" + result.Body); result.Commit(); }).AddListener("queue.worker", result => { Console.WriteLine("Message From queue.worker2:" + result.Body); result.Commit(); }); #endregion #region 发布订阅模式 services.AddRabbitConsumer(options => { options.Hosts = hosts; options.Password = password; options.Port = port; options.UserName = userName; options.VirtualHost = virtualHost; options.Arguments = arguments; options.Durable = true; options.AutoDelete = true; options.FetchCount = 2; options.AutoAck = false; options.RouteQueues = new RouteQueue[] { new RouteQueue() { Queue = "queue.fanout1" }, new RouteQueue() { Queue = "queue.fanout2" } }; options.Type = RabbitExchangeType.Fanout; }).AddListener("exchange.fanout", "queue.fanout1", result => { Console.WriteLine("Message From queue.fanout1:" + result.Body); result.Commit(); }).AddListener("exchange.fanout", "queue.fanout2", result => { Console.WriteLine("Message From queue.fanout2:" + result.Body); result.Commit(); }); #endregion #region 路由模式 services.AddRabbitConsumer(options => { options.Hosts = hosts; options.Password = password; options.Port = port; options.UserName = userName; options.VirtualHost = virtualHost; options.Arguments = arguments; options.Durable = true; options.AutoDelete = true; options.AutoAck = false; options.FetchCount = 2; options.Type = RabbitExchangeType.Direct; options.RouteQueues = new RouteQueue[] { new RouteQueue() { Queue = "queue.direct1", Route = "direct1" }, new RouteQueue() { Queue = "queue.direct2", Route = "direct2" } }; }).AddListener("exchange.direct", "queue.direct1", result => { Console.WriteLine("Message From queue.direct1:" + result.Body); result.Commit(); }).AddListener("exchange.direct", "queue.direct2", result => { Console.WriteLine("Message From queue.direct2:" + result.Body); result.Commit(); }); #endregion #region 主题模式 services.AddRabbitConsumer(options => { options.Hosts = hosts; options.Password = password; options.Port = port; options.UserName = userName; options.VirtualHost = virtualHost; options.Arguments = arguments; options.Durable = true; options.AutoDelete = true; options.FetchCount = 2; options.AutoAck = false; options.RouteQueues = new RouteQueue[] { new RouteQueue() { Queue = "queue.topic1", Route = "topic1.#" }, new RouteQueue() { Queue = "queue.topic2", Route = "topic2.#" } }; options.Type = RabbitExchangeType.Topic; }).AddListener("exchange.topic", "queue.topic1", result => { Console.WriteLine("Message From queue.topic1:" + result.Body); result.Commit(); }).AddListener("exchange.topic", "queue.topic2", result => { Console.WriteLine("Message From queue.topic2:" + result.Body); result.Commit(); }); #endregion services.AddControllers(); }
无论是日志的消费,还是其他五种模式的消费,都是先使用AddRabbitConsumer方法获取到一个IRabbitConsumerBuilder消费者构造对象,然后它的通过AddListener方法添加消息处理程序。
同样的,需要注意的是,普通模式和工作模式是不基于交换机的策略模式,因此需要使用不包含交换机参数的AddListener方法:
#region 普通模式 services.AddRabbitConsumer(options => { options.Hosts = hosts; options.Password = password; options.Port = port; options.UserName = userName; options.VirtualHost = virtualHost; options.Arguments = arguments; options.Durable = true; options.AutoDelete = true; //options.FetchCount = 1; options.AutoAck = false; }).AddListener("queue.simple", result => { Console.WriteLine("Message From queue.simple:" + result.Body); result.Commit(); }); #endregion #region 工作模式 services.AddRabbitConsumer(options => { options.Hosts = hosts; options.Password = password; options.Port = port; options.UserName = userName; options.VirtualHost = virtualHost; options.Arguments = arguments; options.Durable = true; options.AutoDelete = true; //options.FetchCount = 2; options.AutoAck = false; }).AddListener("queue.worker", result => { Console.WriteLine("Message From queue.worker1:" + result.Body); result.Commit(); }).AddListener("queue.worker", result => { Console.WriteLine("Message From queue.worker2:" + result.Body); result.Commit(); }); #endregion
而发布订阅模式、路由模式和主题模式都是基于交换机的策略模式,因此使用需要交换机参数的AddListener方法:
#region 发布订阅模式 services.AddRabbitConsumer(options => { options.Hosts = hosts; options.Password = password; options.Port = port; options.UserName = userName; options.VirtualHost = virtualHost; options.Arguments = arguments; options.Durable = true; options.AutoDelete = true; options.FetchCount = 2; options.AutoAck = false; options.RouteQueues = new RouteQueue[] { new RouteQueue() { Queue = "queue.fanout1" }, new RouteQueue() { Queue = "queue.fanout2" } }; options.Type = RabbitExchangeType.Fanout; }).AddListener("exchange.fanout", "queue.fanout1", result => { Console.WriteLine("Message From queue.fanout1:" + result.Body); result.Commit(); }).AddListener("exchange.fanout", "queue.fanout2", result => { Console.WriteLine("Message From queue.fanout2:" + result.Body); result.Commit(); }); #endregion #region 路由模式 services.AddRabbitConsumer(options => { options.Hosts = hosts; options.Password = password; options.Port = port; options.UserName = userName; options.VirtualHost = virtualHost; options.Arguments = arguments; options.Durable = true; options.AutoDelete = true; options.AutoAck = false; options.FetchCount = 2; options.Type = RabbitExchangeType.Direct; options.RouteQueues = new RouteQueue[] { new RouteQueue() { Queue = "queue.direct1", Route = "direct1" }, new RouteQueue() { Queue = "queue.direct2", Route = "direct2" } }; }).AddListener("exchange.direct", "queue.direct1", result => { Console.WriteLine("Message From queue.direct1:" + result.Body); result.Commit(); }).AddListener("exchange.direct", "queue.direct2", result => { Console.WriteLine("Message From queue.direct2:" + result.Body); result.Commit(); }); #endregion #region 主题模式 services.AddRabbitConsumer(options => { options.Hosts = hosts; options.Password = password; options.Port = port; options.UserName = userName; options.VirtualHost = virtualHost; options.Arguments = arguments; options.Durable = true; options.AutoDelete = true; options.FetchCount = 2; options.AutoAck = false; options.RouteQueues = new RouteQueue[] { new RouteQueue() { Queue = "queue.topic1", Route = "topic1.#" }, new RouteQueue() { Queue = "queue.topic2", Route = "topic2.#" } }; options.Type = RabbitExchangeType.Topic; }).AddListener("exchange.topic", "queue.topic1", result => { Console.WriteLine("Message From queue.topic1:" + result.Body); result.Commit(); }).AddListener("exchange.topic", "queue.topic2", result => { Console.WriteLine("Message From queue.topic2:" + result.Body); result.Commit(); }); #endregion
另外,AddListener中的消息处理委托可以使用一个实现了IRabbitConsumerListener接口的类代替,如Demo中的RabbitConsumerListener:
public class RabbitConsumerListener : IRabbitConsumerListener { public Task ConsumeAsync(RecieveResult recieveResult) { Console.WriteLine("RabbitConsumerListener:" + recieveResult.Body); recieveResult.Commit(); return Task.CompletedTask; } }