看了Nacos服务端的event发布订阅方式其实跟eventbus都是类似的东西，一直以来也只是对eventbus知名而不知其所以，因此单独看了一下verxt里的eventbus实现，这个框架也不是太多人知道，但是其实里面的那一套都是一样的。
本文仅为本人学习用记录，文笔不好
vertx里的EventBus
eventbus是actor模型中连接actor与actor之间的通信管道，类比一下就是golang里面的channel，但是channe作用于协程到协程，而eventbus是作用于角色之间的，每个角色都有自己的业务逻辑。
EventBus能做什么
发送信息给其他actor，获取接受信息的actor的返回值，根据返回继续自己的业务逻辑，像一个披着mq皮的rpc。
如何实现
在看别人的源码前得思考一个问题，这个需求如果交由你来做，该如何实现？
需要实现的是：
1.发布订阅机制
2.获取订阅者的返回并通知发布者
3.解决返回结果的三态问题
源码实现
先介绍下参与对象
上图就是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);
}
});
}
有疑问加站长微信联系（非本文作者）
