quarkus依赖注入之六:发布和消费事件
欢迎访问我的GitHub
这里分类和汇总了欣宸的全部原创(含配套源码):https://github.com/zq2599/blog_demos
本篇概览
- 本文是《quarkus依赖注入》系列的第六篇,主要内容是学习事件的发布和接收
- 如果您用过Kafka、RabbitMQ等消息中间件,对消息的作用应该不会陌生,通过消息的订阅和发布可以降低系统之间的耦合性,这种方式也可以用在应用内部的多个模块之间,在quarkus框架下就是事件的发布和接收
- 本篇会演示quarkus应用中如何发布事件、如何接收事件,全文由以下章节构成
- 同步事件
- 异步事件
- 同一种事件类,用在不同的业务场景
- 优化
- 事件元数据
同步事件
- 同步事件是指事件发布后,事件接受者会在同一个线程处理事件,对事件发布者来说,相当于发布之后的代码不会立即执行,要等到事件处理的代码执行完毕后
- 同步事件发布和接受的开发流程如下图
接下来编码实践,先定义事件类MyEvent.java,如下所示,该类有两个字段,source表示来源,consumeNum作为计数器可以累加
public class MyEvent { /** * 事件源 */ private String source;
/** * 事件被消费的总次数 */ private AtomicInteger consumeNum; public MyEvent(String source) { this.source = source; consumeNum = new AtomicInteger(); } /** * 事件被消费次数加一 * @return */ public int addNum() { return consumeNum.incrementAndGet(); } /** * 获取事件被消费次数 * @return */ public int getNum() { return consumeNum.get(); } @Override public String toString() { return "MyEvent{" + "source='" + source + '\'' + ", consumeNum=" + getNum() + '}'; }
}
然后是发布事件类,有几处要注意的地方稍后会提到
package com.bolingcavalry.event.producer;
import com.bolingcavalry.event.bean.MyEvent; import io.quarkus.logging.Log; import javax.enterprise.context.ApplicationScoped; import javax.enterprise.event.Event; import javax.inject.Inject;
@ApplicationScoped public class MyProducer {
@Inject Event<MyEvent> event; /** * 发送同步消息 * @param source 消息源 * @return 被消费次数 */ public int syncProduce(String source) { MyEvent myEvent = new MyEvent("syncEvent"); Log.infov("before sync fire, {0}", myEvent); event.fire(myEvent); Log.infov("after sync fire, {0}", myEvent); return myEvent.getNum(); }
}
上述代码有以下几点要注意:
- 注入Event,用于发布事件,通过泛型指定事件类型是MyEvent
- 发布同步事件很简单,调用fire即可
- 由于是同步事件,会等待事件的消费者将消费的代码执行完毕后,fire方法才会返回
- 如果消费者增加了myEvent的记数,那么myEvent.getNum()应该等于计数的调用次数
接下来是消费事件的代码,如下所示,只要方法的入参是事件类MyEvent,并且用@Observes修饰该入参,即可成为MyEvent事件的同步消费者,这里用sleep来模拟执行了一个耗时的业务操作
package com.bolingcavalry.event.consumer;
import com.bolingcavalry.event.bean.MyEvent; import io.quarkus.logging.Log; import javax.enterprise.context.ApplicationScoped; import javax.enterprise.event.Observes;
@ApplicationScoped public class MyConsumer {
/** * 消费同步事件 * @param myEvent */ public void syncConsume(@Observes MyEvent myEvent) { Log.infov("receive sync event, {0}", myEvent); // 模拟业务执行,耗时100毫秒 try { Thread.sleep(100); } catch (InterruptedException e) { e.printStackTrace(); } // 计数加一 myEvent.addNum(); }
}
最后,写单元测试类验证功能,在MyProducer的syncProduce方法中,由于是同步事件,MyConsumer.syncConsume方法执行完毕才会继续执行event.fire后面的代码,所以syncProduce的返回值应该等于1
package com.bolingcavalry;
import com.bolingcavalry.event.consumer.MyConsumer; import com.bolingcavalry.event.producer.MyProducer; import com.bolingcavalry.service.HelloInstance; import com.bolingcavalry.service.impl.HelloInstanceA; import com.bolingcavalry.service.impl.HelloInstanceB; import io.quarkus.test.junit.QuarkusTest; import org.junit.jupiter.api.Assertions; import org.junit.jupiter.api.Test;
import javax.enterprise.inject.Instance; import javax.inject.Inject;
@QuarkusTest public class EventTest {
@Inject MyProducer myProducer; @Inject MyConsumer myConsumer; @Test public void testSync() { Assertions.assertEquals(1, myProducer.syncProduce("testSync")); }
}
执行单元测试,如下所示,符合预期,事件的发送和消费在同一线程内顺序执行,另外请关注日志的时间戳,可见MyProducer的第二条日志,是在MyConsumer日志之后的一百多毫秒,这也证明了顺序执行的逻辑
- 以上就是同步事件的相关代码,很多场景中,消费事件的操作是比较耗时或者不太重要(例如写日志),这时候让发送事件的线程等待就不合适了,因为发送事件后可能还有其他重要的事情需要立即去做,这就是接下来的异步事件
异步事件
为了避免事件消费耗时过长对事件发送的线程造成影响,可以使用异步事件,还是用代码来说明
发送事件的代码还是写在MyPorducer.java,如下,有两处要注意的地方稍后提到
public int asyncProduce(String source) { MyEvent myEvent = new MyEvent(source); Log.infov("before async fire, {0}", myEvent); event.fireAsync(myEvent) .handleAsync((e, error) -> { if (null!=error) { Log.error("handle error", error); } else { Log.infov("finish handle, {0}", myEvent); } return null; }); Log.infov("after async fire, {0}", myEvent); return myEvent.getNum(); }
上述代码有以下两点要注意:
- 发送异步事件的API是fireAsync
- fireAsync的返回值是CompletionStage,我们可以调用其handleAsync方法,将响应逻辑(对事件消费结果的处理)传入,这段响应逻辑会在事件消费结束后被执行,上述代码中的响应逻辑是检查异常,若有就打印
消费异步事件的代码写在MyConsumer,与同步的相比唯一的变化就是修饰入参的注解改成了ObservesAsync
public void aSyncConsume(@ObservesAsync MyEvent myEvent) { Log.infov("receive async event, {0}", myEvent); // 模拟业务执行,耗时100毫秒 try { Thread.sleep(100); } catch (InterruptedException e) { e.printStackTrace(); } // 计数加一 myEvent.addNum(); }
单元测试代码,有两点需要注意,稍后会提到
@Test public void testAsync() throws InterruptedException { Assertions.assertEquals(0, myProducer.asyncProduce("testAsync")); // 如果不等待的话,主线程结束的时候会中断正在消费事件的子线程,导致子线程报错 Thread.sleep(150); }
上述代码有以下两点需要注意
- 异步事件的时候,发送事件的线程不会等待,所以myEvent实例的计数器在消费线程还没来得及加一,myProducer.asyncProduce方法就已经执行结束了,返回值是0,所以单元测试的assertEquals位置,期望值应该是0
- testAsync方法要等待100毫秒以上才能结束,否则进程会立即结束,导致正在消费事件的子线程被打断,抛出异常
- 执行单元测试,控制台输出如下图,测试通过,有三个重要信息稍后会提到
- 上图中有三个关键信息
- 事件发布前后的两个日志是紧紧相连的,这证明发送事件之后不会等待消费,而是立即继续执行发送线程的代码
- 消费事件的日志显示,消费逻辑是在一个新的线程中执行的
- 消费结束后的回调代码中也打印了日志,显示这端逻辑又在一个新的线程中执行,此线程与发送事件、消费事件都不在同一线程
- 以上就是基础的异步消息发送和接受操作,接下来去看略为复杂的场景
同一种事件类,用在不同的业务场景
- 设想这样一个场景:管理员发送XXX类型的事件,消费者应该是处理管理员事件的方法,普通用户也发送XXX类型的事件,消费者应该是处理普通用户事件的方法,简单的说就是同一个数据结构的事件可能用在不同场景,如下图
从技术上分析,实现上述功能的关键点是:消息的消费者要精确过滤掉不该自己消费的消息
此刻,您是否回忆起前面文章中的一个场景:依赖注入时,如何从多个bean中选择自己所需的那个,这两个问题何其相似,而依赖注入的选择问题是用Qualifier注解解决的,今天的消息场景,依旧可以用Qualifier来对消息做精确过滤,接下来编码实战
首先定义事件类ChannelEvent.java,管理员和普通用户的消息数据都用这个类(和前面的MyEvent事件类的代码一样)
public class TwoChannelEvent { /** * 事件源 */ private String source;
/** * 事件被消费的总次数 */ private AtomicInteger consumeNum; public TwoChannelEvent(String source) { this.source = source; consumeNum = new AtomicInteger(); } /** * 事件被消费次数加一 * @return */ public int addNum() { return consumeNum.incrementAndGet(); } /** * 获取事件被消费次数 * @return */ public int getNum() { return consumeNum.get(); } @Override public String toString() { return "TwoChannelEvent{" + "source='" + source + '\'' + ", consumeNum=" + getNum() + '}'; }
}
然后就是关键点:自定义注解Admin,这是管理员事件的过滤器,要用Qualifier修饰
package com.bolingcavalry.annonation;
import javax.inject.Qualifier; import java.lang.annotation.Retention; import java.lang.annotation.Target; import static java.lang.annotation.ElementType.FIELD; import static java.lang.annotation.ElementType.PARAMETER; import static java.lang.annotation.RetentionPolicy.RUNTIME;
@Qualifier @Retention(RUNTIME) @Target({FIELD, PARAMETER}) public @interface Admin { }
自定义注解Normal,这是普通用户事件的过滤器,要用Qualifier修饰
@Qualifier @Retention(RUNTIME) @Target({FIELD, PARAMETER}) public @interface Normal { }
Admin和Normal先用在发送事件的代码中,再用在消费事件的代码中,这样就完成了匹配,先写发送代码,有几处要注意的地方稍后会提到
@ApplicationScoped public class TwoChannelWithTwoEvent {
@Inject @Admin Event<TwoChannelEvent> adminEvent; @Inject @Normal Event<TwoChannelEvent> normalEvent; /** * 管理员消息 * @param source * @return */ public int produceAdmin(String source) { TwoChannelEvent event = new TwoChannelEvent(source); adminEvent.fire(event); return event.getNum(); } /** * 普通消息 * @param source * @return */ public int produceNormal(String source) { TwoChannelEvent event = new TwoChannelEvent(source); normalEvent.fire(event); return event.getNum(); }
}
上述代码有以下两点需要注意
- 注入了两个Event实例adminEvent和normalEvent,它们的类型一模一样,但是分别用Admin和Normal
注解修饰,相当于为它们添加了不同的标签,在消费的时候也可以用这两个注解来过滤
- 发送代码并无特别之处,用adminEvent.fire发出的事件,在消费的时候不过滤、或者用Admin过滤,这两种方式都能收到
接下来看消费事件的代码TwoChannelConsumer.java,有几处要注意的地方稍后会提到
@ApplicationScoped public class TwoChannelConsumer {
/** * 消费管理员事件 * @param event */ public void adminEvent(@Observes @Admin TwoChannelEvent event) { Log.infov("receive admin event, {0}", event); // 管理员的计数加两次,方便单元测试验证 event.addNum(); event.addNum(); } /** * 消费普通用户事件 * @param event */ public void normalEvent(@Observes @Normal TwoChannelEvent event) { Log.infov("receive normal event, {0}", event); // 计数加一 event.addNum(); } /** * 如果不用注解修饰,所有TwoChannelEvent类型的事件都会在此被消费 * @param event */ public void allEvent(@Observes TwoChannelEvent event) { Log.infov("receive event (no Qualifier), {0}", event); // 计数加一 event.addNum(); }
}
上述代码有以下两处需要注意
- 消费事件的方法,除了Observes注解,再带上Admin,这样此方法只会消费Admin修饰的Event发出的事件
- allEvent只有Observes注解,这就意味着此方法不做过滤,只要是TwoChannelEvent类型的同步事件,它都会消费
- 为了方便后面的验证,在消费Admin事件时,计数器执行了两次,而Normal事件只有一次,这样两种事件的消费结果就不一样了
以上就是同一事件类在多个场景被同时使用的代码了,接下来写单元测试验证
@QuarkusTest public class EventTest {
@Inject TwoChannelWithTwoEvent twoChannelWithTwoEvent; @Test public void testTwoChnnelWithTwoEvent() { // 对管理员来说, // TwoChannelConsumer.adminEvent消费时计数加2, // TwoChannelConsumer.allEvent消费时计数加1, // 所以最终计数是3 Assertions.assertEquals(3, twoChannelWithTwoEvent.produceAdmin("admin")); // 对普通人员来说, // TwoChannelConsumer.normalEvent消费时计数加1, // TwoChannelConsumer.allEvent消费时计数加1, // 所以最终计数是2 Assertions.assertEquals(2, twoChannelWithTwoEvent.produceNormal("normal")); }
}
执行单元测试顺利通过,如下图
小优化,不需要注入多个Event实例
- 刚才的代码虽然可以正常工作,但是有一点小瑕疵:为了发送不同事件,需要注入不同的Event实例,如下图红框,如果事件类型越来越多,注入的Event实例岂不是越来越多?
quarkus提供了一种缓解上述问题的方式,再写一个发送事件的类TwoChannelWithSingleEvent.java,代码中有两处要注意的地方稍后会提到
/**
@author will
@email zq2599@gmail.com
@date 2022/4/3 10:16
@description 用同一个事件结构体TwoChannelEvent,分别发送不同业务类型的事件 */ @ApplicationScoped public class TwoChannelWithSingleEvent {
@Inject Event
singleEvent; /**
管理员消息
@param source
@return */ public int produceAdmin(String source) { TwoChannelEvent event = new TwoChannelEvent(source);
singleEvent.select(new AnnotationLiteral
() {}) .fire(event); return event.getNum(); }
/** * 普通消息 * @param source * @return */ public int produceNormal(String source) { TwoChannelEvent event = new TwoChannelEvent(source);
singleEvent.select(new AnnotationLiteral<Normal>() {}) .fire(event); return event.getNum();
} }
上述发送消息的代码,有以下两处需要注意
- 不论是Admin事件还是Normal事件,都是用singleEvent发送的,如此避免了事件类型越多Event实例越多的情况发生
- 执行fire方法发送事件前,先执行select方法,入参是AnnotationLiteral的匿名子类,并且通过泛型指定事件类型 ,这和前面TwoChannelWithTwoEvent类发送两种类型消息的效果是一样的
既然用select方法过滤和前面两个Event实例的效果一样,那么消费事件的类就不改动了
写个单元测试来验证效果
@QuarkusTest public class EventTest { @Inject TwoChannelWithSingleEvent twoChannelWithSingleEvent;
@Test public void testTwoChnnelWithSingleEvent() { // 对管理员来说, // TwoChannelConsumer.adminEvent消费时计数加2, // TwoChannelConsumer.allEvent消费时计数加1, // 所以最终计数是3 Assertions.assertEquals(3, twoChannelWithSingleEvent.produceAdmin("admin")); // 对普通人员来说, // TwoChannelConsumer.normalEvent消费时计数加1, // TwoChannelConsumer.allEvent消费时计数加1, // 所以最终计数是2 Assertions.assertEquals(2, twoChannelWithSingleEvent.produceNormal("normal")); }
}
如下图所示,单元测试通过,也就说从消费者的视角来看,两种消息发送方式并无区别
事件元数据
在消费事件时,除了从事件对象中取得业务数据(例如MyEvent的source和consumeNum字段),有时还可能需要用到事件本身的信息,例如类型是Admin还是Normal、Event对象的注入点在哪里等,这些都算是事件的元数据
为了演示消费者如何取得事件元数据,将TwoChannelConsumer.java的allEvent方法改成下面的样子,需要注意的地方稍后会提到
public void allEvent(@Observes TwoChannelEvent event, EventMetadata eventMetadata) { Log.infov(“receive event (no Qualifier), {0}”, event);
// 打印事件类型 Log.infov("event type : {0}", eventMetadata.getType()); // 获取该事件的所有注解 Set<Annotation> qualifiers = eventMetadata.getQualifiers(); // 将事件的所有注解逐个打印 if (null!=qualifiers) { qualifiers.forEach(annotation -> Log.infov("qualify : {0}", annotation)); } // 计数加一 event.addNum();
}
上述代码中,以下几处需要注意
- 给allEvent方法增加一个入参,类型是EventMetadata,bean容器会将事件的元数据设置到此参数
- EventMetadata的getType方法能取得事件类型
- EventMetadata的getType方法能取得事件的所有修饰注解,包括Admin或者Normal
- 运行刚才的单元测试,看修改后的allEvent方法执行会有什么输出,如下图,红框1打印出事件是TwoChannelEvent实例,红框2将修饰事件的注解打印出来了,包括发送时修饰的Admin
- 至此,事件相关的学习和实战就完成了,进程内用事件可以有效地解除模块间的耦合,希望本文能给您一些参考
欢迎关注博客园:程序员欣宸
学习路上,你不孤单,欣宸原创一路相伴…
/xiao-cheng-xu-ding-zhi/quarkusyi-lai-zhu-ru-zhi-liu-fa-bu-he-xiao-fei-shi-jian-332.html