메시지 생성 및 소비

MassTransit 등록하기

builder.Services.AddMassTransit(x =>
{
    x.AddConsumers(typeof(Program).Assembly);
    x.UsingRabbitMq((context, cfg) =>
    {
        cfg.Host("localhost");
        cfg.ConfigureEndpoints(context);
    });
});

소비자 정의

public class OrderPlacedConsumer : IConsumer<OrderPlaced>
{
    public async Task Consume(ConsumeContext<OrderPlaced> context)
    {
        OrderPlaced message = context.Message;
        // handle the event
    }
}

이벤트 게시

public class OrderService
{
    private readonly IPublishEndpoint _publish;
    public OrderService(IPublishEndpoint publish) => _publish = publish;

    public Task PlaceOrderAsync(Guid id, string email) =>
        _publish.Publish(new OrderPlaced(id, email));
}

명령 전송

var endpoint = await _sendProvider
    .GetSendEndpoint(new Uri("queue:charge-payment"));
await endpoint.Send(new ChargePayment(orderId, amount));

Publish와 Send 요약

ConsumeContext의 강력한 기능

public async Task Consume(ConsumeContext<OrderPlaced> context)
{
    await context.Publish(new InvoiceRequested(context.Message.OrderId));
}

요청-응답

var response = await _client
    .GetResponse<OrderStatus>(new GetOrder(orderId));
OrderStatus status = response.Message;

Endpoint 이름 지정

메시지 직렬화

동시성

cfg.ReceiveEndpoint("orders", e =>
{
    e.PrefetchCount = 16;
    e.ConcurrentMessageLimit = 8;
    e.ConfigureConsumer<OrderPlacedConsumer>(context);
});

소비자 테스트

빠른 확인

요약