JiShe.CollectBus/JiShe.CollectBus.RabbitMQ/JiSheCollectBusRabbitMqModule.cs

using JiShe.CollectBus.Common.Interfaces;
using JiShe.CollectBus.RabbitMQ.Consumers;
using MassTransit;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Hosting;

namespace JiShe.CollectBus.RabbitMQ
{
    public class JiSheCollectBusRabbitMqModule: IJiSheModule
    {
        public void ConfigureServices(IServiceCollection services, HostBuilderContext hostContext)
        {
            var configuration = hostContext.Configuration;
            services.AddMassTransit(x =>
            {
                x.AddConsumer<MessageReceivedConsumer>();
                x.AddConsumer<MessageIssuedConsumer>();
                x.AddConsumer<MessageReceivedLoginConsumer>();
                x.AddConsumer<MessageReceivedHeartbeatConsumer>();
                x.UsingRabbitMq((context, cfg) =>
                {
                    cfg.Host(configuration["MQ:Host"], ushort.Parse(configuration["MQ:Port"] ?? string.Empty), configuration["MQ:VirtualHost"], h =>
                    {
                        h.Username(configuration["MQ:UserName"] ?? string.Empty);
                        h.Password(configuration["MQ:Password"] ?? string.Empty);
                    });

                    // 消息接收队列
                    cfg.ReceiveEndpoint(configuration["MQ:Queue:Received"] ?? string.Empty,configurator =>
                    {
                        configurator.ConfigureConsumeTopology = false;
                        configurator.Consumer<MessageReceivedConsumer>(context);
                        configurator.Durable = true;
                    });
                    // 登录
                    cfg.ReceiveEndpoint($"{configuration["MQ:Queue:Received"]}_Login", configurator =>
                    {
                        configurator.ConfigureConsumeTopology = false;
                        configurator.Consumer<MessageReceivedLoginConsumer>(context);
                        configurator.Durable = true;
                    });
                    // 心跳
                    cfg.ReceiveEndpoint($"{configuration["MQ:Queue:Received"]}_Heartbeat", configurator =>
                    {
                        configurator.ConfigureConsumeTopology = false;
                        configurator.Consumer<MessageReceivedHeartbeatConsumer>(context);
                        configurator.Durable = true;
                    });
                    // 消息下发队列
                    cfg.ReceiveEndpoint(configuration["MQ:Queue:Issued"] ?? string.Empty, configurator =>
                    {
                        configurator.ConfigureConsumeTopology = false;
                        configurator.Consumer<MessageIssuedConsumer>(context);
                        configurator.Durable = true;
                    });
                });
            });
        }
    }
}