1. 需求

基于Event-Driven架构设计实现如下需求
* 1. 输入2个数 x 和 y, 并进行相加
* 2. 对1中的结果进行格式化
* 3. 将2中格式化的结果输出（打印）

2. 设计

使用Event-Driven架构进行设计，对于每一种消息类型分别定义 Event、Channel，进行差异化处理；

3. 代码逻辑

接口Message，每种消息实体实现Message接口（此处为Event）；
对于Event这种类型消息，定义子类 InputEvent、FormatEvent、ResultEvent对应于不同的阶段的实体；
定义任务转发路由（Router）
对于上述3中实体，分别对应不同的处理器（Channel）；
注册处理器

4. 代码

4.1 消息实体

package com.hz.design.eventdriven.message;/*** @description: 消息接口* @author: pp_lan* @date: 2022/3/25*/
public interface Message {/*** 返回消息类型* @return*/Class<? extends Message> getType();}

package com.hz.design.eventdriven.message;/*** @description: Event类型的消息* @author: pp_lan* @date: 2022/3/25*/
public class Event implements Message {@Overridepublic Class<? extends Message> getType() {return getClass();}
}

package com.hz.design.eventdriven.message;/*** @description:* @author: pp_lan* @date: 2022/3/25*/
public class InputEvent extends Event {private final int x;private final int y;public InputEvent(int x, int y) {this.x = x;this.y = y;}public int getX() {return x;}public int getY() {return y;}
}

package com.hz.design.eventdriven.message;/*** @description: 格式化事件* @author: pp_lan* @date: 2022/3/25*/
public class FormatEvent extends Event {private int result;public FormatEvent(int result) {this.result = result;}public int getResult() {return result;}
}

package com.hz.design.eventdriven.message;/*** @description:* @author: pp_lan* @date: 2022/3/25*/
public class ResultEvent extends Event {private final String result;public ResultEvent(String result) {this.result = result;}public String getResult() {return result;}
}

4.2 任务转发Router

package com.hz.design.eventdriven.router;import com.hz.design.eventdriven.channel.Channel;
import com.hz.design.eventdriven.message.Message;/*** @description: 动态路由* @author: pp_lan* @date: 2022/3/25*/
public interface DynamicRouter<E extends Message> {/*** 每种类型的消息注册一个通道** @param messageType* @param channel*/void registerChannel(Class<? extends E> messageType, Channel<? extends E> channel);/*** 调度：为Channel分配Message** @param message*/void dispatch(E message);}

package com.hz.design.eventdriven.router;import com.hz.design.eventdriven.channel.Channel;
import com.hz.design.eventdriven.message.Event;
import com.hz.design.eventdriven.message.Message;import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;/*** @description: 调度器* @author: pp_lan* @date: 2022/3/25*/
public class EventDispatcher implements DynamicRouter<Event> {private final Map<Class<? extends Message>, Channel> routerTable;public EventDispatcher() {this.routerTable = new ConcurrentHashMap<>();}@Overridepublic void registerChannel(Class<? extends Event> messageType, Channel<? extends Event> channel) {this.routerTable.put(messageType, channel);}/*** 分发： 根据Event的类型获取对应的处理器（Channel）。如: InputEvent--> InputEventHandler, resultEvent --> ResultEventHandler* @param message*/@Overridepublic void dispatch(Event message) {if (this.routerTable.containsKey(message.getType())) {System.out.format("事件转发: %s\n", message.getType().getSimpleName());routerTable.get(message.getType()).dispatch(message);} else {throw new RuntimeException(String.format("该类型的消息未注册:%s", message.getType()));}}
}

4.3 处理器Channel

package com.hz.design.eventdriven.channel;import com.hz.design.eventdriven.message.Message;/*** @description: 通道* @author: pp_lan* @date: 2022/3/25*/
public interface Channel<E extends Message> {/*** 调度消息** @param message*/void dispatch(E message);
}

package com.hz.design.eventdriven.channel;import com.hz.design.eventdriven.message.FormatEvent;
import com.hz.design.eventdriven.message.InputEvent;
import com.hz.design.eventdriven.message.ResultEvent;
import com.hz.design.eventdriven.router.EventDispatcher;/*** @description:* @author: pp_lan* @date: 2022/3/25*/
public class InputEventHandler implements Channel<InputEvent> {private final EventDispatcher resultDispatcher;public InputEventHandler(EventDispatcher resultDispatcher) {this.resultDispatcher = resultDispatcher;}@Overridepublic void dispatch(InputEvent message) {// 输入后进行计算int result = message.getX() + message.getY();System.out.printf("input结果：%d\n", result);// 将消息交给结果处理器resultDispatcher.dispatch(new FormatEvent(result));}
}

package com.hz.design.eventdriven.channel;import com.hz.design.eventdriven.message.Event;
import com.hz.design.eventdriven.message.FormatEvent;
import com.hz.design.eventdriven.message.ResultEvent;
import com.hz.design.eventdriven.router.EventDispatcher;/*** @description:* @author: pp_lan* @date: 2022/3/25*/
public class FormatEventHandler implements Channel<FormatEvent> {private final EventDispatcher dispatcher;public FormatEventHandler(EventDispatcher dispatcher) {this.dispatcher = dispatcher;}@Overridepublic void dispatch(FormatEvent message) {int result = message.getResult();String resultFormat = String.format("【格式化】标准输出结果为: %d", result);System.out.printf("format结果：%s\n", resultFormat);this.dispatcher.dispatch(new ResultEvent(resultFormat));}
}

package com.hz.design.eventdriven.channel;import com.hz.design.eventdriven.message.ResultEvent;/*** @description:* @author: pp_lan* @date: 2022/3/25*/
public class ResultEventHandler implements Channel<ResultEvent> {@Overridepublic void dispatch(ResultEvent message) {System.out.format("result结果: %s\n", message.getResult());}
}

4.4 测试类

package com.hz.design.eventdriven;import com.hz.design.eventdriven.channel.FormatEventHandler;
import com.hz.design.eventdriven.channel.InputEventHandler;
import com.hz.design.eventdriven.channel.ResultEventHandler;
import com.hz.design.eventdriven.message.FormatEvent;
import com.hz.design.eventdriven.message.InputEvent;
import com.hz.design.eventdriven.message.ResultEvent;
import com.hz.design.eventdriven.router.EventDispatcher;/*** @description: 基于Event-Driven架构设计实现如下需求*  1. 输入2个数 x 和 y, 并进行相加*  2. 对1中的结果进行格式化*  3. 将2中格式化的结果输出（打印）* @author: pp_lan* @date: 2022/3/25*/
public class EventDispatcherTest {public static void main(String[] args) {EventDispatcher dispatcher = new EventDispatcher();// 注册事件（Event）对应的处理器： InputEvent --> InputEventHandler, ResultEvent --> ResultEventHandlerdispatcher.registerChannel(InputEvent.class, new InputEventHandler(dispatcher));// 格式化数据事件dispatcher.registerChannel(FormatEvent.class, new FormatEventHandler(dispatcher));// ResultEventHandler为最后一个结果，不再需要dispatcher进行转发了dispatcher.registerChannel(ResultEvent.class, new ResultEventHandler());dispatcher.dispatch(new InputEvent(2, 5));}
}

5. 执行结果

Connected to the target VM, address: '127.0.0.1:62897', transport: 'socket'
事件转发: InputEvent
input结果：7
事件转发: FormatEvent
format结果：【格式化】标准输出结果为: 7
事件转发: ResultEvent
result结果: 【格式化】标准输出结果为: 7
Disconnected from the target VM, address: '127.0.0.1:62897', transport: 'socket'

参考：《JAVA高并发编程详解》-汪文君

Event-Driven架构设计消息处理系统相关推荐

架构设计：系统间通信（36）——Apache Camel快速入门（上）
1.本专题主旨 1-1.关于技术组件在这个专题中,我们介绍了相当数量技术组件:Flume.Kafka.ActiveMQ.Rabbitmq.Zookeeper.Thrift .Netty.DUBBO等 ...
架构设计：系统存储（16）——Redis事件订阅和持久化存储
接上文<架构设计:系统存储(15)--Redis基本概念和安装使用> 3-4.事件功能和配置项 Redis从2.X版本开始,就支持一种基于非持久化消息的.使用发布/订阅模式实现的事件通知机 ...
架构设计：系统间通信（20）——MQ：消息协议（下）
(接上文<架构设计:系统间通信(19)--MQ:消息协议(上)>) 上篇文章中我们重点讨论了"协议"的重要性,并为各位读者介绍了Stomp协议和XMPP协议.这两种协议 ...
架构设计：系统间通信（31）——其他消息中间件及场景应用（下1）
接上文:<架构设计:系统间通信(30)--Kafka及场景应用(中3)> 5.场景应用--电商平台:浏览记录收集功能事件/日志收集系统是大中型软件不得不面对的话题.目前第三方业务系统对 ...
微服务架构案例(02)：业务架构设计，系统分层管理
本文源码:GitHub·点这里 || GitEE·点这里更新进度(共6节): 01:项目技术选型简介,架构图解说明 02:业务架构设计,系统分层管理一.业务架构设计 1.基础概念服务的架构设计决 ...
架构设计：系统存储（21）——图片服务器：详细设计（1）
(接上文<架构设计:系统存储(20)--图片服务器:需求和技术选型(2)>) 1.概述之前的两篇文章介绍了图片系统的技术组件选型和技术方案设计,从这篇文章开始我们将搭建工程进行详细的编码 ...
架构设计：系统存储（2）——块存储方案（2）
接上文<架构设计:系统存储(1)--块存储方案(1)> 4-2.固态硬盘工作过程本小节我们要解决一个关键问题:既然机械硬盘和固态硬盘从工作原理.制作工艺.技术规范等多个方面都完全不一样, ...
架构设计：系统间通信（10）——RPC的基本概念
1.概述经过了详细的信息格式.网络IO模型的讲解,并且通过JAVA RMI的讲解进行了预热.从这篇文章开始我们将进入这个系列博文的另一个重点知识体系的讲解:RPC.在后续的几篇文章中,我们首先讲解R ...
架构设计：系统存储（20）——图片服务器：需求和技术选型（2）
(接上文<架构设计:系统存储(19)--图片服务器:需求和技术选型(1)>) 3-5.其它技术选型说明 3-5-1.关于关系型数据库关于持久化存储的数据库技术要注意一点,实际上它并不是图 ...
架构设计：系统间通信（4）——IO通信模型和JAVA实践中篇
接上篇<架构设计:系统间通信(3)--IO通信模型和JAVA实践上篇> 4.多路复用IO模型在"上篇"文章中,我们已经提到了使用多线程解决高并发场景的问题所在,这篇 ...

Event-Driven架构设计消息处理系统

1. 需求

2. 设计

3. 代码逻辑

4. 代码

4.1 消息实体

4.2 任务转发Router

4.3 处理器Channel

4.4 测试类

5. 执行结果

Event-Driven架构设计消息处理系统相关推荐

最新文章

热门文章