spring中发布订阅模式实践

概述

在实际开发中，经常会碰到要去解耦合一些依赖调用，比如单成功后会发送手机短信、发送绑定邮箱、数据库更新后通知发送mq等。而且通知这个操作又不希望强耦合在主业务流程中，这个时候我们很容易就想到了发布订阅（观察者）设计模式。

首先看代码实现:

  1. 定义事件类型 

    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    16
    17
    import org.springframework.context.ApplicationEvent;

    public class MessageEvent extends ApplicationEvent {

        private String message;

        public MessageEvent(Object source, String message) {
            super(source);
            this.message = message;
        }


        public String getMessage() {
            return message;
        }
    }

  2. 定义监听类
    写法1：

    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    import org.springframework.context.ApplicationListener;
    import org.springframework.scheduling.annotation.Async;
    import org.springframework.stereotype.Component;

    @Component
    public class MessageListener implements ApplicationListener<MessageEvent> {

        @Override
        // @Async 开启异步执行
        public void onApplicationEvent(MessageEvent messageEvent) {
            System.out.println("用类监听到了消息："+messageEvent.getMessage());
        }
    }

    写法二：

    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    import org.springframework.context.event.EventListener;
    import org.springframework.scheduling.annotation.Async;
    import org.springframework.stereotype.Component;

    @Component
    public class MessageListener2 {

        @EventListener
        // @Async 开启异步执行
        public void messageListener(MessageEvent messageEvent) {
            String message = messageEvent.getMessage();
            System.out.println("用注解类监听到了消息：" + message);
        }
    }

    开启异步执行需要在启动类添加@EnableAsync注解

    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    import org.springframework.boot.SpringApplication;
    import org.springframework.boot.autoconfigure.SpringBootApplication;
    import org.springframework.scheduling.annotation.EnableAsync;

    @SpringBootApplication
    @EnableAsync
    public class DemoApplication {

        public static void main(String[] args) {
            SpringApplication.run(DemoApplication.class, args);
        }

    }

  3. 定义事件发布类

    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    16
    17
    18
    import com.example.demo.listener_demo.MessageEvent;
    import org.junit.jupiter.api.Test;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.boot.test.context.SpringBootTest;
    import org.springframework.context.ApplicationContext;

    @SpringBootTest
    public class ListenerTest {

        @Autowired
        private ApplicationContext applicationContext;

        @Test
        public void listenerTest(){
            applicationContext.publishEvent(new MessageEvent(this,"这是条消息。。。"));
            System.out.println("消息发完了。。。");
        }
    }

  4. 实践结果

未开始异步：

1
2
3
用类监听到了消息：这是条消息。。。
用注解类监听到了消息：这是条消息。。。
消息发完了。。。

开启异步：

1
2
3
消息发完了。。。
用类监听到了消息：这是条消息。。。
用注解类监听到了消息：这是条消息。。。

原理分析

事件

ApplicationEvent该抽象类继承自JDK的EventObject。JDK要求所有事件将继承它，并通过source得到事件源。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
package org.springframework.context;

import java.util.EventObject;

/**
 * Class to be extended by all application events. Abstract as it
 * doesn't make sense for generic events to be published directly.
 *
 * @author Rod Johnson
 * @author Juergen Hoeller
 */
public abstract class ApplicationEvent extends EventObject {

  /** use serialVersionUID from Spring 1.2 for interoperability */
  private static final long serialVersionUID = 7099057708183571937L;

  /** System time when the event happened */
  private final long timestamp;


  /**
   * Create a new ApplicationEvent.
   * @param source the object on which the event initially occurred (never {@code null})
   */
  public ApplicationEvent(Object source) {
    super(source);
    this.timestamp = System.currentTimeMillis();
  }


  /**
   * Return the system time in milliseconds when the event happened.
   */
  public final long getTimestamp() {
    return this.timestamp;
  }

}

发布者

ApplicationEventPublisher及ApplicationEventMulticaster。ApplicationContext该接口继承了ApplicationEventPublisher，并在AbstractApplicationContext实现了具体代码，实际执行是委托给ApplicationEventMulticaster（可以认为是多播）：

ApplicationContext继承自ApplicationEventPublisher

1
public interface ApplicationContext extends EnvironmentCapable, ListableBeanFactory, HierarchicalBeanFactory, MessageSource, ApplicationEventPublisher, ResourcePatternResolver {

ApplicationEventPublisher定义了publishEvent方法:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
public interface ApplicationEventPublisher {

  /**
   * Notify all <strong>matching</strong> listeners registered with this
   * application of an application event. Events may be framework events
   * (such as RequestHandledEvent) or application-specific events.
   * @param event the event to publish
   * @see org.springframework.web.context.support.RequestHandledEvent
   */
  void publishEvent(ApplicationEvent event);

  /**
   * Notify all <strong>matching</strong> listeners registered with this
   * application of an event.
   * <p>If the specified {@code event} is not an {@link ApplicationEvent},
   * it is wrapped in a {@link PayloadApplicationEvent}.
   * @param event the event to publish
   * @since 4.2
   * @see PayloadApplicationEvent
   */
  void publishEvent(Object event);

}

在AbstractApplicationContext实现了具体代码,实际执行是委托给ApplicationEventMulticaster（可以认为是多播）

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
protected void publishEvent(Object event, ResolvableType eventType) {
    Assert.notNull(event, "Event must not be null");
    if (logger.isTraceEnabled()) {
      logger.trace("Publishing event in " + getDisplayName() + ": " + event);
    }

    // Decorate event as an ApplicationEvent if necessary
    ApplicationEvent applicationEvent;
    if (event instanceof ApplicationEvent) {
      applicationEvent = (ApplicationEvent) event;
    }
    else {
      applicationEvent = new PayloadApplicationEvent<Object>(this, event);
      if (eventType == null) {
        eventType = ((PayloadApplicationEvent)applicationEvent).getResolvableType();
      }
    }

    // Multicast right now if possible - or lazily once the multicaster is initialized
    if (this.earlyApplicationEvents != null) {
      this.earlyApplicationEvents.add(applicationEvent);
    }
    else {
      getApplicationEventMulticaster().multicastEvent(applicationEvent, eventType);
    }

    // Publish event via parent context as well...
    if (this.parent != null) {
      if (this.parent instanceof AbstractApplicationContext) {
        ((AbstractApplicationContext) this.parent).publishEvent(event, eventType);
      }
      else {
        this.parent.publishEvent(event);
      }
    }
  }

ApplicationContext自动到本地容器里找一个ApplicationEventMulticaster实现，如果没有自己new一个SimpleApplicationEventMulticaster。 

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
public void multicastEvent(final ApplicationEvent event, ResolvableType eventType) {
    ResolvableType type = (eventType != null ? eventType : resolveDefaultEventType(event));
        // 拿到所有的监听器一一调用
    for (final ApplicationListener<?> listener : getApplicationListeners(event, type)) {
      Executor executor = getTaskExecutor();
      if (executor != null) {
        executor.execute(new Runnable() {
          @Override
          public void run() {
            invokeListener(listener, event);
          }
        });
      }
      else {
        invokeListener(listener, event);
      }
    }
  }

可以看到如果给它一个executor（java.util.concurrent.Executor），它就可以异步支持发布事件了。所以发送事件只需要通过ApplicationContext.publishEvent即可

监听器

以ApplicationListener 为例：
ApplicationListener其继承自JDK的EventListener

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
package org.springframework.context;

import java.util.EventListener;

/**
 * Interface to be implemented by application event listeners.
 * Based on the standard {@code java.util.EventListener} interface
 * for the Observer design pattern.
 *
 * <p>As of Spring 3.0, an ApplicationListener can generically declare the event type
 * that it is interested in. When registered with a Spring ApplicationContext, events
 * will be filtered accordingly, with the listener getting invoked for matching event
 * objects only.
 *
 * @author Rod Johnson
 * @author Juergen Hoeller
 * @param <E> the specific ApplicationEvent subclass to listen to
 * @see org.springframework.context.event.ApplicationEventMulticaster
 */
public interface ApplicationListener<E extends ApplicationEvent> extends EventListener {

  /**
   * Handle an application event.
   * @param event the event to respond to
   */
  void onApplicationEvent(E event);

}

总结

使用spring实现事件监听机制非常简单，涉及到者几个类文件 ：ApplicationEvent(事件类型)、ApplicationListener（事件监听类）、ApplicationEventPublisher（事件发布类）。

Java框架 > spring
#设计模式
spring中发布订阅模式实践
https://www.weypage.com/2020/03/22/Java框架/spring/应用/spring中发布订阅模式实践/
作者
weylan
发布于
2020年3月22日
许可协议