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

Event Driven设计模式

    EDA(Event-Driven Architecture)是一种实现组件之间松耦合、易扩展的架构方式。一个最简单的EDA设计需要包含如下几个组件:

  • Events:需要被处理的数据。一个Event至少包含两个属性,类型和数据,类型决定了Events被哪个Handler处理,数据是Handler中代加工的材料。

  • Event Handlers:处理Events的方式方法。一般是一些方法操作。

  • Event Loop:维护Events和Event Handlers之间的交互流程。接收所有的Event,然后将其分配给合适的Handler处理。

    Message(Event)无论是在同步还是异步的EDA中,没有使用任何同步方式进行控制,根本原因是Event被设计成了不可改变对象,因为Event在经过每一个Channel(Handler)的时候,都会创建一个全新的Event,多个线程之间不会出现资源竞争,不需要同步的保护。

同步方案代码:

public interface Message {
Class<? extends Message> getType();
}
public interface Channel<E extends Message>{
void dispatch(E message);
}
public interface DynamicRouter<E extends Message> {
void registerChannel(Class<? extends E> messageType,Channel<? extends E> channel);
void dispatch(E message) throws MessageMatcherException;
}
public class Event implements Message{
@Override
public Class<? extends Message> getType() {
return getClass();
}
}
import java.util.HashMap;
import java.util.Map;public class EventDispatcher implements DynamicRouter<Message>{
private final Map<Class<? extends Message>,Channel> routerTable;public EventDispatcher() {
this.routerTable=new HashMap<>();
}@Override
public void registerChannel(Class<? extends Message> messageType, Channel<? extends Message> channel) {
this.routerTable.put(messageType, channel);
}@Override
public void dispatch(Message message) throws MessageMatcherException {
if(this.routerTable.containsKey(message.getType())){
this.routerTable.get(message.getType()).dispatch(message);
}else {
throw new MessageMatcherException("Can't match the channel for ["+message.getType()+"] type");
}
}}
public class MessageMatcherException extends Exception {
public MessageMatcherException(String message) {
super(message);
}
}
public class EventDispatcherExample {static 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;
}
}
static class ResultEvent extends Event{
private final int result;
public ResultEvent(int result) {
this.result=result;
}
public int getResult() {
return result;
}
}static class ResultEventHandler implements Channel<ResultEvent>{
@Override
public void dispatch(ResultEvent message) {
System.out.println("The result is:"+message.getResult());
}
}static class InputEventHandler implements Channel<InputEvent>{
private final EventDispatcher dispacher;
public InputEventHandler(EventDispatcher dispatcher) {
this.dispacher=dispatcher;
}@Override
public void dispatch(InputEvent message){
System.out.printf("X:%d,Y:%d\n",message.getX(),message.getY());
int result=message.getX()+message.getY();
try {
this.dispacher.dispatch(new ResultEvent(result));
} catch (MessageMatcherException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
}
}public static void main(String[] args) {
EventDispatcher dispatcher=new EventDispatcher();
dispatcher.registerChannel(InputEvent.class, new InputEventHandler(dispatcher));
dispatcher.registerChannel(ResultEvent.class, new ResultEventHandler());
try {
dispatcher.dispatch(new InputEvent(1,2));
} catch (MessageMatcherException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
}}

异步方案代码是在共用了部分同步代码之后形成的:

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;public abstract class AsyncChannel implements Channel<Event>{
private final ExecutorService executorService;public AsyncChannel(ExecutorService executorService) {
this.executorService=executorService;
}public AsyncChannel() {
this(Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors()*2));
}public final void dispatch(Event message) {
this.executorService.submit(()->this.handle(message));
}protected abstract void handle(Event message);void stop() {
if(null!=this.executorService&&!this.executorService.isShutdown()) {
this.executorService.shutdown();
}
}}
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;public class AsyncEventDispatcher implements DynamicRouter<Event>{
private final Map<Class<? extends Event>,AsyncChannel> routerTable;public AsyncEventDispatcher() {
this.routerTable=new ConcurrentHashMap<>();
}@Override
public void registerChannel(Class<? extends Event> messageType, Channel<? extends Event> channel) {
if(!(channel instanceof AsyncChannel)) {
throw new IllegalArgumentException("The channel must be AsyncChannel type.");
}
this.routerTable.put(messageType, (AsyncChannel)channel);
}@Override
public void dispatch(Event message) throws MessageMatcherException {
if(this.routerTable.containsKey(message.getType())) {
this.routerTable.get(message.getType()).dispatch(message);
}else {
throw new MessageMatcherException("Can't match the channel for ["+message.getType()+"] type");
}
}public void shutdown() {
this.routerTable.values().forEach(AsyncChannel::stop);
}
}
import java.util.concurrent.TimeUnit;public class AsyncEventDispatcherExample {static class AsyncInputEventHandler extends AsyncChannel{
private final AsyncEventDispatcher dispatcher;
AsyncInputEventHandler(AsyncEventDispatcher dispatcher){
this.dispatcher=dispatcher;
}
@Override
protected void handle(Event message) {
EventDispatcherExample.InputEvent inputEvent=(EventDispatcherExample.InputEvent) message;
System.out.printf("X:%d,Y:%d\n",inputEvent.getX(),inputEvent.getY());
try {
TimeUnit.SECONDS.sleep(10);
} catch (InterruptedException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
int result = inputEvent.getX()+inputEvent.getY();
try {
this.dispatcher.dispatch(new EventDispatcherExample.ResultEvent(result));
} catch (MessageMatcherException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
}
}static class AsyncResultEventHandler extends AsyncChannel{
@Override
protected void handle(Event message) {
EventDispatcherExample.ResultEvent resultEvent=(EventDispatcherExample.ResultEvent) message;
try {
TimeUnit.SECONDS.sleep(10);
} catch (InterruptedException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
System.out.println("the result is:"+resultEvent.getResult());
}
}public static void main(String[] args) {
AsyncEventDispatcher dispatcher=new AsyncEventDispatcher();
dispatcher.registerChannel(EventDispatcherExample.InputEvent.class, new AsyncInputEventHandler(dispatcher));
dispatcher.registerChannel(EventDispatcherExample.ResultEvent.class, new AsyncResultEventHandler());
try {
dispatcher.dispatch(new EventDispatcherExample.InputEvent(1, 2));
} catch (MessageMatcherException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
}}

相关文章:

Event Driven设计模式

EDA&#xff08;Event-Driven Architecture&#xff09;是一种实现组件之间松耦合、易扩展的架构方式。一个最简单的EDA设计需要包含如下几个组件&#xff1a; Events&#xff1a;需要被处理的数据。一个Event至少包含两个属性&#xff0c;类型和数据&#xff0c;类型决定了Eve…...

PostgreSql 设置自增字段

一、概述 序列类型是 PostgreSQL 特有的创建一个自增列的方法。包含 smallserial、serial和 bigserial 类型&#xff0c;它们不是真正的类型&#xff0c;只是为了创建唯一标识符列而存在的方便符号。其本质也是调用的序列&#xff0c;序列详情可参考&#xff1a;《PostgreSql 序…...

什么是泊松图像混合

泊松图像混合&#xff08;Poisson Image Editing&#xff09;的原理基于泊松方程。该方法旨在保持图像中的梯度一致性&#xff0c;从而在图像编辑中实现平滑和无缝的混合。以下是泊松图像混合的基本原理和公式&#xff1a; 泊松方程 泊松方程是一个偏微分方程&#xff0c;通常…...

OpenAI 承认 ChatGPT 最近确实变懒,承诺修复问题

文章目录 一. ChatGPT 指令遵循能力下降引发用户投诉1.1 用户抱怨回应速度慢、敷衍回答、拒绝回答和中断会话 二. OpenAI 官方确认 ChatGPT 存在问题&#xff0c;展开调查三. OpenAI 解释模型行为差异&#xff0c;回应用户质疑四. GPT-4 模型变更受人事动荡和延期影响 一. Chat…...

创作活动(四十九)———低代码:美味膳食或垃圾食品?

#低代码&#xff1a;美味膳食或垃圾食品&#xff1f;# 一、什么是低代码 低代码是一种开发方法&#xff0c;通过可视化界面和少量的编码&#xff0c;使开发者能够快速构建应用程序。它的目标是提高开发效率、降低开发成本&#xff0c;并支持快速迭代和敏捷开发。 二、低代码的…...

【DL-TensorFlow遇错】TensorFlow中遇错合集

TensorFlow中遇错合集 一、AttributeError: module tensorflow has no attribute placeholder二、RuntimeError: tf.placeholder() is not compatible with eager execution. 一、AttributeError: module tensorflow has no attribute placeholder 错误原因 tensorflow版本问…...

pymysql代替mysqlclient,解决mysqlclient因版本不兼容无法安装成功而无法连接mysql的问题

pymysql代替mysqlclient&#xff0c;解决mysqlclient因版本不兼容无法安装成功而无法连接mysql的问题 原因&#xff1a;版本或者环境兼容问题&#xff0c;导致如centos或者其他Linux无法安装mysqlclient模块 解决办法&#xff1a;安装pymysql作为替代 在Django中连接MySQL数…...

uni-app 设置当前page界面进入直接变为横屏模式

首先 我们打开项目的 manifest.json 在左侧导航栏中找到 源码视图 然后找到 app-plus 配置 在下面加上 "orientation": [//竖屏正方向"portrait-primary",//竖屏反方向"portrait-secondary",//横屏正方向"landscape-primary",//横屏…...

Mysql的多表联合查询

内连接 隐式内连接 select column from tb1,tb2 where 条件; 显示内连接 关键字&#xff1a;[inner] join on 显示内连接与外连接的不同是新增的关键字&#xff0c;inner join 以及 使用on 替换了where select column from tb1 [inner] join tb2 on 条件; 外连接 左外…...

Linux上使用Python的requests库进行HTTP请求

在Linux上使用Python的requests库进行HTTP请求是一种非常方便和高效的方式。requests库是一个第三方库&#xff0c;用于发送HTTP请求并获取响应。下面是一个简单的示例&#xff0c;演示如何使用requests库发送GET请求并获取响应。 首先&#xff0c;你需要安装requests库。你可…...

图像处理领域的应用

图像处理领域的应用 文章目录 图像处理领域的应用1.图像类型2.图像转换3.彩色图像表示模式4.图像变换5.图像增强 1.图像类型 #mermaid-svg-x6mNS3Y1YkPvWUsQ {font-family:"trebuchet ms",verdana,arial,sans-serif;font-size:16px;fill:#333;}#mermaid-svg-x6mNS3Y1…...

MySQL笔记-第18章_MySQL8其它新特性

视频链接&#xff1a;【MySQL数据库入门到大牛&#xff0c;mysql安装到优化&#xff0c;百科全书级&#xff0c;全网天花板】 文章目录 第18章_MySQL8其它新特性1. MySQL8新特性概述1.1 MySQL8.0 新增特性1.2 MySQL8.0移除的旧特性 2. 新特性1&#xff1a;窗口函数2.1 使用窗口…...

C语言—每日选择题—Day46

第一题 1. 下列程序段的输出结果是&#xff08;&#xff09; #include <stdio.h> int main() {int x 1,a 0,b 0;switch(x) {case 0: b;case 1: a;case 2: a;b;}printf("a%d,b%d\n", a, b);return 0; } A&#xff1a;a2,b1 B&#xff1a;a1,b1 C&#xf…...

flex布局,换行的元素上下设置间距

要生成的效果图如下&#xff1a; display:flexflex-direction: row;flex-wrap: wrap;当我们使用弹性盒子布局后&#xff0c;默认元素是没有外边距的&#xff0c;紧挨着样式就有点丑&#xff0c;如果想使换行后&#xff0c;元素的外边距有个距离&#xff0c;可以用如下方法解决…...

【智能家居】八、监控摄像采集、人脸识别比对进行开门功能点

一、使用 fswebcam 测试 USB 摄像头 二、根据demo来实现功能点 三、功能点编写编译运行实现 四、mjpg实现监控识别 五、V4L2 视频设备 Linux 内核模块的一部分 一、使用 fswebcam 测试 USB 摄像头 a. 安装 fswebcam orangepiorangepi:~$ sudo apt update orangepiorangepi:~…...

golang的文件操作

获取文件列表路径 package _caseimport ("fmt""log""os""strings" )// 获取文件路径 // 源文件目录 const sourceDir "file/"// 目标文件目录 const destDir "det_file/"// 拿到目录下完整的路径 func geFiles…...

数据库版本管理框架-Flyway(从入门到精通)

一、flyway简介 Flyway是一个简单开源数据库版本控制器&#xff08;约定大于配置&#xff09;&#xff0c;主要提供migrate、clean、info、validate、baseline、repair等命令。它支持SQL&#xff08;PL/SQL、T-SQL&#xff09;方式和Java方式&#xff0c;支持命令行客户端等&am…...

外网访问内网服务器使用教程

如何在任何地方都能访问自己家里的笔记本上的应用&#xff1f;如何让局域网的服务器可以被任何地方访问到&#xff1f;有很多类似的需求&#xff0c;我们可以统一用一个解决方案&#xff1a;内网穿透。内网穿透的工具及方式有很多&#xff0c;如Ngrok、Ssh、autossh、Natapp、F…...

C# Dictionary 利用 ContainsValue 查询指定值是否已经存在

.NET Framework : 4.7.2IDE : Visual Studio Community 2022OS : Windows 10 x64typesetting : Markdownblog : niaoge.blog.csdn.net 简介 本文介绍如何查询Dictionary 中某个值是否已经存在。 ContainsValue 命名空间: System.Collections.Generic 程序集: System.Collect…...

招不到人?用C语言采集系统批量采集简历

虽说现在大环境不太好&#xff0c;很多人面临着失业再就业风险&#xff0c;包括企业则面临着招人人&#xff0c;找对口专业难得问题。想要找到适合自己公司的人员&#xff0c;还要得通过爬虫获取筛选简历才能从茫茫人海中找到公司得力干将。废话不多说&#xff0c;直接开整。 1…...

React Native 开发环境搭建(全平台详解)

React Native 开发环境搭建&#xff08;全平台详解&#xff09; 在开始使用 React Native 开发移动应用之前&#xff0c;正确设置开发环境是至关重要的一步。本文将为你提供一份全面的指南&#xff0c;涵盖 macOS 和 Windows 平台的配置步骤&#xff0c;如何在 Android 和 iOS…...

遍历 Map 类型集合的方法汇总

1 方法一 先用方法 keySet() 获取集合中的所有键。再通过 gey(key) 方法用对应键获取值 import java.util.HashMap; import java.util.Set;public class Test {public static void main(String[] args) {HashMap hashMap new HashMap();hashMap.put("语文",99);has…...

2025 后端自学UNIAPP【项目实战:旅游项目】6、我的收藏页面

代码框架视图 1、先添加一个获取收藏景点的列表请求 【在文件my_api.js文件中添加】 // 引入公共的请求封装 import http from ./my_http.js// 登录接口&#xff08;适配服务端返回 Token&#xff09; export const login async (code, avatar) > {const res await http…...

【android bluetooth 框架分析 04】【bt-framework 层详解 1】【BluetoothProperties介绍】

1. BluetoothProperties介绍 libsysprop/srcs/android/sysprop/BluetoothProperties.sysprop BluetoothProperties.sysprop 是 Android AOSP 中的一种 系统属性定义文件&#xff08;System Property Definition File&#xff09;&#xff0c;用于声明和管理 Bluetooth 模块相…...

ETLCloud可能遇到的问题有哪些?常见坑位解析

数据集成平台ETLCloud&#xff0c;主要用于支持数据的抽取&#xff08;Extract&#xff09;、转换&#xff08;Transform&#xff09;和加载&#xff08;Load&#xff09;过程。提供了一个简洁直观的界面&#xff0c;以便用户可以在不同的数据源之间轻松地进行数据迁移和转换。…...

代理篇12|深入理解 Vite中的Proxy接口代理配置

在前端开发中,常常会遇到 跨域请求接口 的情况。为了解决这个问题,Vite 和 Webpack 都提供了 proxy 代理功能,用于将本地开发请求转发到后端服务器。 什么是代理(proxy)? 代理是在开发过程中,前端项目通过开发服务器,将指定的请求“转发”到真实的后端服务器,从而绕…...

推荐 github 项目:GeminiImageApp(图片生成方向,可以做一定的素材)

推荐 github 项目:GeminiImageApp(图片生成方向&#xff0c;可以做一定的素材) 这个项目能干嘛? 使用 gemini 2.0 的 api 和 google 其他的 api 来做衍生处理 简化和优化了文生图和图生图的行为(我的最主要) 并且有一些目标检测和切割(我用不到) 视频和 imagefx 因为没 a…...

处理vxe-table 表尾数据是单独一个接口,表格tableData数据更新后,需要点击两下,表尾才是正确的

修改bug思路&#xff1a; 分别把 tabledata 和 表尾相关数据 console.log() 发现 更新数据先后顺序不对 settimeout延迟查询表格接口 ——测试可行 升级↑&#xff1a;async await 等接口返回后再开始下一个接口查询 ________________________________________________________…...

SQL慢可能是触发了ring buffer

简介 最近在进行 postgresql 性能排查的时候,发现 PG 在某一个时间并行执行的 SQL 变得特别慢。最后通过监控监观察到并行发起得时间 buffers_alloc 就急速上升,且低水位伴随在整个慢 SQL,一直是 buferIO 的等待事件,此时也没有其他会话的争抢。SQL 虽然不是高效 SQL ,但…...

【C++特殊工具与技术】优化内存分配(一):C++中的内存分配

目录 一、C 内存的基本概念​ 1.1 内存的物理与逻辑结构​ 1.2 C 程序的内存区域划分​ 二、栈内存分配​ 2.1 栈内存的特点​ 2.2 栈内存分配示例​ 三、堆内存分配​ 3.1 new和delete操作符​ 4.2 内存泄漏与悬空指针问题​ 4.3 new和delete的重载​ 四、智能指针…...