欢迎访问我的GitHub

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

本篇概览

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

1. 同步事件
2. 异步事件
3. 同一种事件类，用在不同的业务场景
4. 优化
5. 事件元数据

同步事件

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

• 接下来编码实践，先定义事件类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();
    }
}

• 上述代码有以下几点要注意：

1. 注入Event，用于发布事件，通过泛型指定事件类型是MyEvent
2. 发布同步事件很简单，调用fire即可
3. 由于是同步事件，会等待事件的消费者将消费的代码执行完毕后，fire方法才会返回
4. 如果消费者增加了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 myProduce