quarkus依赖注入之六：发布和消费事件

转x33g5p2x  于2022-04-10 转载在 其他  
字(10.7k)|赞(0)|评价(0)|浏览(558)

欢迎访问我的GitHub

这里分类和汇总了欣宸的全部原创(含配套源码)：https://github.com/zq2599/blog_demos

本篇概览

  • 本文是《quarkus依赖注入》系列的第六篇，主要内容是学习事件的发布和接收
  • 如果您用过Kafka、RabbitMQ等消息中间件，对消息的作用应该不会陌生，通过消息的订阅和发布可以降低系统之间的耦合性，这种方式也可以用在应用内部的多个模块之间，在quarkus框架下就是事件的发布和接收
  • 本篇会演示quarkus应用中如何发布事件、如何接收事件，全文由以下章节构成
  1. 同步事件
  2. 异步事件
  3. 同一种事件类，用在不同的业务场景
  4. 优化
  5. 事件元数据

同步事件

  • 同步事件是指事件发布后，事件接受者会在同一个线程处理事件，对事件发布者来说，相当于发布之后的代码不会立即执行，要等到事件处理的代码执行完毕后
  • 同步事件发布和接受的开发流程如下图

  • 接下来编码实践，先定义事件类MyEvent.java，如下所示，该类有两个字段，source表示来源，consumeNum作为计数器可以累加
  1. public class MyEvent {
  2.     /**
  3.      * 事件源
  4.      */
  5.     private String source;
  7.     /**
  8.      * 事件被消费的总次数
  9.      */
  10.     private AtomicInteger consumeNum;
  12.     public MyEvent(String source) {
  13.         this.source = source;
  14.         consumeNum = new AtomicInteger();
  15.     }
  17.     /**
  18.      * 事件被消费次数加一
  19.      * @return
  20.      */
  21.     public int addNum() {
  22.         return consumeNum.incrementAndGet();
  23.     }
  25.     /**
  26.      * 获取事件被消费次数
  27.      * @return
  28.      */
  29.     public int getNum() {
  30.         return consumeNum.get();
  31.     }
  33.     @Override
  34.     public String toString() {
  35.         return "MyEvent{" +
  36.                 "source='" + source + '\'' +
  37.                 ", consumeNum=" + getNum() +
  38.                 '}';
  39.     }
  40. }
  • 然后是发布事件类，有几处要注意的地方稍后会提到
  1. package com.bolingcavalry.event.producer;
  3. import com.bolingcavalry.event.bean.MyEvent;
  4. import io.quarkus.logging.Log;
  5. import javax.enterprise.context.ApplicationScoped;
  6. import javax.enterprise.event.Event;
  7. import javax.inject.Inject;
  9. @ApplicationScoped
  10. public class MyProducer {
  12.     @Inject
  13.     Event<MyEvent> event;
  15.     /**
  16.      * 发送同步消息
  17.      * @param source 消息源
  18.      * @return 被消费次数
  19.      */
  20.     public int syncProduce(String source) {
  21.         MyEvent myEvent = new MyEvent("syncEvent");
  22.         Log.infov("before sync fire, {0}", myEvent);
  23.         event.fire(myEvent);
  24.         Log.infov("after sync fire, {0}", myEvent);
  25.         return myEvent.getNum();
  26.     }
  27. }
  • 上述代码有以下几点要注意：
  1. 注入Event，用于发布事件，通过泛型指定事件类型是MyEvent
  2. 发布同步事件很简单，调用fire即可
  3. 由于是同步事件，会等待事件的消费者将消费的代码执行完毕后，fire方法才会返回
  4. 如果消费者增加了myEvent的记数，那么myEvent.getNum()应该等于计数的调用次数
  • 接下来是消费事件的代码，如下所示，只要方法的入参是事件类MyEvent，并且用@Observes修饰该入参，即可成为MyEvent事件的同步消费者，这里用sleep来模拟执行了一个耗时的业务操作
  1. package com.bolingcavalry.event.consumer;
  3. import com.bolingcavalry.event.bean.MyEvent;
  4. import io.quarkus.logging.Log;
  5. import javax.enterprise.context.ApplicationScoped;
  6. import javax.enterprise.event.Observes;
  8. @ApplicationScoped
  9. public class MyConsumer {
  11.     /**
  12.      * 消费同步事件
  13.      * @param myEvent
  14.      */
  15.     public void syncConsume(@Observes MyEvent myEvent) {
  16.         Log.infov("receive sync event, {0}", myEvent);
  18.         // 模拟业务执行，耗时100毫秒
  19.         try {
  20.             Thread.sleep(100);
  21.         } catch (InterruptedException e) {
  22.             e.printStackTrace();
  23.         }
  25.         // 计数加一
  26.         myEvent.addNum();
  27.     }
  28. }
  • 最后，写单元测试类验证功能，在MyProducer的syncProduce方法中，由于是同步事件，MyConsumer.syncConsume方法执行完毕才会继续执行event.fire后面的代码，所以syncProduce的返回值应该等于1
  1. package com.bolingcavalry;
  3. import com.bolingcavalry.event.consumer.MyConsumer;
  4. import com.bolingcavalry.event.producer.MyProducer;
  5. import com.bolingcavalry.service.HelloInstance;
  6. import com.bolingcavalry.service.impl.HelloInstanceA;
  7. import com.bolingcavalry.service.impl.HelloInstanceB;
  8. import io.quarkus.test.junit.QuarkusTest;
  9. import org.junit.jupiter.api.Assertions;
  10. import org.junit.jupiter.api.Test;
  12. import javax.enterprise.inject.Instance;
  13. import javax.inject.Inject;
  15. @QuarkusTest
  16. public class EventTest {
  18.     @Inject
  19.     MyProducer myProducer;
  21.     @Inject
  22.     MyConsumer myConsumer;
  24.     @Test
  25.     public void testSync() {
  26.         Assertions.assertEquals(1, myProducer.syncProduce("testSync"));
  27.     }
  28. }
  • 执行单元测试，如下所示，符合预期，事件的发送和消费在同一线程内顺序执行，另外请关注日志的时间戳，可见MyProducer的第二条日志，是在MyConsumer日志之后的一百多毫秒，这也证明了顺序执行的逻辑

  • 以上就是同步事件的相关代码，很多场景中，消费事件的操作是比较耗时或者不太重要（例如写日志），这时候让发送事件的线程等待就不合适了，因为发送事件后可能还有其他重要的事情需要立即去做，这就是接下来的异步事件

异步事件

  • 为了避免事件消费耗时过长对事件发送的线程造成影响，可以使用异步事件，还是用代码来说明
  • 发送事件的代码还是写在MyPorducer.java，如下，有两处要注意的地方稍后提到
  1. public int asyncProduce(String source) {
  2.         MyEvent myEvent = new MyEvent(source);
  3.         Log.infov("before async fire, {0}", myEvent);
  4.         event.fireAsync(myEvent)
  5.              .handleAsync((e, error) -> {
  6.                  if (null!=error) {
  7.                      Log.error("handle error", error);
  8.                  } else {
  9.                      Log.infov("finish handle, {0}", myEvent);
  10.                  }
  12.                  return null;
  13.              });
  14.         Log.infov("after async fire, {0}", myEvent);
  15.         return myEvent.getNum();
  16.     }
  • 上述代码有以下两点要注意：
  1. 发送异步事件的API是fireAsync
  2. fireAsync的返回值是CompletionStage，我们可以调用其handleAsync方法，将响应逻辑（对事件消费结果的处理）传入，这段响应逻辑会在事件消费结束后被执行，上述代码中的响应逻辑是检查异常，若有就打印
  • 消费异步事件的代码写在MyConsumer，与同步的相比唯一的变化就是修饰入参的注解改成了ObservesAsync
  1. public void aSyncConsume(@ObservesAsync MyEvent myEvent) {
  2.         Log.infov("receive async event, {0}", myEvent);
  4.         // 模拟业务执行，耗时100毫秒
  5.         try {
  6.             Thread.sleep(100);
  7.         } catch (InterruptedException e) {
  8.             e.printStackTrace();
  9.         }
  11.         // 计数加一
  12.         myEvent.addNum();
  13.     }
  • 单元测试代码，有两点需要注意，稍后会提到
  1. @Test
  2.     public void testAsync() throws InterruptedException {
  3.         Assertions.assertEquals(0, myProducer.asyncProduce("testAsync"));
  4.         // 如果不等待的话，主线程结束的时候会中断正在消费事件的子线程，导致子线程报错
  5.         Thread.sleep(150);
  6.     }
  • 上述代码有以下两点需要注意
  1. 异步事件的时候，发送事件的线程不会等待，所以myEvent实例的计数器在消费线程还没来得及加一，myProducer.asyncProduce方法就已经执行结束了，返回值是0，所以单元测试的assertEquals位置，期望值应该是0
  2. testAsync方法要等待100毫秒以上才能结束，否则进程会立即结束，导致正在消费事件的子线程被打断，抛出异常
  • 执行单元测试，控制台输出如下图，测试通过，有三个重要信息稍后会提到

  • 上图中有三个关键信息
  1. 事件发布前后的两个日志是紧紧相连的，这证明发送事件之后不会等待消费，而是立即继续执行发送线程的代码
  2. 消费事件的日志显示，消费逻辑是在一个新的线程中执行的
  3. 消费结束后的回调代码中也打印了日志，显示这端逻辑又在一个新的线程中执行，此线程与发送事件、消费事件都不在同一线程
  • 以上就是基础的异步消息发送和接受操作，接下来去看略为复杂的场景

同一种事件类，用在不同的业务场景

  • 设想这样一个场景：管理员发送XXX类型的事件，消费者应该是处理管理员事件的方法，普通用户也发送XXX类型的事件，消费者应该是处理普通用户事件的方法，简单的说就是同一个数据结构的事件可能用在不同场景，如下图

  • 从技术上分析，实现上述功能的关键点是：消息的消费者要精确过滤掉不该自己消费的消息
  • 此刻，您是否回忆起前面文章中的一个场景：依赖注入时，如何从多个bean中选择自己所需的那个，这两个问题何其相似，而依赖注入的选择问题是用Qualifier注解解决的，今天的消息场景，依旧可以用Qualifier来对消息做精确过滤，接下来编码实战
  • 首先定义事件类ChannelEvent.java，管理员和普通用户的消息数据都用这个类（和前面的MyEvent事件类的代码一样）
  1. public class TwoChannelEvent {
  2.     /**
  3.      * 事件源
  4.      */
  5.     private String source;
  7.     /**
  8.      * 事件被消费的总次数
  9.      */
  10.     private AtomicInteger consumeNum;
  12.     public TwoChannelEvent(String source) {
  13.         this.source = source;
  14.         consumeNum = new AtomicInteger();
  15.     }
  17.     /**
  18.      * 事件被消费次数加一
  19.      * @return
  20.      */
  21.     public int addNum() {
  22.         return consumeNum.incrementAndGet();
  23.     }
  25.     /**
  26.      * 获取事件被消费次数
  27.      * @return
  28.      */
  29.     public int getNum() {
  30.         return consumeNum.get();
  31.     }
  33.     @Override
  34.     public String toString() {
  35.         return "TwoChannelEvent{" +
  36.                 "source='" + source + '\'' +
  37.                 ", consumeNum=" + getNum() +
  38.                 '}';
  39.     }
  40. }
  • 然后就是关键点：自定义注解Admin，这是管理员事件的过滤器，要用Qualifier修饰
  1. package com.bolingcavalry.annonation;
  3. import javax.inject.Qualifier;
  4. import java.lang.annotation.Retention;
  5. import java.lang.annotation.Target;
  6. import static java.lang.annotation.ElementType.FIELD;
  7. import static java.lang.annotation.ElementType.PARAMETER;
  8. import static java.lang.annotation.RetentionPolicy.RUNTIME;
  10. @Qualifier
  11. @Retention(RUNTIME)
  12. @Target({FIELD, PARAMETER})
  13. public @interface Admin {
  14. }
  • 自定义注解Normal，这是普通用户事件的过滤器，要用Qualifier修饰
  1. @Qualifier
  2. @Retention(RUNTIME)
  3. @Target({FIELD, PARAMETER})
  4. public @interface Normal {
  5. }
  • Admin和Normal先用在发送事件的代码中，再用在消费事件的代码中，这样就完成了匹配，先写发送代码，有几处要注意的地方稍后会提到
  1. @ApplicationScoped
  2. public class TwoChannelWithTwoEvent {
  4.     @Inject
  5.     @Admin
  6.     Event<TwoChannelEvent> adminEvent;
  8.     @Inject
  9.     @Normal
  10.     Event<TwoChannelEvent> normalEvent;
  12.     /**
  13.      * 管理员消息
  14.      * @param source
  15.      * @return
  16.      */
  17.     public int produceAdmin(String source) {
  18.         TwoChannelEvent event = new TwoChannelEvent(source);
  19.         adminEvent.fire(event);
  20.         return event.getNum();
  21.     }
  23.     /**
  24.      * 普通消息
  25.      * @param source
  26.      * @return
  27.      */
  28.     public int produceNormal(String source) {
  29.         TwoChannelEvent event = new TwoChannelEvent(source);
  30.         normalEvent.fire(event);
  31.         return event.getNum();
  32.     }
  33. }
  • 上述代码有以下两点需要注意
  1. 注入了两个Event实例adminEvent和normalEvent，它们的类型一模一样，但是分别用Admin和Normal

注解修饰，相当于为它们添加了不同的标签，在消费的时候也可以用这两个注解来过滤

  1. 发送代码并无特别之处，用adminEvent.fire发出的事件，在消费的时候不过滤、或者用Admin过滤，这两种方式都能收到
  • 接下来看消费事件的代码TwoChannelConsumer.java，有几处要注意的地方稍后会提到
  1. @ApplicationScoped
  2. public class TwoChannelConsumer {
  4.     /**
  5.      * 消费管理员事件
  6.      * @param event
  7.      */
  8.     public void adminEvent(@Observes @Admin TwoChannelEvent event) {
  9.         Log.infov("receive admin event, {0}", event);
  10.         // 管理员的计数加两次，方便单元测试验证
  11.         event.addNum();
  12.         event.addNum();
  13.     }
  15.     /**
  16.      * 消费普通用户事件
  17.      * @param event
  18.      */
  19.     public void normalEvent(@Observes @Normal TwoChannelEvent event) {
  20.         Log.infov("receive normal event, {0}", event);
  21.         // 计数加一
  22.         event.addNum();
  23.     }
  25.     /**
  26.      * 如果不用注解修饰，所有TwoChannelEvent类型的事件都会在此被消费
  27.      * @param event
  28.      */
  29.     public void allEvent(@Observes TwoChannelEvent event) {
  30.         Log.infov("receive event (no Qualifier), {0}", event);
  31.         // 计数加一
  32.         event.addNum();
  33.     }
  34. }
  • 上述代码有以下两处需要注意
  1. 消费事件的方法，除了Observes注解，再带上Admin，这样此方法只会消费Admin修饰的Event发出的事件
  2. allEvent只有Observes注解，这就意味着此方法不做过滤，只要是TwoChannelEvent类型的同步事件，它都会消费
  3. 为了方便后面的验证，在消费Admin事件时，计数器执行了两次，而Normal事件只有一次，这样两种事件的消费结果就不一样了
  • 以上就是同一事件类在多个场景被同时使用的代码了，接下来写单元测试验证
  1. @QuarkusTest
  2. public class EventTest {
  3.   
  4.     @Inject
  5.     TwoChannelWithTwoEvent twoChannelWithTwoEvent;
  7.     @Test
  8.     public void testTwoChnnelWithTwoEvent() {
  9.         // 对管理员来说，
  10.         // TwoChannelConsumer.adminEvent消费时计数加2，
  11.         // TwoChannelConsumer.allEvent消费时计数加1，
  12.         // 所以最终计数是3
  13.         Assertions.assertEquals(3, twoChannelWithTwoEvent.produceAdmin("admin"));
  15.         // 对普通人员来说，
  16.         // TwoChannelConsumer.normalEvent消费时计数加1，
  17.         // TwoChannelConsumer.allEvent消费时计数加1，
  18.         // 所以最终计数是2
  19.         Assertions.assertEquals(2, twoChannelWithTwoEvent.produceNormal("normal"));
  20.     }
  21. }
  • 执行单元测试顺利通过，如下图

小优化，不需要注入多个Event实例

  • 刚才的代码虽然可以正常工作，但是有一点小瑕疵：为了发送不同事件，需要注入不同的Event实例，如下图红框，如果事件类型越来越多，注入的Event实例岂不是越来越多？

  • quarkus提供了一种缓解上述问题的方式，再写一个发送事件的类TwoChannelWithSingleEvent.java，代码中有两处要注意的地方稍后会提到
  1. /**
  2.  * @author will
  3.  * @email zq2599@gmail.com
  4.  * @date 2022/4/3 10:16
  5.  * @description 用同一个事件结构体TwoChannelEvent，分别发送不同业务类型的事件
  6.  */
  7. @ApplicationScoped
  8. public class TwoChannelWithSingleEvent {
  10.     @Inject
  11.     Event<TwoChannelEvent> singleEvent;
  12.     
  13.     /**
  14.      * 管理员消息
  15.      * @param source
  16.      * @return
  17.      */
  18.     public int produceAdmin(String source) {
  19.         TwoChannelEvent event = new TwoChannelEvent(source);
  21.         singleEvent.select(new AnnotationLiteral<Admin>() {})
  22.                    .fire(event);
  24.         return event.getNum();
  25.     }
  27.     /**
  28.      * 普通消息
  29.      * @param source
  30.      * @return
  31.      */
  32.     public int produceNormal(String source) {
  33.         TwoChannelEvent event = new TwoChannelEvent(source);
  35.         singleEvent.select(new AnnotationLiteral<Normal>() {})
  36.                 .fire(event);
  38.         return event.getNum();
  39.     }
  40. }
  • 上述发送消息的代码，有以下两处需要注意
  1. 不论是Admin事件还是Normal事件，都是用singleEvent发送的，如此避免了事件类型越多Event实例越多的情况发生
  2. 执行fire方法发送事件前，先执行select方法，入参是AnnotationLiteral的匿名子类，并且通过泛型指定事件类型，这和前面TwoChannelWithTwoEvent类发送两种类型消息的效果是一样的
  • 既然用select方法过滤和前面两个Event实例的效果一样，那么消费事件的类就不改动了
  • 写个单元测试来验证效果
  1. @QuarkusTest
  2. public class EventTest {
  3.     @Inject
  4.     TwoChannelWithSingleEvent twoChannelWithSingleEvent;
  6.     @Test
  7.     public void testTwoChnnelWithSingleEvent() {
  8.         // 对管理员来说，
  9.         // TwoChannelConsumer.adminEvent消费时计数加2，
  10.         // TwoChannelConsumer.allEvent消费时计数加1，
  11.         // 所以最终计数是3
  12.         Assertions.assertEquals(3, twoChannelWithSingleEvent.produceAdmin("admin"));
  14.         // 对普通人员来说，
  15.         // TwoChannelConsumer.normalEvent消费时计数加1，
  16.         // TwoChannelConsumer.allEvent消费时计数加1，
  17.         // 所以最终计数是2
  18.         Assertions.assertEquals(2, twoChannelWithSingleEvent.produceNormal("normal"));
  19.     }
  20. }
  • 如下图所示，单元测试通过，也就说从消费者的视角来看，两种消息发送方式并无区别

事件元数据

  • 在消费事件时，除了从事件对象中取得业务数据（例如MyEvent的source和consumeNum字段），有时还可能需要用到事件本身的信息，例如类型是Admin还是Normal、Event对象的注入点在哪里等，这些都算是事件的元数据
  • 为了演示消费者如何取得事件元数据，将TwoChannelConsumer.java的allEvent方法改成下面的样子，需要注意的地方稍后会提到
  1. public void allEvent(@Observes TwoChannelEvent event, EventMetadata eventMetadata) {
  2.         Log.infov("receive event (no Qualifier), {0}", event);
  4.         // 打印事件类型
  5.         Log.infov("event type : {0}", eventMetadata.getType());
  7.         // 获取该事件的所有注解
  8.         Set<Annotation> qualifiers = eventMetadata.getQualifiers();
  10.         // 将事件的所有注解逐个打印
  11.         if (null!=qualifiers) {
  12.             qualifiers.forEach(annotation -> Log.infov("qualify : {0}", annotation));
  13.         }
  15.         // 计数加一
  16.         event.addNum();
  17. }
  • 上述代码中，以下几处需要注意
  1. 给allEvent方法增加一个入参，类型是EventMetadata，bean容器会将事件的元数据设置到此参数
  2. EventMetadata的getType方法能取得事件类型
  3. EventMetadata的getType方法能取得事件的所有修饰注解，包括Admin或者Normal
  • 运行刚才的单元测试，看修改后的allEvent方法执行会有什么输出，如下图，红框1打印出事件是TwoChannelEvent实例，红框2将修饰事件的注解打印出来了，包括发送时修饰的Admin

  • 至此，事件相关的学习和实战就完成了，进程内用事件可以有效地解除模块间的耦合，希望本文能给您一些参考

你不孤单，欣宸原创一路相伴

  1. Java系列
  2. Spring系列
  3. Docker系列
  4. kubernetes系列
  5. 数据库+中间件系列
  6. DevOps系列

《新程序员》：云原生和全面数字化实践

50位技术专家共同创作，文字、视频、音频交互阅读

版权说明 : 本文为转载文章, 版权归原作者所有 版权申明

原文链接 : https://xinchen.blog.csdn.net/article/details/124069627

内容来源于网络，如有侵权，请联系作者删除！

quarkus云原生java服务端java云原生

相关文章

    查看更多

    热门标签

    更多
    JavaquerypythonNode开发语言requestUtil数据库Table后端算法LoggerMessageElementParser

    最新文章

    更多
    • 蜀ICP备13028337号-1 大数据知识库 https://www.saoniuhuo.com © All rights reserved
    • 本站内容来源互联网,如果侵犯您的权益请联系我们删除, 联系方式：448109455@qq.com