Vert.x EventBus：一个地址，解耦所有组件

EventBus 是 Vert.x 的"神经系统"，目标只有一件事——让分布式系统里各组件通过地址松耦合通信。一套 API 同时覆盖同 JVM、跨进程、跨集群，还自带三种通信模式和负载均衡。这篇把它的核心作用、三个通信模式、六个真实场景，以及一个地址怎么打通整套系统讲清楚。

· 7 分钟阅读 分布式
Vert.x EventBus：一个地址，解耦所有组件

如果说 Vert.x 靠”事件循环 + 非阻塞”解决了”一个请求怎么跑得更快”，那 EventBus 解决的是另一件事：一堆组件之间怎么互相说话，还能互不打听。

它被称为 Vert.x 的”神经系统”。设计目标很单纯——让系统里各个部分通过”地址（Address）“松耦合通信。发送方只管往某个地址发消息，接收方在那个地址上挂着监听，两边谁也不用认识谁。更重要的是，同一套 API 既能让同一个 JVM 里的 Verticle 通信，也能在跨 JVM 的集群里透明地通信。

如果你熟悉 Spring，可以把它类比成 ApplicationEvent 的”Pro 版”：Spring 的事件只在单个应用上下文里转，EventBus 天然支持集群、多语言和高吞吐。

一、EventBus 到底解决了什么

1. 解耦：只知道地址，不认识对方

这是它存在的根本理由。订单模块要通知库存模块扣减，传统写法订单模块得持有库存模块的引用（HTTP Client、Feign、RPC stub……）。一旦库存模块的地址或协议变了，订单模块跟着改。

EventBus 把它变成：订单模块只往 inventory.deduct 这个地址发消息，至于谁在听、听的人换了没有，它一概不管。各 Verticle 可以独立开发、独立部署、独立扩缩。

2. 统一通信抽象：一套 API，三种距离

同一行 vertx.eventBus()，在不同部署形态下表现不同，调用方无感：

  • 同 JVM 内的多个 Verticle 通信；
  • 同进程多线程通信；
  • **跨 JVM（集群）**的分布式通信（配好 Cluster Manager 后自动生效）。

你写代码时不用关心”对方在本机还是隔壁机器”，EventBus 帮你路由。

3. 三种通信模式

这是最容易混的，先用一句话区分三个动作：

动作方法语义接收方数量
发布/订阅publish(address, msg)广播，所有人收到所有订阅者
点对点send(address, msg)轮询发给其中一个一个（自动负载均衡）
请求/响应request(address, msg, replyHandler)发完等对方 reply一个，且能拿回结果
// 广播：所有监听者都收到（适合"通知类"消息）
vertx.eventBus().publish("order.created", orderJson);

// 点对点：多个同地址消费者里轮询一个（适合"任务分发"）
vertx.eventBus().send("image.process", taskJson);

// 请求/响应：发完等回包（适合"要结果的调用"）
vertx.eventBus().<JsonObject>request("user.info", userId, reply -> {
    if (reply.succeeded()) {
        JsonObject user = reply.result().body();
    }
});

4. 跨语言

EventBus 上的消息按 JSON 约定传输，所以 Java、Kotlin、Scala、JavaScript、Python 等写的 Verticle 可以互相通信——这对异构系统混编很实用。

二、场景一：微服务内部通信

单体拆成多个 Verticle 后，它们之间怎么协调？用地址串起来即可。

// 订单服务：只管往地址发，不关心谁在听
public class OrderVerticle extends AbstractVerticle {
    @Override
    public void start() {
        vertx.eventBus().consumer("order.created", msg -> {
            JsonObject order = (JsonObject) msg.body();
            // 通知库存、支付——都是 publish/send 到地址
            vertx.eventBus().send("inventory.deduct", order);
            vertx.eventBus().send("payment.process", order);
        });
    }
}

// 库存服务：独立部署，订阅自己的地址
public class InventoryVerticle extends AbstractVerticle {
    @Override
    public void start() {
        vertx.eventBus().consumer("inventory.deduct", msg -> {
            JsonObject order = (JsonObject) msg.body();
            // 扣减库存……
            msg.reply("库存扣减成功");
        });
    }
}

多个 InventoryVerticle 实例都订阅 inventory.deduct 时，send 会自动轮询——水平扩展零成本。

三、场景二：事件驱动架构（EDA）

用户注册后，要发欢迎邮件、发券、更新统计、写审计。把这些后续动作全塞进注册方法，是经典的”上帝函数”。用 EventBus 拆开：

public class UserRegistrationVerticle extends AbstractVerticle {
    @Override
    public void start() {
        vertx.eventBus().consumer("user.registered", msg -> {
            JsonObject user = (JsonObject) msg.body();
            // 只管 publish，下游各自认领
            vertx.eventBus().publish("user.welcome.email", user);
            vertx.eventBus().publish("user.coupon.grant", user);
            vertx.eventBus().publish("user.stats.update", user);
            vertx.eventBus().publish("user.audit.log", user);
        });
    }
}

// 邮件服务：只关心 welcome.email
public class EmailVerticle extends AbstractVerticle {
    @Override
    public void start() {
        vertx.eventBus().consumer("user.welcome.email", msg -> {
            sendWelcomeEmail(((JsonObject) msg.body()).getString("email"));
        });
    }
}

新增一个”注册后送积分”的功能？加一个消费者 Verticle 就行，注册主流程一行不动。这就是事件驱动最实在的好处。

四、场景三：实时数据推送（WebSocket）

股票行情、比分直播、聊天室——本质都是”服务端有更新就推给订阅客户端”。下面这个例子特意把 consumer 注册放在连接建立时做一次，而不是每次收到消息都注册（后者会泄漏 consumer）：

public class StockVerticle extends AbstractVerticle {

    @Override
    public void start() {
        vertx.createHttpServer()
            .webSocketHandler(ws -> {
                String clientId = ws.textHandlerID();
                // 客户端订阅某只股票：为这个连接注册一个 consumer，且只注册一次
                ws.textMessageHandler(symbol -> {
                    String address = "stock." + symbol;
                    vertx.eventBus().<JsonObject>consumer(address, msg -> {
                        ws.writeTextMessage(msg.body().encode());
                    });
                });
                ws.closeHandler(v -> { /* 连接关闭时记得 unregister，避免泄漏 */ });
            })
            .listen(8080);

        // 行情源：周期性往地址 publish，所有订阅该地址的 WS 都收到
        vertx.setPeriodic(1000, id -> {
            JsonObject quote = new JsonObject()
                .put("symbol", "AAPL")
                .put("price", 150 + Math.random() * 10);
            vertx.eventBus().publish("stock.AAPL", quote);
        });
    }
}

注意：consumer 注册是有去有回的——consumer() 返回一个 MessageConsumer，不需要时调用它的 unregister()。示例里连接在 closeHandler 中应把对应 consumer 注销掉，否则客户端来来去去，consumer 越积越多。这是用 EventBus 做推送时最容易踩的坑。

五、场景四：集群下的负载均衡（任务队列）

图片处理、视频转码、批量计算这类重活，天然适合丢给一堆 Worker。EventBus 的 send + 多实例部署就是现成的任务队列：

// 派发器：周期性投递任务，send 自动轮询到某个 Worker
public class TaskDispatcherVerticle extends AbstractVerticle {
    @Override
    public void start() {
        vertx.setPeriodic(1000, id -> {
            JsonObject task = new JsonObject().put("imageId", "img_" + System.currentTimeMillis());
            vertx.eventBus().<String>request("image.process", task, reply -> {
                if (reply.succeeded()) System.out.println("done: " + reply.result().body());
            });
        });
    }
}

// 工作节点：耗时操作包进 executeBlocking，交 Worker 线程，不挡 Event Loop
public class ImageWorkerVerticle extends AbstractVerticle {
    @Override
    public void start() {
        vertx.eventBus().<JsonObject>consumer("image.process", msg -> {
            JsonObject task = msg.body();
            vertx.executeBlocking(promise -> {
                promise.complete(processImage(task.getString("imageId")));
            }, res -> msg.reply((String) res.result()));
        });
    }
}

// 部署 10 个 worker 实例，send 自动在它们之间轮询
vertx.deployVerticle(ImageWorkerVerticle.class.getName(),
        new DeploymentOptions().setInstances(10));

这里有个 Vert.x 的铁律顺带提一句：耗 CPU/阻塞的操作必须进 executeBlocking，否则会卡住 Event Loop 线程，拖垮整个节点的吞吐。

六、场景五：API 网关聚合

一个仪表盘要拼用户、订单、消息、积分四块数据，各自是独立服务。用 CompositeFuture.all 并行请求再汇总：

public class ApiGatewayVerticle extends AbstractVerticle {
    @Override
    public void start() {
        vertx.createHttpServer()
            .requestHandler(req -> {
                String userId = req.getParam("userId");
                Future<JsonObject> userFuture  = getUserInfo(userId);
                Future<JsonArray>  ordersFuture = getOrders(userId);
                Future<JsonArray>  messagesFuture = getMessages(userId);
                Future<Integer> pointsFuture = getPoints(userId);

                CompositeFuture.all(userFuture, ordersFuture, messagesFuture, pointsFuture)
                    .onSuccess(c -> req.response().end(new JsonObject()
                        .put("user", userFuture.result())
                        .put("orders", ordersFuture.result())
                        .put("messages", messagesFuture.result())
                        .put("points", pointsFuture.result()).encode()))
                    .onFailure(err -> req.response().setStatusCode(500).end(err.getMessage()));
            })
            .listen(8080);
    }

    private Future<JsonObject> getUserInfo(String userId) {
        Promise<JsonObject> p = Promise.promise();
        vertx.eventBus().<JsonObject>request("user.info", userId,
            reply -> p.complete(reply.result().body()));
        return p.future();
    }
    // getOrders / getMessages / getPoints 同理
}

四路请求并行，all 等全部完成再聚合——比串行调用快好几倍，而且每个下游又是独立扩缩的 Verticle。

七、场景六：分布式日志收集

各服务不打日志到本地文件，而是把日志事件发到总线，由专门的收集者统一落库、对错误日志告警：

public class LogCollectorVerticle extends AbstractVerticle {
    @Override
    public void start() {
        vertx.eventBus().<JsonObject>consumer("log.collector", msg -> {
            JsonObject log = msg.body();
            saveToElasticsearch(log);
            if ("ERROR".equals(log.getString("level"))) {
                vertx.eventBus().send("alert.system", log);   // 错误日志转告警
            }
        });
    }
}

八、典型架构长什么样

                    ┌─────────────────────────────────────┐
                    │         API Gateway                  │
                    │    (HTTP Server + EventBus)          │
                    └────────────┬────────────────────────┘
                                 │
                    ┌────────────▼────────────────────────┐
                    │         Event Bus (核心总线)         │
                    └─┬──────────┬──────────┬────────┬────┘
                      │          │          │        │
            ┌─────────▼──┐ ┌────▼────┐ ┌───▼────┐ ┌──▼─────┐
            │ Order      │ │Payment  │ │Inventory│ │ User   │
            │ Service    │ │Service  │ │Service  │ │ Service│
            └────────────┘ └─────────┘ └─────────┘ └────────┘
                 │             │           │           │
            ┌────▼────┐   ┌────▼────┐ ┌────▼────┐ ┌──▼─────┐
            │  DB     │   │ 银行网关 │ │  Redis  │ │ MongoDB│
            └─────────┘   └─────────┘ └─────────┘ └────────┘

所有 Service 之间不直接引用，全走中间那条 Event Bus。网关负责把外部 HTTP 翻译成 EventBus 调用，Service 之间只认地址。

九、核心价值与边界

维度说明
解耦组件零直接依赖，独立演进
弹性动态增减实例，自动负载均衡
可扩展加功能 = 加消费者，主流程不动
性能异步非阻塞；官方基准在轻负载小消息下可达百万级消息/秒的量级（实际取决于消息大小和消费逻辑，非严格压测数字，别直接当承诺）
分布式配 Cluster Manager 后透明支持跨 JVM
多语言异构 Verticle 无缝通信

什么时候该用：

  • 微服务内部多个 Verticle 协调；
  • 事件驱动架构，解耦生产者与消费者；
  • 实时推送（WebSocket / SSE）；
  • 异步任务队列；
  • API 网关聚合；
  • 集中式日志 / 监控收集。

什么时候别用（换消息中间件）：

  • 大规模数据批量传输：上 Kafka / RabbitMQ；
  • 需要持久化保证：EventBus 本身不做消息持久化，节点挂了在途消息可能丢；
  • 复杂路由规则：交给专业消息中间件。

小结

EventBus 的设计哲学就三个词：简单、快速、对分布式友好。它把”组件怎么通信”这个最容易写出耦合的地方，收敛成”一个地址 + 三种动作”。在跑 Vert.x 的系统里（比如我们 FOLIO 的登录/认证模块也建立在 Vert.x 之上），EventBus 几乎就是内部通信的默认总线——会用了，模块之间的”硬连线”就自然松开了。

记住一句够用一辈子的：publish 是广播、send 是轮询、request 是等回包；consumer 用完记得 unregister。

奇妙感 本文采用署名-非商业性使用-相同方式共享协议，转载请注明出处与作者。

相关阅读

浏览全部 →
Vert.x 比 Spring 快，到底快在哪？

很多人以为 Vert.x 快是因为"线程池调得好"，其实差在架构哲学：Spring 传统模型是一个请求一个线程（阻塞），Vert.x 是事件循环 + 非阻塞。这篇把两套线程模型拆开讲，顺带说清 WebFlux 在哪、以及到底什么时候该用谁。

2026-10-20 · 5 分钟 · 后端

目录