JAVA | Guava EventBus 使用 发布/订阅模式的步骤

前言

EventBus 是 Guava 的事件处理机制,是观察者模式(生产/消费模型)的一种实现。

观察者模式在我们日常开发中使用非常广泛,例如在订单系统中,订单状态或者物流信息的变更会向用户发送APP推送、短信、通知卖家、买家等等;审批系统中,审批单的流程流转会通知发起审批用户、审批的领导等等。

Observer模式也是 JDK 中自带就支持的,其在 1.0 版本就已经存在 Observer,不过随着 Java 版本的飞速升级,其使用方式一直没有变化,许多程序库提供了更加简单的实现,例如 Guava EventBus、RxJava、EventBus 等

一、为什么要用 Observer模式以及 EventBus 优点 ?

EventBus 优点

  • 相比 Observer 编程简单方便
  • 通过自定义参数可实现同步、异步操作以及异常处理
  • 单进程使用,无网络影响

缺点

  • 只能单进程使用
  • 项目异常重启或者退出不保证消息持久化

如果需要分布式使用还是需要使用 MQ

二、EventBus 使用步骤

1. 引入库

Gradle

compile group: 'com.google.guava', name: 'guava', version: '29.0-jre'

Maven

<dependency>
  <groupId>com.google.guava</groupId>
  <artifactId>guava</artifactId>
  <version>29.0-jre</version>
</dependency>

引入依赖后,这里我们主要使用 com.google.common.eventbus.EventBus 类进行操作,其提供了 registerunregisterpost 来进行注册订阅、取消订阅和发布消息

public void register(Object object);

public void unregister(Object object);

public void post(Object event);

2. 同步使用

1. 首先创建一个 EventBus

EventBus eventBus = new EventBus();

2. 创建一个订阅者

在 Guava EventBus 中,是根据参数类型进行订阅,每个订阅的方法只能由一个参数,同时需要使用 @Subscribe 标识

class EventListener {

 /**
  * 监听 Integer 类型的消息
  */
 @Subscribe
 public void listenInteger(Integer param) {
  System.out.println("EventListener#listenInteger ->" + param);
 }

 /**
  * 监听 String 类型的消息
  */
 @Subscribe
 public void listenString(String param) {
  System.out.println("EventListener#listenString ->" + param);
 }
}

3. 注册到 EventBus 上并发布消息

EventBus eventBus = new EventBus();

eventBus.register(new EventListener());

eventBus.post(1);
eventBus.post(2);
eventBus.post("3");

运行结果为

EventListener#listenInteger ->1
EventListener#listenInteger ->2
EventListener#listenString ->3

根据需要我们可以创建多个订阅者完成订阅信息,同时如果一个类型存在多个订阅者,则所有订阅方法都会执行

为什么说这么做是同步的呢?

Guava Event 实际上是使用线程池来处理订阅消息的,通过源码可以看出,当我们使用默认的构造方法创建 EventBus 的时候,其中 executorMoreExecutors.directExecutor(),其具体实现中直接调用的 Runnable#run 方法,使其仍然在同一个线程中执行,所以默认操作仍然是同步的,这种处理方法也有适用的地方,这样既可以解耦也可以让方法在同一个线程中执行获取同线程中的便利,比如事务的处理

EventBus 部分源码

public class EventBus {
 private static final Logger logger = Logger.getLogger(EventBus.class.getName());
 private final String identifier;
 private final Executor executor;
 private final SubscriberExceptionHandler exceptionHandler;
 private final SubscriberRegistry subscribers;
 private final Dispatcher dispatcher;

 public EventBus() {
  this("default");
 }

 public EventBus(String identifier) {
  this(identifier, MoreExecutors.directExecutor(), Dispatcher.perThreadDispatchQueue(), EventBus.LoggingHandler.INSTANCE);
 }

 public EventBus(SubscriberExceptionHandler exceptionHandler) {
  this("default", MoreExecutors.directExecutor(), Dispatcher.perThreadDispatchQueue(), exceptionHandler);
 }

 EventBus(String identifier, Executor executor, Dispatcher dispatcher, SubscriberExceptionHandler exceptionHandler) {
  this.subscribers = new SubscriberRegistry(this);
  this.identifier = (String)Preconditions.checkNotNull(identifier);
  this.executor = (Executor)Preconditions.checkNotNull(executor);
  this.dispatcher = (Dispatcher)Preconditions.checkNotNull(dispatcher);
  this.exceptionHandler = (SubscriberExceptionHandler)Preconditions.checkNotNull(exceptionHandler);
 }
}

DirectExecutor 部分源码

enum DirectExecutor implements Executor {
 INSTANCE;

 private DirectExecutor() {
 }

 public void execute(Runnable command) {
  command.run();
 }

 public String toString() {
  return "MoreExecutors.directExecutor()";
 }
}

3. 异步使用

通过上面的源码,可以看出只要将构造方法中的 executor 换成一个线程池实现即可, 同时 Guava EventBus 为了简化操作,提供了一个简化的方案即 AsyncEventBus

EventBus eventBus = new AsyncEventBus(Executors.newCachedThreadPool());

这样即可实现异步使用

AsyncEventBus 源码

public class AsyncEventBus extends EventBus {
 public AsyncEventBus(String identifier, Executor executor) {
  super(identifier, executor, Dispatcher.legacyAsync(), LoggingHandler.INSTANCE);
 }

 public AsyncEventBus(Executor executor, SubscriberExceptionHandler subscriberExceptionHandler) {
  super("default", executor, Dispatcher.legacyAsync(), subscriberExceptionHandler);
 }

 public AsyncEventBus(Executor executor) {
  super("default", executor, Dispatcher.legacyAsync(), LoggingHandler.INSTANCE);
 }
}

4. 异常处理

如果处理时发生异常应该如何处理? 在看源码中,无论是 EventBus 还是 AsyncEventBus 都可传入自定义的 SubscriberExceptionHandler 该 handler 当出现异常时会被调用,我可可以从参数 exception 获取异常信息,从 context 中获取消息信息进行特定的处理

其接口声明为

public interface SubscriberExceptionHandler {
 /** Handles exceptions thrown by subscribers. */
 void handleException(Throwable exception, SubscriberExceptionContext context);
}

总结

在上面的基础上,我们可以定义一些消息类型来实现不同消息的监听和处理,通过实现 SubscriberExceptionHandler 来处理异常的情况,无论时同步还是异步都能游刃有余

参考

https://github.com/google/guava
https://github.com/greenrobot/EventBus
https://github.com/ReactiveX/RxJava

以上就是JAVA | Guava EventBus 使用 发布/订阅模式的步骤的详细内容,更多关于Guava EventBus 使用 发布/订阅模式的资料请关注我们其它相关文章!

(0)

相关推荐

  • Java Guava排序器Ordering原理及代码实例

    一 创建排序器 排序器:可以用来为构建复杂的比较器,以完成集合排序的功能: 本质上来说,Ordering 实例无非就是一个特殊的Comparator 实例. Ordering把很多基于Comparator的静态方法(如Collections.max)包装为自己的实例方法(非静态方法), 并且提供了链式调用方法,来定制和增强现有的比较器 //创建排序器 @Test public void createOreing(){ //对可排序类型做自然排序,如数字按大小,日期按先后排序 Ordering<C

  • Guava Cache的使用简介

    1 引入 说到缓存,可能大家最先想到的还是Redis.作为基于键值对的非关系型数据库,Redis具有高性能.丰富的数据结构.持久化.高可用.分布式等特性,使其在业内得到了广泛的认可和使用.但是,使用Redis必然涉及到网络连接,当网络连接不稳定或网络耗时严重时,必然会影响到我们的业务使用.如果我们想提高我们的业务性能,又减少对其他机器的依赖,那么,使用本地缓存会是一个不错的选择. 使用本地缓存时,大多时候我们会采用ConcurrentHashMap来实现.对于本地缓存的使用,现在有一些较为成熟的

  • Java编程guava RateLimiter实例解析

    本文主要研究的是Java编程guava RateLimiter的相关内容,具体如下. 令牌桶算法(token bucket algorithm) 场景1 在流量监管中的应用 约定访问速率(CAR)是流量监管常用技术之一,可以应用在端口进和出方向,一般应用在入方向,它的监管原理如图1所示. a. 按特定的速率向令牌桶投放令牌 b. 根据预设的匹配规则先对报文进行分类,不符合匹配规则的报文不需要经过令牌桶的处理,直接发送: c. 符合匹配规则的报文,则需要令牌桶进行处理.当桶中有足够的令牌则报文可以

  • 详解Guava Cache本地缓存在Spring Boot应用中的实践

    概述 在如今高并发的互联网应用中,缓存的地位举足轻重,对提升程序性能帮助不小.而 3.x开始的 Spring也引入了对 Cache的支持,那对于如今发展得如火如荼的 Spring Boot来说自然也是支持缓存特性的.当然 Spring Boot默认使用的是 SimpleCacheConfiguration,即使用 ConcurrentMapCacheManager 来实现的缓存.但本文将讲述如何将 Guava Cache缓存应用到 Spring Boot应用中. Guava Cache是一个全内

  • java的Guava工具包介绍

    集合 普通集合 List<String> list = Lists.newArrayList(); Set<String> set = Sets.newHashSet(); Map<String, String> map = Maps.newHashMap(); Set 取交集.并集.差集 HashSet<Integer> setA = Sets.newHashSet(1, 2, 3, 4, 5); HashSet<Integer> setB =

  • JAVA | Guava EventBus 使用 发布/订阅模式的步骤

    前言 EventBus 是 Guava 的事件处理机制,是观察者模式(生产/消费模型)的一种实现. 观察者模式在我们日常开发中使用非常广泛,例如在订单系统中,订单状态或者物流信息的变更会向用户发送APP推送.短信.通知卖家.买家等等:审批系统中,审批单的流程流转会通知发起审批用户.审批的领导等等. Observer模式也是 JDK 中自带就支持的,其在 1.0 版本就已经存在 Observer,不过随着 Java 版本的飞速升级,其使用方式一直没有变化,许多程序库提供了更加简单的实现,例如 Gu

  • JS前端设计模式之发布订阅模式详解

    目录 引言 例子1: version1: version2: 总结 引言 昨天我发布了一篇关于策略模式和代理模式的文章,收到的反响还不错,于是今天我们继续来学习前端中常用的设计模式之一:发布-订阅模式. 说到发布订阅模式大家应该都不陌生,它在我们的日常学习和工作中出现的频率简直不要太高,常见的有EventBus.框架里的组件间通信.鉴权业务等等......话不多说,让我们一起进入今天的学习把!!! 发布-订阅模式又叫观察者模式,它定义对象间的一种一对多的依赖关系 当一个对象的状态发生改变时,所有

  • node.js 发布订阅模式的实例

    实例如下: //导入内置模块 let EventEmitter = require('events'); let util=require('util'); //Man继承EventEmitter util.inherits(Man,EventEmitter); //创建一个函数 function Man(){} //实例化函数 let man=new Man(); function findGirl() { console.log('找新的女朋友') } function saveMoney(

  • JavaScript中发布/订阅模式的简单实例

    上次研究观察者模式,很多文章说它也叫Subscribe/Publish(发布/订阅模式).可在<Javascript设计模式>一书中,这两种模式还是有些区别的.书中原话如下: 1.Observer模式要求希望接收到主题通知者的观察者必须订阅内容改变的事件. 2.Subscribe/Publish模式使用了一个主题/事件通道,这个通道介于订阅者和发布者之间.该事件系统允许代码定义应用程序的特定事件,该事件可以传递自定义参数,自定义参数包含订阅者所需要的值.其目的是避免订阅者和发布者产生依赖关系.

  • js 发布订阅模式的实例讲解

    废话不多说,直接上代码 //发布订阅模式 class EventEmiter{ constructor(){ //维护一个对象 this._events={ } } on(eventName,callback){ if( this._events[eventName]){ //如果有就放一个新的 this._events[eventName].push(callback); }else{ //如果没有就创建一个数组 this._events[eventName]=[callback] } } e

  • 浅谈发布订阅模式与观察者模式

    背景 设计模式并非是软件开发的专业术语,实际上,"模式"最早诞生于建筑学. 设计模式的定义是:在面向对象软件设计过程中针对特定问题的简洁而优雅的解决方案.通俗一点说,设计模式是在某种场合下对某个问题的一种解决方案.如果再通俗一点说,设计模式就是给面向对象软件开发中的一些好的设计取个名字. 这些"好的设计"并不是谁发明的,而是早已存在于软件开发中.一个稍有经验的程序员也许在不知不觉中数次使用过这些设计模式.GoF(Gang of Four--四人组,<设计模式&

  • JavaScript实现与使用发布/订阅模式详解

    本文实例讲述了JavaScript实现与使用发布/订阅模式.分享给大家供大家参考,具体如下: 一.发布/订阅模式简介 发布/订阅模式(即观察者模式): 设计该模式背后的主要动力是促进形成松散耦合.在这种模式中,并不是一个对象调用另一个对象的方法,而是一个订阅者对象订阅发布者对象的特定活动,并在发布者对象的状态发生改变后,订阅者对象获得通知.订阅者也称为观察者,而被观察的对象称为发布者或主题.当发生了一个重要的事件时,发布者将会通知(调用)所有订阅者,并且可能经常以事件对象的形式传递消息. 基本思

  • JavaScript事件发布/订阅模式原理与用法分析

    本文实例讲述了JavaScript事件发布/订阅模式原理与用法.分享给大家供大家参考,具体如下: 1.发布/订阅模式也是诸多设计模式当中的一种: 2.这种方式可以在es5下相当优雅地处理异步操作: 3.什么是发布/订阅呢?我们举个栗子: 假设fn1,fn2,fn3都可以视作一个事件的发布者,执行它,就会发布一个事件.这个时候,我们可以通过一个事件的订阅者去批量订阅并处理这些事件,包括它们的先后顺序.增加一个消息订阅者的方法: class AsyncFunArr { constructor (..

  • JavaScript设计模式之观察者模式(发布订阅模式)原理与实现方法示例

    本文实例讲述了JavaScript设计模式之观察者模式(发布订阅模式)原理与实现方法.分享给大家供大家参考,具体如下: 观察者模式,又称为发布订阅模式,它定义了一种一对多的关系,让多个观察者对象同时监听某一个主题对象,这个主题对象的状态发生变化时就会通知所有的观察者对象,使得它们能够自动更新自己的状态. 在观察者模式中,并不是一个对象调用另一个对象的方法,而是一个对象订阅另一个对象的特定活动并在状态改变后获得通知.订阅者也称为观察者,而被观察的对象称为发布者或主题.当发生了一个重要的事件时,发布

  • 发布订阅模式在vue中的实际运用实例详解

    订阅发布模式定义了一种一对多的依赖关系,让多个订阅者对象同时监听某一个主题对象.这个主题对象在自身状态变化时,会通知所有订阅者对象,使它们能够自动更新自己的状态. 比如addEventListener 这个api就是个发布订阅模式 如果用过vue的同学,可以把他类比于 watch 下面我们看一个例子 var observe={ fnsObj:{}, // 订阅方法 on:function(key,fn){ if(!observe.fnsObj[key]){ observe.fnsObj[key]

随机推荐