Guava工具类EventBus使用

在项目开发过程中，为了提升系统性能，增大吞吐量，有时会将服务接口设计成异步的：也就是接口先接收到客户端请求，然后将请求数据交给其他线程执行处理逻辑，这样请求线程就不会被阻塞可以继续处理其他的请求。在这种设计模式中，我们可以选择线程池来做异步处理，可以通过队列方式实现异步处理，也可以使用消息队列解耦异步处理。对于消息队列，如果项目是分布式或集群化部署，可以选择kafka、rocketmq等消息队列；如果只想在单个项目内做消息队列功能，可以选择springboot提供的EventBus；在这里推荐一个guava提供的EventBus，使用起来非常简单。
下面展示一下如何使用：

定义事件实体类，用于发布事件：这种实体类可以定义多个，对于不同消息可以发布不同的事件实体，这样对于消费者那边就可以根据实体类不同接收到不同类型的消息了

public class EventData {

    private String message;

    public EventData(String message) {
        this.message = message;
    }

    public String getMessage() {
        return message;
    }

    public void setMessage(String message) {
        this.message = message;
    }
}

定义一个事件订阅者，用于接收生产者发布的消息：在方法上加@Subscribe注解就实现了订阅功能，它就是消息的消费者，可以接收生产者的消息，可以根据方法的参数类型不同接收不同的消息内容，同样类型的消息也可以指定多个消费者，他们都可以接收到消息。在接收消息的方法上添加@AllowConcurrentEvents，通过这个注解可以实现消息的并行处理，否则消息的生产者和消息的消费者会在一个线程中执行，不能达到异步的效果。

import com.google.common.eventbus.AllowConcurrentEvents;
import com.google.common.eventbus.Subscribe;

public class GuavaListener {

    @Subscribe
    @AllowConcurrentEvents
    public void listen1(EventData event) {
        System.out.println("接收到消息监听方法1：" + event.getMessage() + "|" + Thread.currentThread().getName());
        throw new RuntimeException("exception");
    }

    @Subscribe
    @AllowConcurrentEvents
    public void listen2(String event) {
        System.out.println("接收到消息监听方法2：" + event + "|" + Thread.currentThread().getName());
    }


    @Subscribe
    public void listen3(EventData event) {
        System.out.println("接收到消息监听方法3：" + event.getMessage() + "|" + Thread.currentThread().getName());
    }

    @Subscribe
    public void listen4(String event) {
        System.out.println("接收到消息监听方法4：" + event + "|" + Thread.currentThread().getName());
    }
}

消息的生产者：以上内容都定义好了之后，就可以通过生产者发布消息了，首先要定义一个EventBus对象，它有两种方式：一种直接new EventBus()对象，该方式发布的消息生产者和消费者都在同一个线程中执行；另外一种是new AsyncEventBus()对象，该方式发布的消息可以实现生产者和消费者在不同线程中执行，但是需要配合@AllowConcurrentEvents注解。其实在同一个线程中执行也是有使用场景的，比如要实现事务功能就需要在同一个线程中。实例化EventBus后，还需要将消费者类注册进来，通过register()进行注册。以上步骤完成后就可以通过EventBus中的post()方法发送消息。

import com.google.common.eventbus.AsyncEventBus;
import com.google.common.eventbus.EventBus;
import com.google.common.eventbus.SubscriberExceptionContext;
import com.google.common.eventbus.SubscriberExceptionHandler;

import java.util.concurrent.*;

public class GuavaPushTest {

    public static void main(String[] args) {
        // 同步消息
        // EventBus eventBus = new EventBus("push");

        // 异步消息
        EventBus eventBus = new AsyncEventBus(new ThreadPoolExecutor(5, 10, 120, TimeUnit.SECONDS, new LinkedBlockingDeque<>()),
                new SubscriberExceptionHandler() {

                    @Override
                    public void handleException(Throwable throwable, SubscriberExceptionContext context) {
                        System.out.println("event: " + context.getEvent());
                    }
                });

        // 注册消费者类
        eventBus.register(new GuavaListener());
        // eventBus.register(new GuavaListener2());

        for(int i = 0; i < 10; ++i) {
            final int num = i;

            eventBus.post(new EventData("message" + num));
            eventBus.post("strMessage" + num);

            System.out.println("push ==> " + num + "|" + Thread.currentThread().getName());

//            new Thread(new Runnable() {
//                @Override
//                public void run() {
//                    eventBus.post(new GuavaEvent("message" + num));
//                    eventBus.post("strMessage" + num);
//
//                    System.out.println("push ==> " + num + "|" + Thread.currentThread().getName());
//                }
//            }, "subThread:" + i).start();
        }
    }
}