Vert.x Local EventBus

吱吱歪 · · 81 次点击 · · 开始浏览    

看了Nacos服务端的event发布订阅方式其实跟eventbus都是类似的东西,一直以来也只是对eventbus知名而不知其所以,因此单独看了一下verxt里的eventbus实现,这个框架也不是太多人知道,但是其实里面的那一套都是一样的。
本文仅为本人学习用记录,文笔不好

vertx里的EventBus

eventbus是actor模型中连接actor与actor之间的通信管道,类比一下就是golang里面的channel,但是channe作用于协程到协程,而eventbus是作用于角色之间的,每个角色都有自己的业务逻辑。

EventBus能做什么

发送信息给其他actor,获取接受信息的actor的返回值,根据返回继续自己的业务逻辑,像一个披着mq皮的rpc。

如何实现

在看别人的源码前得思考一个问题,这个需求如果交由你来做,该如何实现?
需要实现的是:
1.发布订阅机制
2.获取订阅者的返回并通知发布者
3.解决返回结果的三态问题

源码实现

先介绍下参与对象

image.png

上图就是eventbus里的主要参与人员
message:要发送的POJO都会被包装成Message,原框架只支持JsonObject与基础类型的编码方式,其他的方式POJO映射需要自己实现。
eventbus:eventbus的实现逻辑,主要发送分为send(单向),request(双向),publish(广播)三种形式,与consume方法对消息进行消费。通过String类型的address作为标识。
ReplyHandler:实现了注册类,用于处理双向通信中的回调,即调用者后续逻辑
MessageConsumerHandler:同样也实现了注册类,处理消费者的逻辑。
InboundDeliveryContext:接受到的message的上下文
OutBoundDeliveryContext:发送的message的上下文
HandlerHolder:其实就是replyHandler与messageConsumerHandler的父类,因为HandlerRegister是个抽象类。

Request方法

位于EventBus接口,双向通信发送的入口,是一个default接口
入参:1.订阅者订阅的address,2.要发送的消息,3.接收回调的信息后的业务逻辑。Handler<>是一个函数式接口,里面就可以写我们的回调逻辑啦。

  @Fluent
  default <T> EventBus request(String address, @Nullable Object message, DeliveryOptions options, Handler<AsyncResult<Message<T>>> replyHandler) {
    Future<Message<T>> reply = request(address, message, options);
    reply.onComplete(replyHandler);
    return this;
  }

request将address,message,与默认配置传入EventBusImpl.request函数中,返回一个Future,这里的Future跟netty的差不多,会在Future达成后运行我们定义的回调逻辑。因此reply方法将信息发送了出去,并注册了回调的Future,在订阅者返回时召唤线程跑业务逻辑。

EventBusImpl.request

包装message很简单就不看了,我们先不看回调注册,因为调用很深。先看发送有助于后续理解。

  @Override
  public <T> Future<Message<T>> request(String address, Object message, DeliveryOptions options) {
    //封装Message,指定编码,codecManage参与
    MessageImpl msg = createMessage(true, address, options.getHeaders(), message, options.getCodecName());
    //注册handler
    ReplyHandler<T> handler = createReplyHandler(msg, true, options);
    //发送
    sendOrPubInternal(msg, options, handler, null);
    return handler.result();
  }

我们将包装好的信息,配置,注册的回调交给sendOrPubInternal方法

EventBusImpl.sendOrPubInternal

这里将传入的message,replyHandler都封装为一个OutboundDeliveryContext

public <T> void sendOrPubInternal(MessageImpl message, DeliveryOptions options,
                                    ReplyHandler<T> handler, Promise<Void> writePromise) {
    checkStarted();
    sendOrPubInternal(newSendContext(message, options, handler, writePromise));
  }

  public <T> void sendOrPubInternal(OutboundDeliveryContext<T> senderCtx) {
    checkStarted();
    senderCtx.iter = sendInterceptors.iterator();
    senderCtx.bus = this;
    senderCtx.metrics = metrics;
    senderCtx.next();
  }

这里将bus中的迭代器(用于拦截相当于切面处理),自身传入context中,调用context的next方法。

OutboundDeliveryContext.next

这里一大串都是走切面逻辑与tracer的,直接看最后的sendOrPub(this),调用过程最后走到deliverMessageLocally方法,

@Override
  public void next() {
    if (iter.hasNext()) {
      Handler<DeliveryContext> handler = iter.next();
      try {
        if (handler != null) {
          handler.handle(this);
        } else {
          next();
        }
      } catch (Throwable t) {
        EventBusImpl.log.error("Failure in interceptor", t);
      }
    } else {
      VertxTracer tracer = ctx.tracer();
      if (tracer != null) {
        if (message.trace == null) {
          src = true;
          BiConsumer<String, String> biConsumer = (String key, String val) -> message.headers().set(key, val);
          TracingPolicy tracingPolicy = options.getTracingPolicy();
          if (tracingPolicy == null) {
            tracingPolicy = TracingPolicy.PROPAGATE;
          }
          message.trace = tracer.sendRequest(ctx, SpanKind.RPC, tracingPolicy, message, message.send ? "send" : "publish", biConsumer, MessageTagExtractor.INSTANCE);
        } else {
          // Handle failure here
          tracer.sendResponse(ctx, null, message.trace, null, TagExtractor.empty());
        }
      }
      bus.sendOrPub(this);
    }
  }

  protected <T> void sendOrPub(OutboundDeliveryContext<T> sendContext) {
    sendLocally(sendContext);
  }

  private <T> void sendLocally(OutboundDeliveryContext<T> sendContext) {
    ReplyException failure = deliverMessageLocally(sendContext.message);
    if (failure != null) {
      sendContext.written(failure);
    } else {
      sendContext.written(null);
    }
  }

其实调用到最后就是走一个execute方法进行线程的切换,从一个eventloop到另一个eventloop

  void receive(MessageImpl msg) {
    if (bus.metrics != null) {
      bus.metrics.scheduleMessage(metric, msg.isLocal());
    }
    context.nettyEventLoop().execute(() -> {
      // Need to check handler is still there - the handler might have been removed after the message were sent but
      // before it was received
      if (!doReceive(msg) && bus.metrics != null) {
        bus.metrics.discardMessage(metric, msg.isLocal(), msg);
      }
    });
  }

有疑问加站长微信联系(非本文作者)

本文来自:简书

感谢作者:吱吱歪

查看原文:Vert.x Local EventBus

入群交流(和以上内容无关):加入Go大咖交流群,或添加微信:liuxiaoyan-s 备注:入群;或加QQ群:701969077

81 次点击  
加入收藏 微博
暂无回复
添加一条新回复 (您需要 登录 后才能回复 没有账号 ?)
  • 请尽量让自己的回复能够对别人有帮助
  • 支持 Markdown 格式, **粗体**、~~删除线~~、`单行代码`
  • 支持 @ 本站用户;支持表情(输入 : 提示),见 Emoji cheat sheet
  • 图片支持拖拽、截图粘贴等方式上传