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

闵行网站建设公司韩国服务器ip地址

news 2025/10/10 12:32:54
闵行网站建设公司,韩国服务器ip地址,所有网站的分辨率,网页设计与制作教案详案简介 EventBus，又称消息总线，类似我们常见的消息中间件。支持点对点、请求与响应、发布订阅模式，支持跨服务跨语言通讯，分布式消息系统。 常见用法 点对点send 消息发送到某个地址上，Vertx把消息分发到注册到这个地…

简介

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

常见用法

点对点send

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

Producer

public class Producer extends AbstractVerticle {@Overridepublic 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 {@Overridepublic 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 {@Overridepublic 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 {@Overridepublic 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 {@Overridepublic 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 {@Overridepublic 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*/@Overridepublic 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*/@Overridepublic 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*/@Overridepublic User transform(User user) {return user;}@Overridepublic String name() {return "user";}@Overridepublic byte systemCodecID() {return -1;}
}

Producer

public class Producer extends AbstractVerticle {@Overridepublic 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 {@Overridepublic 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> {@Overridepublic void encodeToWire(Buffer buffer, RestRequest<T> request) {Buffer encoded = Json.CODEC.toBuffer(request, false);buffer.appendInt(encoded.length());buffer.appendBuffer(encoded);}@Overridepublic 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();}@Overridepublic T transform(RestRequest<T> request) {return request.getReqBody();}@Overridepublic String name() {return "req";}@Overridepublic 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>> {@Overridepublic void encodeToWire(Buffer buffer, RestResponse<T> response) {Buffer encoded = Json.CODEC.toBuffer(response, false);buffer.appendInt(encoded.length());buffer.appendBuffer(encoded);}@Overridepublic 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<>() {});}@Overridepublic RestResponse<T> transform(RestResponse<T> response) {return response;}@Overridepublic String name() {return "resp";}@Overridepublic byte systemCodecID() {return -1;}
}

Producer

public class Producer extends AbstractVerticle {@Overridepublic 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 {@Overridepublic 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 thisMember [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-ed0bbbb7041eMember [127.0.0.1]:5704 - 289a8535-cb6f-4fef-8501-04c40c56f189 this
]***********************
生产者收到回应:RestResponse(code=200, data=null, msg=插入成功)

DeliveryOptions

设置请求响应超时时间

public class Producer extends AbstractVerticle {@Overridepublic 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.dtcms.com/a/462493.html

相关文章：

  • 丝杆模组从结构到应用，有哪些核心类型？
  • 字节面试题：MSE的优化
  • 建设通网站会员共享密码佛山外贸网站制作
  • 哪个素材网站免费免费行情网站app斗印
  • 聚类的数据集
  • ElasticSearch八股
  • 梦中的统计：C++实现与算法分析（洛谷P1554）
  • 鸿蒙9568322问题
  • 破解工地防盗难题：如何利用EasyCVR实现视频监控统一管理？
  • 网站注册协议模板wordpress 调用摘要
  • 电商商拍革命！2025年AI工具实战测评
  • javascript中的三角关系
  • 网站的总体风格包括石家庄 网站开发
  • 【开题答辩全过程】以 宝鸡文化艺术品拍卖系统为例，包含答辩的问题和答案
  • 天猫交易网站宁波网站制作公司费用价格
  • 如何搭建网站本地安装好的wordpress怎么传到服务器上
  • ros2 setup.cfg 各个配置项详细范例
  • Android通用开发——日志常用技术总结
  • 申请网站官网网页版微信和电脑版微信的区别
  • 【2025年清华计算机考研826算法题】
  • 网上网城网站食品经营许可网站增项怎么做
  • 大模型前世今生（九）：词向量汇聚为“大海”
  • 导购网站开发要多少钱个人网站不备案
  • com后缀的网站国际新闻最新消息今天大事件
  • Windows系统下MySQL 8.0详细安装教程
  • 【vLLM】源码解读：高性能大语言模型推理引擎的工程设计与实现
  • 网站seo优化有哪些陕西锦宇建设有限公司网站
  • Looper、MessageQueue、Message及Handler的关系是什么？如何保证MessageQueue的并发访问安全？
  • ELK运维之路(Elasticsearch7集群组建-7.17.24)
  • 网站建设管理工作总结室外平台设计