RabbitMQ在.net core中的应用

飞不高  金牌会员 | 2024-9-24 15:09:27 | 显示全部楼层 | 阅读模式
打印 上一主题 下一主题
楼主

主题 868|帖子 868|积分 2604

RabbitMQ 是一个可靠且成熟的消息传递和流署理，它很容易摆设在云环境、内部摆设和本地呆板上。它目前被全天下数百万人使用。

1.基本概念

生产者（Producer）
  1. 生产者是一个发送消息的程序。发送消息的程序可以是任何语言编写的，只要它能够连接到RabbitMQ服务器，并且能够发送消息到RabbitMQ服务器。
复制代码
消费者（Consumer）
  1. 消费者是一个接收消息的程序。接收消息的程序可以是任何语言编写的，只要它能够连接到RabbitMQ服务器，并且能够从RabbitMQ服务器接收消息。
复制代码
队列（Queue）
  1. 队列是RabbitMQ的内部对象，用于存储消息。多个生产者可以向一个队列发送消息，多个消费者可以尝试从一个队列接收消息。队列支持多种消息分发策略。
复制代码
互换机（Exchange）
  1. 交换机是消息的分发中心。它接收来自生产者的消息，然后将这些消息分发给队列。交换机有多种类型，包括直连交换机、主题交换机、扇形交换机、头交换机。
复制代码
绑定（Binding）
  1. 绑定是交换机和队列之间的关联关系。绑定可以使用路由键进行绑定，也可以使用通配符进行绑定。
复制代码
路由键（Routing Key）
  1. 路由键是生产者发送消息时附带的一个属性。路由键的作用是决定消息被分发到哪个队列。
复制代码
通配符（Wildcard）
  1. 通配符是一种模式匹配的方式。RabbitMQ支持两种通配符：`*`和`#`。
复制代码
绑定键（Binding Key）
  1. 绑定键是交换机和队列之间的关联关系。绑定键可以使用路由键进行绑定，也可以使用通配符进行绑定。
复制代码
长期化（Durable）
  1. 持久化是指RabbitMQ服务器重启后，消息是否还存在。持久化可以应用到交换机、队列、绑定、消息等。
复制代码
确认机制（Acknowledge）
  1. 确认机制是指消费者接收到消息后，向RabbitMQ服务器发送一个确认消息。RabbitMQ服务器收到确认消息后，会删除这条消息。
  3. 自动确认
  4.         消费者接收到消息后，RabbitMQ服务器会自动删除这条消息。
  6. 手动确认
  7.         消费者接收到消息后，需要向RabbitMQ服务器发送一个确认消息。RabbitMQ服务器收到确认消息后，会删除这条消息。
复制代码
拒绝机制（Reject）
  1. 拒绝机制是指消费者接收到消息后，向RabbitMQ服务器发送一个拒绝消息。RabbitMQ服务器收到拒绝消息后，会将这条消息重新发送给其他消费者。
复制代码
死信队列（Dead Letter Queue）
  1. 死信队列是指消息被拒绝、过期或者达到最大重试次数后，会被发送到死信队列。
复制代码
消息过期（Message TTL）
  1. 消息过期是指消息在指定时间内没有被消费者消费，会被删除。
复制代码
消息优先级（Message Priority）
  1. 消息优先级是指消息在队列中的优先级。消息优先级高的消息会被优先消费。
复制代码
消息分发
  1. 消息分发是指消息在队列中的分发策略。消息分发策略包括轮询分发、公平分发、负载均衡分发。
复制代码
2.环境搭建

Docker 安装 RabbitMQ
  1. docker run -d --name rabbitmq -p 5672:5672 -p 15672:15672 --restart=always --hostname my-rabbit -e RABBITMQ_DEFAULT_USER=admin -e RABBITMQ_DEFAULT_PASS=admin -e TZ=Asia/Shanghai rabbitmq:management
复制代码

  • -d：后台运行
  • --restart：重启战略
  • --name：容器名称
  • -p：端口映射
  • --hostname：主机名
  • -e：环境变量

    • RABBITMQ_DEFAULT_USER：默认用户名
    • RABBITMQ_DEFAULT_PASS：默认暗码

  • TZ：时区
  • rabbitmq:management：镜像名称
Docker Compose 安装 RabbitMQ
  1. version: "3.1"
  2. services:
  3. rabbitmq:
  4.     restart: always
  5.     image: rabbitmq:management
  6.     container_name: rabbitmq
  7.     hostname: my-rabbit
  8.     ports:
  9.         - 5672:5672
  10.         - 15672:15672 # RabbitMQ管理界面端口
  11.     environment:
  12.         TZ: Asia/Shanghai
  13.             RABBITMQ_DEFAULT_USER: admin
  14.             RABBITMQ_DEFAULT_PASS: admin
复制代码

  • restart：重启战略
  • image：镜像名称
  • container_name：容器名称
  • hostname：主机名
  • ports：端口映射
  • environment：环境变量

    • TZ：时区
    • RABBITMQ_DEFAULT_USER：默认用户名
    • RABBITMQ_DEFAULT_PASS：默认暗码

  • rabbitmq:management：镜像名称
3.使用

客户端SDK代码在GitHub：https://github.com/Tangtang1997/IKunLibrary
新建 TestRequest 类，实现 IRabbitMqRequest 接口，定义消息体
  1. public class TestRequest : IRabbitMqRequest
  2. {
  3. /// <summary>
  4. /// 重试次数
  5. /// </summary>
  6. public int RetryCount { get; set; }
  8. #region 自定义字段
  10. /// <summary>
  11. /// id
  12. /// </summary>
  13. public string Id { get; set; } = default!;
  15. /// <summary>
  16. /// 名称
  17. /// </summary>
  18. public string Name { get; set; } = default!;
  20. /// <summary>
  21. /// 年龄
  22. /// </summary>
  23. public int Age { get; set; }
  25. #endregion
  26. }
复制代码
新建TestRequestHandler类，实现IRabbitMqRequestHandler接口，处理消息
  1. public class TestRequestHanlder : IRequestProcessorHandler<TestRequest>
  2. {
  3. private readonly ILogger<TestRequestHanlder> _logger;
  5. public TestRequestHanlder(ILogger<TestRequestHanlder> logger)
  6. {
  7.     _logger = logger;
  8. }
  10. public Task StartAsync(CancellationToken cancellationToken)
  11. {
  12.     return Task.CompletedTask;
  13. }
  15. public Task StopAsync(int milliseconds, CancellationToken cancellationToken = default)
  16. {
  17.     return Task.CompletedTask;
  18. }
  20. public async Task HandleAsync(TestRequest request, CancellationToken cancellationToken = default)
  21. {
  22.     _logger.LogInformation($"开始处理消息: {request.Id}");
  24.     //模拟处理消息耗时操作
  25.     await Task.Delay(1000, cancellationToken);
  27.     _logger.LogInformation($"消息处理完成: {request.Id}");
  28.     }
  29. }
复制代码
使用 IHostedService 来托管服务
  1. public class SampleHostedService : IHostedService
  2. {
  3. private readonly IConsumerProcessorManager<TestRequest> _consumerProcessorManager;
  4. private readonly IHostApplicationLifetime _applicationLifetime;
  5. private readonly ILogger<SampleHostedService> _logger;
  7. public SampleHostedService(
  8.     IConsumerProcessorManager<TestRequest> consumerProcessorManager,
  9.     IHostApplicationLifetime applicationLifetime,
  10.     ILogger<SampleHostedService> logger)
  11. {
  12.     _consumerProcessorManager = consumerProcessorManager;
  13.     _applicationLifetime = applicationLifetime;
  14.     _logger = logger;
  15. }
  17. public async Task StartAsync(CancellationToken cancellationToken)
  18. {
  19.     _applicationLifetime.ApplicationStarted.Register(() =>
  20.     {
  21.         _logger.LogInformation("SampleHostedService is starting.");
  22.         _consumerProcessorManager.StartAsync(cancellationToken);
  23.     });
  25.     _applicationLifetime.ApplicationStopping.Register(() =>
  26.     {
  27.         _logger.LogInformation("SampleHostedService is stopping.");
  28.         _consumerProcessorManager.StopAsync(3000, cancellationToken);
  29.     });
  31.     await Task.CompletedTask;
  32. }
  34. public async Task StopAsync(CancellationToken cancellationToken)
  35. {
  36.     await Task.CompletedTask;
  37.     }
  38. }
复制代码
注册并启用服务
  1. IHost host = Host.CreateDefaultBuilder(args)
  2.     .ConfigureServices(services =>
  3. {
  4.     services.AddHostedService<SampleHostedService>();
  6.     var configuration = services.BuildServiceProvider().GetRequiredService<IConfiguration>();
  8.     var hostName = configuration["RabbitMq:Host"] ?? throw new Exception("HostName is not configured");
  9.     var port = int.Parse(configuration["RabbitMq:Port"] ?? throw new Exception("Port is not configured"));
  10.     var userName = configuration["RabbitMq:Username"] ?? throw new Exception("Username is not configured");
  11.     var password = configuration["RabbitMq:Password"] ?? throw new Exception("Password is not configured");
  12.     var queueName = configuration["RabbitMq:QueueName"] ?? throw new Exception("QueueName is not configured");
  14.     services.AddRabbitMq<TestRequest, TestRequestHanlder>(options =>
  15.     {
  16.         options.UseSsl = false;
  17.         options.HostName = hostName;
  18.         options.Port = port;
  19.         options.UserName = userName;
  20.         options.Password = password;
  21.         options.Durable = true;
  22.         options.NetworkRecoveryInterval = 10000;
  23.         options.ExchangeType = ExchangeType.Direct;
  24.         options.QueueName = queueName;
  25.         options.Exchange = $"{queueName}_SERVICE_EXCHANGE";
  26.         options.RoutingKey = $"{queueName}_ROUTING_KEY";
  27.         options.DeadLetterExchange = $"{queueName}_SERVICE_EXCHANGE_DEAD";
  28.         options.DeadLetterQueueName = $"{queueName}_DEAD";
  29.         options.DeadLetterRoutingKey = $"{queueName}_ROUTING_KEY";
  30.     });
  31. })
  32. .Build();
  34. await host.RunAsync();
复制代码
4.参考资料

https://www.cnblogs.com/Tangtang1997/p/18067763

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。
回复

使用道具 举报

0 个回复

倒序浏览
返回列表

快速回复

高级模式
B Color Image Link Quote Code Smilies
您需要登录后才可以回帖 登录 or 立即注册

本版积分规则

飞不高

金牌会员
这个人很懒什么都没写!

楼主热帖

标签云

挺好的 服务器
微信订阅号
微信服务号
微信客服
小程序
H5
关于我们 商务合作 网站地图
©Copyright 2022-2024 杭州祥升科技有限公司 版权所有 浙ICP备20004199号
快速回复 返回顶部 返回列表