译MassTransit 生产消息

简介: 生产消息 应用程序或服务可以使用两种不同的方法生产消息。可以使用Sead发送消息,也可以使用Publish发布消息。每个方法的行为是非常不同的,但是通过查看每个特定方法所涉及的消息类型,可以很容易理解。

生产消息

应用程序或服务可以使用两种不同的方法生产消息。可以使用Sead发送消息,也可以使用Publish发布消息。每个方法的行为是非常不同的,但是通过查看每个特定方法所涉及的消息类型,可以很容易理解。

当消息是 sent时,它使用DestinationAddress 传递交付到特定的端点。当消息是published,它不会发送到特定的端点,而是广播给订阅了该消息类型的任何消费者。对于这两个单独的行为, 我们描述作为命令发送的消息, 以及作为事件发布的消息。

在创建文档的消息契约部分时,将对此进行深入讨论。 

 发送命令

将命令发送到端点 需要 ISendEndpoint 引用,它可以从任何发送端点提供程序(支持ISendEndpointProvider的对象)中获得。应用程序应该始终使用与之最接近的对象来获取发送端点, 并且每次需要它时都应该这样做--应用程序不应缓存发送端点引用。例如,IBus 实例是一个发送端点提供程序,但它不应该被消费者用来获取ISendEndpoint.ConsumeContext 还可以提供发送端点,并且应该使用它,因为它更接近于消费者。

注意:这不能足够强调——总是从最接近的应用程序代码接口获得发送端点。使用会话、关联和发起标识符将消息流绑定在一起具有广泛的逻辑。通过跳过级别并在最接近的范围之外,可以丢失该信息,从而阻止有用的跟踪标识符被正确处理。

要从发送端点提供程序获得发送端点,请使用GetSendEndpoint()方法,如下所示。返回发送端点后,它就可以用于发送消息。

public async Task SendOrder(ISendEndpointProvider sendEndpointProvider)
{
    var endpoint = await sendEndpointProvider.GetSendEndpoint(_serviceAddress);

    await endpoint.Send(new SubmitOrder(...));
}

Send 方法有很多重载。因为MassTransit 是围绕filters和pipes构建的,所以pipes被用来定制发送的消息传递行为。还有一些有用的重载(通过扩展方法),使得 pipe 的构建更容易、更少噪音,等等。

通过接口发送

由于一般的建议是使用接口, 所以在不需要在下面创建消息类的情况下, 有一种方便的方法来初始化接口。虽然消息的版本化仍然需要支持多个接口的类,但下面显示了一种简单的发送接口消息的方法。

public interface SubmitOrder
{
    string OrderId { get; }
    DateTime OrderDate { get; }
    decimal OrderAmount { get; }
}

public async Task SendOrder(ISendEndpoint endpoint)
{
    await endpoint.Send<SubmitOrder>(new
    {
        OrderId = "27",
        OrderDate = DateTime.UtcNow,
        OrderAmount = 123.45m
    });
}

设置消息标题
有多种消息头可用以用于消息的相关性和跟踪。当发生故障时,也可以重写MassTransit 的一些默认行为。例如,当消费者抛出异常时,通常会发布故障。如果应用程序希望传递到特定地址的故障,则可以通过报头指定故障地址。如何做到这一点如下所示。

public interface SubmitOrder
{
    string OrderId { get; }
    DateTime OrderDate { get; }
    decimal OrderAmount { get; }
}

public async Task SendOrder(ISendEndpoint endpoint)
{
    await endpoint.Send<SubmitOrder>(new
    {
        OrderId = "27",
        OrderDate = DateTime.UtcNow,
        OrderAmount = 123.45m
    }, context => context.FaultAddress = new Uri("rabbitmq://localhost/order_faults"));
}

发布事件

消息的发布与消息的发送方式类似,但在这种情况下,使用单个IPublishEndpoint 。应用相同的端点规则,应该使用发布终结点的最接近实例。因此, 对消费者的 ConsumeContext, 以及 IBus 在消费者上下文之外发布的应用程序。

要发布消息,请参见下面的代码。 

public interface OrderSubmitted
{
    string OrderId { get; }
    DateTime OrderDate { get; }
}

public async Task NotifyOrderSubmitted(IPublishEndpoint publishEndpoint)
{
    await publishEndpoint.Publish<OrderSubmitted>(new
    {
        OrderId = "27",
        OrderDate = DateTime.UtcNow,
    });
}

 

目录
相关文章
|
5月前
|
消息中间件 存储 安全
【消息队列开发】 实现ConsumerManager类——消费消息的核心逻辑
【消息队列开发】 实现ConsumerManager类——消费消息的核心逻辑
|
3月前
|
消息中间件 大数据 Kafka
Kafka消息封装揭秘:从Producer到Consumer,一文掌握高效传输的秘诀!
【8月更文挑战第24天】在分布式消息队列领域,Apache Kafka因其实现的高吞吐量、良好的可扩展性和数据持久性备受开发者青睐。Kafka中的消息以Record形式存在,包括固定的头部与可变长度的消息体。生产者(Producer)将消息封装为`ProducerRecord`对象后发送;消费者(Consumer)则从Broker拉取并解析为`ConsumerRecord`。消息格式简化示意如下:消息头 + 键长度 + 键 + 值长度 + 值。键和值均为字节数组,需使用特定的序列化/反序列化器。理解Kafka的消息封装机制对于实现高效、可靠的数据传输至关重要。
72 4
|
3月前
|
消息中间件 负载均衡 Kafka
【Kafka消费秘籍】深入了解消费者组与独立模式,掌握消息消费的两种超能力!
【8月更文挑战第24天】Apache Kafka是一款高性能的分布式消息系统,支持灵活多样的消费模型以适应不同的应用场景。消息按主题组织,每个主题可划分为多个分区,确保消息顺序性。本文深入探讨了Kafka中的两大核心消费模式:消费者组(Consumer Group)和独立消费者(Standalone Consumer)。消费者组允许多个消费者协同工作,实现负载均衡及故障恢复,是最常用的消费模式。独立消费者模式则适用于需要高度定制化处理逻辑的场景,如消息重放等。通过对比这两种模式的特点和提供的示例代码,开发者可以根据具体需求选择最合适的消费策略,从而更好地利用Kafka构建高效的数据流应用程序。
75 3
|
3月前
|
消息中间件 存储 负载均衡
RocketMQ消费者消费消息核心原理(含长轮询机制)
这篇文章深入探讨了Apache RocketMQ消息队列中消费者消费消息的核心原理,特别是长轮询机制。文章从消费者和Broker的交互流程出发,详细分析了Push和Pull两种消费模式的内部实现,以及它们是如何通过长轮询机制来优化消息消费的效率。文章还对RocketMQ的消费者启动流程、消息拉取请求的发起、Broker端处理消息拉取请求的流程进行了深入的源码分析,并总结了RocketMQ在设计上的优点,如单一职责化和线程池的使用等。
RocketMQ消费者消费消息核心原理(含长轮询机制)
|
5月前
|
消息中间件 监控 中间件
中间件消费者处理消息
【6月更文挑战第6天】
34 3
|
6月前
|
移动开发 小程序 Go
【社区每周】小程序消息订阅插件升级为消息订阅接口(2022年8月第五期)
【社区每周】小程序消息订阅插件升级为消息订阅接口(2022年8月第五期)
40 0
|
6月前
|
消息中间件 缓存 Java
RabbitMQ在项目中做什么用?怎么消费消息?具体怎么使用的?
RabbitMQ在项目中做什么用?怎么消费消息?具体怎么使用的?
|
消息中间件
ActiveMQ消费者消费消息(点对点模式)
上篇博客写了生产者生产消息:ActiveMQ向消息队列存入消息
|
消息中间件 存储 运维
RocketMQ 消息集成:多类型业务消息-普通消息
本篇将从业务集成场景的诉求开始,介绍 RocketMQ 作为业务消息集成方案的核心能力和优势,通过功能场景、应用案例以及最佳实践等角度介绍 RocketMQ 普通消息类型的使用。
291 0
RocketMQ  消息集成:多类型业务消息-普通消息
|
消息中间件 RocketMQ 开发者
消息消费方准备工作|学习笔记
快速学习消息消费方准备工作
消息消费方准备工作|学习笔记