微服务实战(四):落地微服务架构到直销系统(将生产者与消费者接入消息总线)

码海星语
• 阅读 2080

微服务实战(四):落地微服务架构到直销系统(将生产者与消费者接入消息总线)

前一篇文章我们已经完成了基于RabbitMq实现的的消息总线,这篇文章就来看看生产者(订单微服务)与消费者(经销商微服务)如何接入消息总线实现消息的发送与消息的接收处理。

定义需要发送的消息:

下单消息要被发送到消息总线,并被经销商微服务的处理器处理。经销商微服务处理时,需要知道要对哪个经销商处理多少的PV值与电子币余额。这些信息就是事件消息需要承载的重要信息。

public class OrderCreatedProcessDealerEvent:BaseEvent

{
    public decimal OrderTotalPrice { get; set; }
    public decimal OrderTotalPV { get; set; }
    public Guid DealerId { get; set; }
    public Guid OrderId { get; set; }
    public OrderCreatedProcessDealerEvent(Guid dealerid,Guid orderid,decimal ordertotalprice,decimal
        ordertotalpv)
    {
        this.OrderTotalPrice = ordertotalprice;
        this.OrderTotalPV = ordertotalpv;
        this.DealerId = dealerid;
        this.OrderId = orderid;
    }
}

生产者(订单微服务)连接到消息总线:

生产者-订单微服务通过Asp.net core WebApi自带的依赖注入,连接到RabbitMq消息总线。

        services.AddSingleton<IEventHandlerExecutionContext>(new EventHandlerExecutionContext(services));
        var connectionFactory = new ConnectionFactory { HostName = "localhost" };
        services.AddSingleton<IEventBus>(sp => new RabbitMqEB(connectionFactory,
            sp.GetRequiredService<IEventHandlerExecutionContext>(), "exchange2", "direct", "ordereventqueue", 1));

从上面代码可以看出,生产者连接到了localhost的Rabbit服务器,并通过调用消息总线的构造函数,定义了发送消息的通道。构造函数具体内容可以查看上一篇文章。

生产者(订单微服务)发送消息到消息总线:
ieventbus.Publish(new OrderCreatedProcessDealerEvent(orderdto.DealerId,

                    orderid, order.OrderTotalPrice.TotalPrice, order.OrderTotalPV.TotalPV));

ieventbus是注入到订单微服务的构造函数中,并传递到订单创建的用例中。

实现消费者(经销商微服务)的消息处理器:

消费者会连接到消息总线,接收到特定类型的消息(这里是OrderCreatedProcessDealerEvent),会交给特定的处理器进行处理,所以需要先定义并实现消息处理器。

public class OrderCreatedEventHandler : IEventHandler

{
    ServiceLocator servicelocator = new ServiceLocator();
    public Task<bool> HandleAsync<TEvent>(TEvent @event) where TEvent : IEvent
    {
        var idealercontext = servicelocator.GetService<IDealerContext>();
        var irepository =
            servicelocator.GetService<IRepository>(new ParameterOverrides { { "context", idealercontext } });
        var idealerrepository = servicelocator.GetService<IDealerRepository>(new ParameterOverrides { { "context", idealercontext } });

//先将接收到的消息转换成特定类型
var ordercreatedevent = @event as OrderCreatedProcessDealerEvent;

        using (irepository)
        {
            try
            {
              //根据消息内容,处理自己的逻辑与持久化
                idealerrepository.SubParentEleMoney(ordercreatedevent.DealerId, ordercreatedevent.OrderTotalPrice);
                idealerrepository.AddDealerPV(ordercreatedevent.DealerId, ordercreatedevent.OrderTotalPV);
                irepository.Commit();
            }
            catch (EleMoneyNotEnoughException)
            {
                 //先不处理电子币余额不足的情况                   

            }
        }
        return Task.FromResult(true);
    }
}

消费者(经销商微服务)连接到消息总线:

需要在经销商微服务指定需要连接到的消息总线,并订阅哪个类型的消息交给哪个事件处理器进行处理。

        //用于侦听订单上下文传递的消息
        services.AddSingleton<IEventHandlerExecutionContext>(new EventHandlerExecutionContext(services));
        var connectionFactory = new ConnectionFactory { HostName = "localhost" };
        services.AddSingleton<IEventBus>(sp => new RabbitMqEB(connectionFactory,
            sp.GetRequiredService<IEventHandlerExecutionContext>(), "exchange2", "direct", "ordereventqueue", 2));
        var eventbus = app.ApplicationServices.GetService<IEventBus>();

//订阅消息

                    eventbus.Subscribe<OrderCreatedProcessDealerEvent, OrderCreatedEventHandler>();
                    

这样,两个微服务直接就能通过RabbitMq消息总线进行消息的发送、消息的接收与处理了,实现了解耦。

QQ讨论群:309287205
微服务实战视频请关注微信公众号:msshcj

点赞
收藏
评论区
推荐文章
Centos7安装RabbitMQ详细教程 - 附带软件基本解释 - CSDN博客
MQ引言什么是MQMQ:messageQueue翻译为消息队列,通过典型的生产者和消费者模型不断向消息队列中生产消息,消费者不断从队列中获取消息。因为消息的生产和消费都是一部的,而且只关心消息的发送和接收,没有业务逻辑的侵入,轻松的实现了系统之间的解耦。别名是消息中间件,通过利用高效的消息传递机制进行与平台无关的数据交流,并基于数据通信来进行分布式系
Stella981 Stella981
3年前
MQ对比之RabbitMQ & Redis
消息队列选择:RabbitMQ&RedisRabbitMQRabbitMQ是一个由erlang开发的AMQP(AdvancedMessageQueue)的开源实现的产品,RabbitMQ是一个消息代理,从“生产者”接收消息并传递消息至“消费者”,期间可根据规则路由、缓存、持久化消息。“生产者”也即message
Stella981 Stella981
3年前
Linux系统 Centos7 环境基于Docker部署Rocketmq服务
消息队列基本概述MQ,MessageQueue,基于TCP协议构建的简单协议,区别于具体的通信协议。基于通信协议定义和抽象的更高层次的通信模型,一般都是生产者和消费者模型,又或者说服务端和客户端模型。生产者/消费者模型:一般通过定义生产者和消费者实现消息通信从而屏
Stella981 Stella981
3年前
Spring Boot(七):RabbitMQ 详解
一、RabbitMQ简介RabbitMQ即一个消息队列,主要是用来实现应用程序的异步和解耦,同时也能起到消息缓冲,消息分发的作用。消息中间件在互联网公司的使用中越来越多,消息中间件最主要的作用是解耦,中间件最标准的用法是生产者生产消息传送到队列,消费者从队列中拿取消息并处理,生产者不用关心是谁来消费,消费者不用关心谁在生产消息,从而达到解耦的
Wesley13 Wesley13
3年前
3.rabbitmq
rabbitmq发布订阅模式模型组成一个消费者Producer,一个交换机Exchange,多个消息队列Queue,多个消费者Consumer一个生产者,多个消费者,每一个消费者都有自己的一个队列,生产者没有将消息直接发送到队列,而是发送到了交换机,每个队列绑定交换机,生产者发送
Wesley13 Wesley13
3年前
ActiveMQ简述,使用
官网地址:http://activemq.apache.org/参考文章:http://my.oschina.net/nk2011/blog/366395JMS支持两种消息发送和接收模型。一种称为P2P(PonittoPoint)模型,即采用点对点的方式发送消息。P2P模型是基于队列的,消息生产者发送消息到队列,消息消费者从队列中接收消息,队列的
Stella981 Stella981
3年前
RabbitMQ如何保证队列里的消息99.99%被消费?
1\.本篇概要其实,还有1种场景需要考虑:当消费者接收到消息后,还没处理完业务逻辑,消费者挂掉了,那消息也算丢失了?,比如用户下单,订单中心发送了1个消息到RabbitMQ里的队列,积分中心收到这个消息,准备给这个下单的用户增加20积分,但积分还没增加成功呢,积分中心自己挂掉了,导致数据出现问题。那么如何解
Stella981 Stella981
3年前
RabbitMQ——队列消息数
背景在实际使用过程,会遇到这么些情况:生产者发送的消息数量与消费者接收的消息数量不一致。例如生产者向rabbitmq投递了100条消息,消费者只从队列中接收到了80条消息,并且当前队列中已经没有任何消息。要定位这个问题,通常是分段来定位,一方面统计生产者到底发送了多少消息,一方面统计有多少消息是正确路由到
Wesley13 Wesley13
3年前
JMS消息的概念解释
1、默认生产者消息是持久的:会存数据库\消费者的持久:createDurableSubscriber是指消费者能收到所有它订阅时间点之后的消息,即使消费者注册后关闭,当它重启就能收到注册时间点之后所有的消息;即当此消费用户ID(AAA)在producer发送之前就已经注册,那么此id能收到producer发送的所有消息,如果是在produce
Stella981 Stella981
3年前
Spring Cloud Bus 消息总线
什么是消息总线?在微服务架构的系统中,我们通常会使用轻量级的消息代理来构建一个共用的消息主题让系统中所有微服务实例都连接上来,由于该主题中产生的消息会被所有实例监听和消费,所以我们称它为消息总线。在总线上的各个实例都可以方便地广播一些需要让其他连接在该主题上的实例都知道的消息,例如配置信息的变更或者其他一些管