Java事件总线编程初探

简介:

Why?

在平时写代码的过程中,我们需要实现这样一种功能:当执行某个逻辑时,希望能够进行其他逻辑的处理。最粗暴的方法是直接依赖其他模块,调用该模块的相应函数或者方法。但是,这样做带来一些问题。

  • 模块间相互依赖,耦合度高。以下订单为例,订单提交后需要进行支付以及进行一些其他处理,如发邮件等操作。相关的代码可能是这样。可以看到:订单模块依赖了支付服务以及用户服务。
  • 维护困难。由于模块间相互依赖,当需要修改订单逻辑时则需要修改submitOrder方法的源代码,而某些时候可能无法修改。再者,如果有多个这种逻辑,修改时可能涉及到多处操作。
public class OrderPage {
 
        private PaymentService paymentService;
        private UserService userService;
         
        public void submitOrder() {
                Integer userId = 1;
                BigDecimal amount = BigDecimal.TEN;
                 
                paymentService.doPayment(userId, amount);
                userService.registerPayment(userId, amount);
        }
}
public class PaymentService {
 
        private MailService mailService;
         
        public void doPayment(Integer userId, BigDecimal amount) {
                //Do payment...
                mailService.sendPaymentEmail(userId, amount);
        }
}
public class UserService {
 
        public String getEmailAddress(Integer userId) {
                return "foo@bar.com";
        }
         
        public void registerPayment(Integer userId, BigDecimal amount) {
                //Register payment in database...
        }
}
 
public class MailService {
 
        private UserService userService;
         
        public void sendPaymentEmail(Integer userId, BigDecimal amount) {
                String emailAddress = userService.getEmailAddress(userId);
                //Send email...
        }
}

观察者模式

有时被称作发布/订阅模式,观察者模式定义了一种一对多的依赖关系,让多个观察者对象同时监听某一个主题对象。这个主题对象在状态发生变化时,会通知所有观察者对象,使它们能够自动更新自己。

通过观察者模式来进行解耦,当对象发生变化时,通知其观察者,由观察者进行相应的处理。体现在订单逻辑中时即为,定义多个观察者观察下订单这个主题,当下订单的动作发生时,通知其所有观察者。再由每个观察者进行处理。依据观察者模式的实现,以上逻辑可改为如下代码:

interface OrderListener {

	public void onSubmitOrder(Integer userId, BigDecimal amount);
}


public class OrderPage {

	private List<OrderListener> orderListeners=new ArrayList<OrderListener>();

	public void submitOrder() {
		Integer userId = 1;
		BigDecimal amount = BigDecimal.TEN;

		for (OrderListener orderListener : orderListeners) {
			orderListener.onSubmitOrder(userId, amount);
		}
	}
	
	public void addOrderListener(OrderListener orderListener){
		this.orderListeners.add(orderListener);
	}
}

class PaymentService implements OrderListener {

	private MailService mailService;

	public void doPayment(Integer userId, BigDecimal amount) {
		// Do payment...
		mailService.sendPaymentEmail(userId, amount);
	}

	@Override
	public void onSubmitOrder(Integer userId, BigDecimal amount) {

		doPayment(userId, amount);
	}
}

class UserService implements OrderListener {

	public String getEmailAddress(Integer userId) {
		return "foo@bar.com";
	}

	public void registerPayment(Integer userId, BigDecimal amount) {
		// Register payment in database...
	}

	@Override
	public void onSubmitOrder(Integer userId, BigDecimal amount) {

		registerPayment(userId, amount);

	}
}

class MailService {

	private UserService userService;

	public void sendPaymentEmail(Integer userId, BigDecimal amount) {
		String emailAddress = userService.getEmailAddress(userId);
		// Send email...
	}
}

可以看到,首先定义了OrderListener接口,接口中有一个onSubmitOrder方法。原始的实现中的PayService和UserService实现了该接口。OrderPage中维护了一个OrderListener列表,当提交订单时调用所有监听者的onSubmitOrder方法。可以看到此实现的订单逻辑没有直接依赖付款模块和用户模块。 主程序通过添加监听器来使其得到通知.

 public static void main(String[] args) {
    	PaymentService paymentService=new PaymentService();
    	UserService userService=new UserService();
    	
    	OrderPage orderPage=new OrderPage();
    	
    	orderPage.addOrderListener(paymentService);
    	orderPage.addOrderListener(userService);
	}

Guava EventBus——监听者模式的优雅实现

虽然监听者模式对源代码进行了解耦,但是还是有一些不足。

  • 相关模块需要实现相应接口;
  • 需要主动调用相关的addListener方法设置监听器。
  • 一个监听器智能监听一种操作.

EventBus是Guava对于监听者模式的实现,其使用非常简单。使用EventBus来实现监听者模式,只需要三步操作。

  1. 通过注解@Subscribe来声明事件回调方法;
  2. 调用EventBus的register方法来注册监听器;
  3. 通过post方法来触发事件;

订单逻辑通过EventBus事件总线来实现,大概是以下这个样子。

public class OrderPage {

	public static EventBus eventBus = new EventBus();

	public void submitOrder() {
		Integer userId = 1;
		BigDecimal amount = BigDecimal.TEN;

		eventBus.post(new PayEvent(userId, amount));
	}

}

class PaymentService {

	private MailService mailService;

	@Subscribe
	public void doPayment(PayEvent  payEvent) {
		// Do payment...
		mailService.sendPaymentEmail(payEvent.getUserId(), payEvent.getAmount());
	}

}

class UserService {

	public String getEmailAddress(Integer userId) {
		return "foo@bar.com";
	}

	@Subscribe
	public void registerPayment(PayEvent payEvent) {
		// Register payment in database...
	}
}

class PayEvent {

	private Integer userId;
	private BigDecimal amount;
	
	public PayEvent(Integer userId, BigDecimal amount) {
	}
	
	public Integer getUserId() {
		return userId;
	}
	public BigDecimal getAmount() {
		return amount;
	}
}

 public static void main(String[] args) {
    	PaymentService paymentService=new PaymentService();
    	UserService userService=new UserService();
    	
    	OrderPage orderPage=new OrderPage();
    	
    	orderPage.eventBus.register(paymentService);
    	orderPage.eventBus.register(userService);
	}

要实现监听者模式,时需要调用eventBus的register方法进行注册,在需要处理事件的方法上使用@Subscribe注解。最后通过eventBus发布事件即可。使用事件总线,不需要定义特定的接口,不需要主动添加监听器;

事件订阅

EventBus通过register方法来注册处理相应事件的类,

public void register(Object object) {
    Multimap<Class<?>, EventSubscriber> methodsInListener =
        finder.findAllSubscribers(object);
    subscribersByTypeLock.writeLock().lock();
    try {
      subscribersByType.putAll(methodsInListener);
    } finally {
      subscribersByTypeLock.writeLock().unlock();
    }
  }

其核心是findAllSubscribers,找到实例中所有有Subscribe注解的方法并保存。返回的是一个Multimap < Class<?>,EventSubscriber>类型,其中Class是事件类型,EventSubsciber包含了类实例和具体处理事件的方法。Multimap保证了一种事件可以有多个监听者来处理。

public Multimap<Class<?>, EventSubscriber> findAllSubscribers(Object listener) {
    Multimap<Class<?>, EventSubscriber> methodsInListener = HashMultimap.create();
    Class<?> clazz = listener.getClass();
    for (Method method : getAnnotatedMethods(clazz)) {
      Class<?>[] parameterTypes = method.getParameterTypes();
      Class<?> eventType = parameterTypes[0];
      EventSubscriber subscriber = makeSubscriber(listener, method);
      methodsInListener.put(eventType, subscriber);
    }
    return methodsInListener;
  }

发布事件

EventBus通过post方法来发布事件,首先通过事件类型找到需要处理的事件:事件本身以及其父类。根据事件类型从事件订阅的缓存中取出处理该事件的订阅者,并将其入队。最后处理该队列中的数据。

public void post(Object event) {
    Set<Class<?>> dispatchTypes = flattenHierarchy(event.getClass());

    boolean dispatched = false;
    for (Class<?> eventType : dispatchTypes) {
      subscribersByTypeLock.readLock().lock();
      try {
        Set<EventSubscriber> wrappers = subscribersByType.get(eventType);

        if (!wrappers.isEmpty()) {
          dispatched = true;
          for (EventSubscriber wrapper : wrappers) {
            enqueueEvent(event, wrapper);
          }
        }
      } finally {
        subscribersByTypeLock.readLock().unlock();
      }
    }

    if (!dispatched && !(event instanceof DeadEvent)) {
      post(new DeadEvent(this, event));
    }

    dispatchQueuedEvents();
  }
目录
相关文章
|
14天前
|
安全 Java 开发者
深入理解Java并发编程:线程安全与性能优化
【4月更文挑战第9天】本文将深入探讨Java并发编程的核心概念,包括线程安全和性能优化。我们将详细解析Java中的同步机制,包括synchronized关键字、Lock接口以及并发集合等,并探讨它们如何影响程序的性能。此外,我们还将讨论Java内存模型,以及它如何影响并发程序的行为。最后,我们将提供一些实用的并发编程技巧和最佳实践,帮助开发者编写出既线程安全又高效的Java程序。
22 3
|
16天前
|
Java
Java 并发编程:深入理解线程池
【4月更文挑战第8天】本文将深入探讨 Java 中的线程池技术,包括其工作原理、优势以及如何使用。线程池是 Java 并发编程的重要工具,它可以有效地管理和控制线程的执行,提高系统性能。通过本文的学习,读者将对线程池有更深入的理解,并能在实际开发中灵活运用。
|
12天前
|
安全 算法 Java
深入理解Java并发编程:线程安全与性能优化
【4月更文挑战第11天】 在Java中,高效的并发编程是提升应用性能和响应能力的关键。本文将探讨Java并发的核心概念,包括线程安全、锁机制、线程池以及并发集合等,同时提供实用的编程技巧和最佳实践,帮助开发者在保证线程安全的前提下,优化程序性能。我们将通过分析常见的并发问题,如竞态条件、死锁,以及如何利用现代Java并发工具来避免这些问题,从而构建更加健壮和高效的多线程应用程序。
|
16天前
|
Java
Java并发编程:深入理解线程池
【4月更文挑战第7天】在现代软件开发中,多线程编程已经成为一种不可或缺的技术。为了提高程序性能和资源利用率,Java提供了线程池这一强大工具。本文将深入探讨Java线程池的原理、使用方法以及如何根据实际需求定制线程池,帮助读者更好地理解和应用线程池技术。
15 0
|
17天前
|
缓存 安全 Java
Java并发编程进阶:深入理解Java内存模型
【4月更文挑战第6天】Java内存模型(JMM)是多线程编程的关键,定义了线程间共享变量读写的规则,确保数据一致性和可见性。主要包括原子性、可见性和有序性三大特性。Happens-Before原则规定操作顺序,内存屏障和锁则保障这些原则的实施。理解JMM和相关机制对于编写线程安全、高性能的Java并发程序至关重要。
|
4天前
|
IDE Java 物联网
《Java 简易速速上手小册》第1章:Java 编程基础(2024 最新版)
《Java 简易速速上手小册》第1章:Java 编程基础(2024 最新版)
13 0
|
5天前
|
安全 Java 开发者
Java并发编程:深入理解Synchronized关键字
【4月更文挑战第19天】 在Java多线程编程中,为了确保数据的一致性和线程安全,我们经常需要使用到同步机制。其中,`synchronized`关键字是最为常见的一种方式,它能够保证在同一时刻只有一个线程可以访问某个对象的特定代码段。本文将深入探讨`synchronized`关键字的原理、用法以及性能影响,并通过具体示例来展示如何在Java程序中有效地应用这一技术。
|
6天前
|
安全 Java 调度
Java并发编程:深入理解线程与锁
【4月更文挑战第18天】本文探讨了Java中的线程和锁机制,包括线程的创建(通过Thread类、Runnable接口或Callable/Future)及其生命周期。Java提供多种锁机制,如`synchronized`关键字、ReentrantLock和ReadWriteLock,以确保并发访问共享资源的安全。此外,文章还介绍了高级并发工具,如Semaphore(控制并发线程数)、CountDownLatch(线程间等待)和CyclicBarrier(同步多个线程)。掌握这些知识对于编写高效、正确的并发程序至关重要。
|
6天前
|
缓存 分布式计算 监控
Java并发编程:深入理解线程池
【4月更文挑战第17天】在Java并发编程中,线程池是一种非常重要的技术,它可以有效地管理和控制线程的执行,提高系统的性能和稳定性。本文将深入探讨Java线程池的工作原理,使用方法以及在实际开发中的应用场景,帮助读者更好地理解和使用Java线程池。
|
7天前
|
Java API 数据库
深研Java异步编程:CompletableFuture与反应式编程范式的融合实践
【4月更文挑战第17天】本文探讨了Java中的CompletableFuture和反应式编程在提升异步编程体验上的作用。CompletableFuture作为Java 8引入的Future扩展,提供了一套流畅的链式API,简化异步操作,如示例所示的非阻塞数据库查询。反应式编程则关注数据流和变化传播,通过Reactor等框架实现高度响应的异步处理。两者结合,如将CompletableFuture转换为Mono或Flux,可以兼顾灵活性和资源管理,适应现代高并发环境的需求。开发者可按需选择和整合这两种技术,优化系统性能和响应能力。