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

使用spring自带的发布订阅来实现发布订阅

news 2025/7/14 7:28:52

背景

公司的项目以前代码里面有存在使用spring自带发布订阅的代码，因此稍微学习一下如何使用，并了解一下这种实现方式的优缺点。

优点

实现方便，代码方面基本只需要定义消息体和消费者，适用于小型应用程序。
不依赖外部中间件，因而不需要复杂的配置、部署。

缺点

无法提供消息持久性，项目一旦重启，消息就会丢失，因而不适合实现延迟队列。
对比消息队列，无法实现复杂的消息过滤、路由过滤。
无法实现跨应用程序的事件通信。不同应用程序之间的事件发布和订阅更为容易。

发布订阅模式的优缺点我就不说了，就说说不同实现方式之间的优缺点。

一、创建消息类

消息类需要继承ApplicationEvent类。因为java调用构造函数的机制就是默认会调用父类的构造函数，而ApplicationEvent类只有一个单参数的构造函数，无法自动调用，每个构造函数都需要显式调用父类的构造函数。也就是super(source);

package org.jeecg.modules.test.testPublic;
import org.springframework.context.ApplicationEvent;
import java.util.Objects;/*** @ClassName: MyEvent* @Author: zjc* @Date: 2023/8/30 18:22* @Description:**/
public class MyEvent extends ApplicationEvent {private String taskId;private Integer sourceType;public MyEvent(Object source) {super(source);}/****  @param source   触发事件的对象，可随便传，不过建议传自己可能用得到的对象。好像在调用放直接this的挺多* @param taskId    任务id，自己定义的事件要处理的内容* @param sourceType    自己定义的源类型，用来在多场景触发情况下区分不同场景的标志* @return: null**/public MyEvent(Object source,String taskId,Integer sourceType) {super(source);this.taskId=taskId;this.sourceType=sourceType;}public String getTaskId() {return taskId;}public void setTaskId(String taskId) {this.taskId = taskId;}public Integer getSourceType() {return sourceType;}public void setSourceType(Integer sourceType) {this.sourceType = sourceType;}@Overridepublic boolean equals(Object o) {if (this == o) return true;if (o == null || getClass() != o.getClass()) return false;MyEvent myEvent = (MyEvent) o;return Objects.equals(taskId, myEvent.taskId) && Objects.equals(sourceType, myEvent.sourceType);}@Overridepublic int hashCode() {return Objects.hash(taskId, sourceType);}@Overridepublic String toString() {return "MyEvent{" +"taskId='" + taskId + '\'' +", sourceType=" + sourceType +'}';}
}

二、发布消息

发布消息可以直接使用ApplicationContext对象调用publishEvent方法。因为ApplicationContext接口继承了ApplicationEventPublisher接口。注意消息类必须要继承ApplicationEvent类才能作为参数发布消息。

package org.jeecg.modules.test.testPublic;import io.swagger.annotations.Api;
import io.swagger.annotations.ApiOperation;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.ApplicationContext;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;/*** @ClassName: EventController* @Author: * @Date: 2023/8/31 17:53* @Description:**/
@RestController
@Api("test")
@RequestMapping("/test")
public class EventController {@Autowiredprivate ApplicationContext applicationContext;@ApiOperation("testEvent")@GetMapping("/testEvent")public void testEvent(){MyEvent myEvent=new MyEvent(this,"123456",1);applicationContext.publishEvent(myEvent);}@ApiOperation("testEvent1")@GetMapping("/testEvent1")public void testEvent1(){MyEvent myEvent=new MyEvent(this,"123456",2);applicationContext.publishEvent(myEvent);}
}

三、监听消息

监听消息类需要实现ApplicationListener接口，并通过泛型传入要监听的消息类，并重写onApplicationEvent方法。spring内的同一个消息可以有多个监听类，一旦监听到消息，监听该消息的全部监听类都会执行。

监听类1：

package org.jeecg.modules.test.testPublic;import org.springframework.context.ApplicationListener;
import org.springframework.stereotype.Component;/*** @ClassName: MyEventListener* @Author: * @Date: 2023/8/31 17:50* @Description:**/
@Component
public class MyEventListener implements ApplicationListener<MyEvent> {@Overridepublic void onApplicationEvent(MyEvent event) {System.out.println("消费者开始消费"+event.toString()+event.getSource().toString());}
}

消费者2

package org.jeecg.modules.test.testPublic;import org.springframework.context.ApplicationListener;
import org.springframework.stereotype.Component;/*** @ClassName: MyEventListener* @Author: * @Date: 2023/8/31 17:50* @Description:**/
@Component
public class MyEventListener1 implements ApplicationListener<MyEvent> {@Overridepublic void onApplicationEvent(MyEvent event) {System.out.println("消费者1开始消费"+event.toString()+event.getSource().toString());}
}