当前位置： 首页 > article >正文

Vertx-EventBus篇

article 2025/2/26 0:35:53

简介

EventBus，又称消息总线，类似我们常见的消息中间件。支持点对点、请求与响应、发布订阅模式，支持跨服务跨语言通讯，分布式消息系统。

常见用法

点对点send

消息发送到某个地址上，Vertx把消息分发到注册到这个地址上的某个消费者上，若存在多个消费者，使用轮询算法选择一个消费者接收消息。

Producer

public class Producer extends AbstractVerticle {
    @Override
    public void start(Promise<Void> startPromise) throws Exception {
        vertx.setPeriodic(5000L, h -> {
            System.out.println("***********************");
            vertx.eventBus().send("test", "你吃了吗");
        });
        super.start(startPromise);
    }
}

Consumer

public class Consumer extends AbstractVerticle {
    @Override
    public void start(Promise<Void> startPromise) throws Exception {
        Random random = new Random();
        int num = random.nextInt(10000);
        vertx.eventBus().<String>consumer("test", msg -> {
            System.out.println(String.format("消费者%s接收到消息:%s", num, msg.body()));
        });
        super.start(startPromise);
    }
}

App

public class App {
    public static void main(String[] args) {
        Vertx vertx = Vertx.vertx();
        vertx.deployVerticle(Consumer.class.getName(), new DeploymentOptions().setInstances(3))
                .compose(res -> vertx.deployVerticle(Producer.class.getName()));
    }

}

AppRunLog

***********************
消费者7414接收到消息:你吃了吗
***********************
消费者6421接收到消息:你吃了吗
***********************
消费者5802接收到消息:你吃了吗

请求响应request

请求与响应也是点对点模式的一种，区别在于消费者可以回复结果，两者可以进行会话交流。

Producer

public class Producer extends AbstractVerticle {
    @Override
    public void start(Promise<Void> startPromise) throws Exception {
        vertx.setPeriodic(5000L, h -> {
            System.out.println("***********************");
            vertx.eventBus().<String>request("test", "你吃了吗")
                    .onComplete(res -> {
                        if (res.succeeded()) {
                            System.out.println("生产者收到回应:"+ res.result().body());
                        }
                    });
        });
        super.start(startPromise);
    }
}

Consumer

public class Consumer extends AbstractVerticle {
    @Override
    public void start(Promise<Void> startPromise) throws Exception {
        Random random = new Random();
        int num = random.nextInt(10000);
        vertx.eventBus().<String>consumer("test", msg -> {
            System.out.println(String.format("消费者%s接收到消息:%s", num, msg.body()));
            msg.reply(num+"吃过啦");
        });
        super.start(startPromise);
    }
}

App

public class App {
    public static void main(String[] args) {
        Vertx vertx = Vertx.vertx();
        vertx.deployVerticle(Consumer.class, new DeploymentOptions().setInstances(3))
                .compose(res -> vertx.deployVerticle(Producer.class.getName()));
    }

}

AppRunLog

***********************
消费者8585接收到消息:你吃了吗
生产者收到回应:8585吃过啦
***********************
消费者9672接收到消息:你吃了吗
生产者收到回应:9672吃过啦
***********************
消费者3615接收到消息:你吃了吗
生产者收到回应:3615吃过啦

发布订阅publish

消息发送到某个地址上，Vertx把消息分发到注册到这个地址上的所有消费者上。

Producer

public class Producer extends AbstractVerticle {
    @Override
    public void start(Promise<Void> startPromise) throws Exception {
        vertx.setPeriodic(5000L, h -> {
            System.out.println("***********************");
            vertx.eventBus().publish("test", "你吃了吗");
        });
        super.start(startPromise);
    }
}

Consumer

public class Consumer extends AbstractVerticle {
    @Override
    public void start(Promise<Void> startPromise) throws Exception {
        Random random = new Random();
        int num = random.nextInt(10000);
        vertx.eventBus().<String>consumer("test", msg -> {
            System.out.println(String.format("消费者%s接收到消息:%s", num, msg.body()));
        });
        super.start(startPromise);
    }
}

App

public class App {
    public static void main(String[] args) {
        Vertx vertx = Vertx.vertx();
        vertx.deployVerticle(Consumer.class.getName(), new DeploymentOptions().setInstances(3))
                .compose(res -> vertx.deployVerticle(Producer.class.getName()));
    }

}

AppRunLog

***********************
消费者5195接收到消息:你吃了吗
消费者9192接收到消息:你吃了吗
消费者2572接收到消息:你吃了吗
***********************
消费者2572接收到消息:你吃了吗
消费者5195接收到消息:你吃了吗
消费者9192接收到消息:你吃了吗

编码解码器

Vertx消息总线默认只对一些基础类型的消息提供编码解码，若想要发送一个实体对象消息，那么需要自定义消息编码解码器，不然会报错 No message codec for type.

自定义单个编解码器

User

@Data
public class User {
    private String id;
    private String name;
}

UserCodec

public class UserCodec  implements MessageCodec<User, User> {

    /**
     * 编码
     * @param buffer
     * @param user
     */
    @Override
    public void encodeToWire(Buffer buffer, User user) {
        Buffer encoded = Json.CODEC.toBuffer(user, false);
        buffer.appendInt(encoded.length());
        buffer.appendBuffer(encoded);
    }

    /**
     * 集群传输解码
     * @param pos
     * @param buffer
     * @return
     */
    @Override
    public User decodeFromWire(int pos, Buffer buffer) {
        int length = buffer.getInt(pos);
        pos += 4;
        Buffer slice = buffer.slice(pos, pos + length);
        String json = slice.toString();
        return JacksonUtil.jsonToBean(json, User.class);
    }

    /**
     * 本地传输解码
     * @param user
     * @return
     */
    @Override
    public User transform(User user) {
        return user;
    }

    @Override
    public String name() {
        return "user";
    }

    @Override
    public byte systemCodecID() {
        return -1;
    }
}

Producer

public class Producer extends AbstractVerticle {
    @Override
    public void start(Promise<Void> startPromise) throws Exception {
        vertx.setTimer(2000L, h -> {
            System.out.println("***********************");
            User user = new User();
            user.setId("1");
            user.setName("张三");
            vertx.eventBus().<User>request("test", user)
                    .onComplete(res -> {
                        if (res.succeeded()) {
                            System.out.println("生产者收到回应:"+ res.result().body());
                        }
                    });
        });
        super.start(startPromise);
    }
}

Consumer

public class Consumer extends AbstractVerticle {
    @Override
    public void start(Promise<Void> startPromise) throws Exception {
        vertx.eventBus().<User>consumer("test", msg -> {
            System.out.println(String.format("消费者接收到消息:%s",msg.body()));
            User user = new User();
            user.setId("2");
            user.setName("李四");
            msg.reply(user);
        });
        super.start(startPromise);
    }
}

App

public class App {
    public static void main(String[] args) {
        Vertx vertx = Vertx.vertx();
        vertx.eventBus().registerDefaultCodec(User.class, new UserCodec());
        vertx.deployVerticle(Consumer.class.getName())
                .compose(res -> vertx.deployVerticle(Producer.class.getName()));
    }

}

AppRunLog

***********************
消费者接收到消息:User(id=1, name=张三)
生产者收到回应:User(id=2, name=李四)

自定义公用编解码器

RestRequest

公用请求参数类

@Data
public class RestRequest<T> {
    private String reqTopic;
    private T reqBody;
}

ReqMessageCodec

公用请求参数消息编解码类

public class ReqMessageCodec<T> implements MessageCodec<RestRequest<T>, T> {



    @Override
    public void encodeToWire(Buffer buffer, RestRequest<T> request) {
        Buffer encoded = Json.CODEC.toBuffer(request, false);
        buffer.appendInt(encoded.length());
        buffer.appendBuffer(encoded);
    }



    @Override
    public T decodeFromWire(int pos, Buffer buffer) {
        int length = buffer.getInt(pos);
        pos += 4;
        Buffer slice = buffer.slice(pos, pos + length);
        String s = slice.toString();
        RestRequest<T> request = JacksonUtil.jsonToBean(s, new TypeReference<>() {
        });
        return request.getReqBody();
    }



    @Override
    public T transform(RestRequest<T> request) {
        return request.getReqBody();
    }



    @Override
    public String name() {
        return "req";
    }

    @Override
    public byte systemCodecID() {
        return -1;
    }
}

RestResponse

公用响应消息类

@Data
public class RestResponse<T> {
    private int code;
    private T data;
    private String msg;
}

RespMessageCodec

公用响应消息编解码类

public class RespMessageCodec<T> implements MessageCodec<RestResponse<T>, RestResponse<T>> {

    @Override
    public void encodeToWire(Buffer buffer, RestResponse<T> response) {
        Buffer encoded = Json.CODEC.toBuffer(response, false);
        buffer.appendInt(encoded.length());
        buffer.appendBuffer(encoded);
    }


    @Override
    public RestResponse<T> decodeFromWire(int pos, Buffer buffer) {
        int length = buffer.getInt(pos);
        pos += 4;
        Buffer slice = buffer.slice(pos, pos + length);
        return JacksonUtil.jsonToBean(slice.toString(), new TypeReference<>() {
        });
    }

    @Override
    public RestResponse<T> transform(RestResponse<T> response) {
        return response;
    }



    @Override
    public String name() {
        return "resp";
    }

    @Override
    public byte systemCodecID() {
        return -1;
    }
}

Producer

public class Producer extends AbstractVerticle {
    @Override
    public void start(Promise<Void> startPromise) throws Exception {
        vertx.setTimer(2000L, h -> {
            System.out.println("***********************");
            User user = new User();
            user.setId("1");
            user.setName("张三");
            RestRequest<User> req = new RestRequest<>();
            req.setReqTopic("addUser");
            req.setReqBody(user);
            vertx.eventBus().<RestResponse>request(req.getReqTopic(), req)
                    .onComplete(res -> {
                        if (res.succeeded()) {
                            System.out.println("生产者收到回应:"+ res.result().body());
                        }else {
                            System.out.println("生产者发送消息失败："+res.cause().getMessage());
                        }
                    });
        });
        super.start(startPromise);
    }
}

Consumer

public class Consumer extends AbstractVerticle {
    @Override
    public void start(Promise<Void> startPromise) throws Exception {
        vertx.eventBus().<User>consumer("addUser", msg -> {
            System.out.println(String.format("消费者接收到消息:%s",msg.body()));
            RestResponse<String> response = new RestResponse();
            response.setCode(200);
            response.setMsg("插入成功");
            msg.reply(response);
        });
        super.start(startPromise);
    }
}

SingleApp

本地传输数据

public class SingleApp {
    public static void main(String[] args) {
        Vertx vertx = Vertx.vertx();
        vertx.eventBus().registerDefaultCodec(RestRequest.class, new ReqMessageCodec())
                        .registerDefaultCodec(RestResponse.class, new RespMessageCodec());
        vertx.deployVerticle(Consumer.class.getName())
                .compose(res -> vertx.deployVerticle(Producer.class.getName()));
    }

}

SingleAppRunLog

***********************
消费者接收到消息:User(id=1, name=张三)
生产者收到回应:RestResponse(code=200, data=null, msg=插入成功)

ClusterApp

集群模式传输数据

public class ClusterApp {
    static Vertx vertx;

    public static void initCluster(Verticle service) {
        ClusterManager mgr = new HazelcastClusterManager();
        VertxOptions options = new VertxOptions().setClusterManager(mgr);
        Vertx.clusteredVertx(options)
                .compose(v -> {
                    vertx = v;
                    vertx.eventBus().registerDefaultCodec(RestRequest.class, new ReqMessageCodec())
                            .registerDefaultCodec(RestResponse.class, new RespMessageCodec());
                    return vertx.deployVerticle(service);
                }).onSuccess(h -> {
                    System.out.println("App Start Complete!");
                })
                .onFailure(err -> {
                    err.printStackTrace();
                    System.err.println("App Start Failed "+ err.getMessage());
                });
    }
}

ConsumerApp

消费服务

public class ConsumerApp extends ClusterApp{
    public static void main(String[] args) {
        initCluster(new Consumer());
    }
}

ConsumerAppRunLog

Members {size:2, ver:2} [
	Member [127.0.0.1]:5703 - 29afc653-a4ef-4933-ab6a-ed0bbbb7041e this
	Member [127.0.0.1]:5704 - 289a8535-cb6f-4fef-8501-04c40c56f189
]
消费者接收到消息:{id=1, name=张三}

ProducerApp

生产服务

public class ProducerApp extends ClusterApp{
    public static void main(String[] args) {
        initCluster(new Producer());
    }
}

ProducerAppRunLog

Members {size:2, ver:2} [
	Member [127.0.0.1]:5703 - 29afc653-a4ef-4933-ab6a-ed0bbbb7041e
	Member [127.0.0.1]:5704 - 289a8535-cb6f-4fef-8501-04c40c56f189 this
]

***********************
生产者收到回应:RestResponse(code=200, data=null, msg=插入成功)

DeliveryOptions

设置请求响应超时时间

public class Producer extends AbstractVerticle {
    @Override
    public void start(Promise<Void> startPromise) throws Exception {
        System.out.println(new Date()+"***********************");
        vertx.eventBus().<String>request("test", "你吃了吗",
                new DeliveryOptions().setSendTimeout(5000))
                .onComplete(res -> {
                    if (res.succeeded()) {
                        System.out.println(new Date()+"生产者接收应答消息："+res.result().body());
                    } else {
                        System.out.println(new Date()+"生产者发送消息失败"+res.cause().getMessage());

                    }
                });
        super.start(startPromise);
    }
}

http://www.kler.cn/a/131824.html

相关文章：

  • 基于java web的中小型人力资源管理系统
  • golang学习笔记——编写 FizzBuzz 程序
  • 最新AI创作系统ChatGPT系统运营源码/支持最新GPT-4-Turbo模型/支持DALL-E3文生图
  • 蓝凌OA sysUiComponent 任意文件上传漏洞复现
  • Android安装过程二 系统进程中PackageInstallerSession对象的创建
  • 利用NVIDIA DALI读取视频帧
  • 在.net 6版本以上的web api中添加像.net 5一样的Startup.cs
  • nacos客户端连接服务端报Client not connected, current status:STARTING
  • NOIP2023模拟19联测40 诡异键盘
  • 953a-954(fpdlink3)寄存器配置
  • 【IDEA】IntelliJ IDEA的使用2.0——结合实际场景提升工具使用
  • SSM项目初始化流程与操作概念解释-SpringBoot简化版
  • Java 类之 java.lang.System
  • 常用Git命令记录
  • Java编程中，异步操作流程中，最终一致性以及重试补偿的设计与实现
  • Nginx在实际使用中的常用基本配置
  • PyTorch 成功安装验证
  • 关于数据mysql -＞maxwell-＞kafka的数据传输
  • 【入门篇】1.2 Redis 客户端之 Jedis 详解和示例
  • 248: vue+openlayers 以静态图片作为底图，并在上面绘制矢量多边形