(十三) Nepxion-Thunder分布式RPC集成框架 - 事件发布

Nepxion-Thunder（QQ 群 471164539）发布在https://github.com/Nepxion/

 

基于Google Guava EventBus，实现事件驱动发布框架内部事件，解除耦合；发布外部事件，进行重试补偿，异常通知(邮件或短信通知)

 

1）eventbus - 基于Google Guava的进程内发布/订阅机制，支持同步和异步事件发布
2）mobile - 基于EventBus的异常信息的短信通知，未实现
3）protocol - 基于EventBus的异常信息的发布/拦截，是其它事件通知模块的基础
4）registry - 基于EventBus的上下线通知，通过Zookeeper的Watcher事件来发布
5）smtp - 基于EventBus的Smtp协议的邮件通知，支持普通邮件和SSL认证邮件

1. 系统通知

邮件通知，一旦框架中有异常抛出，立即通过给定的邮件予以通知，

 

 

短信通知，未实现

 

2. 业务层面的异常事件拦截

ProtocolEventInterceptor可以拦截和通知如下的异常：

•  业务系统中抛出的任何异常
• 业务系统定义的超时异常（异步和同步，但不支持广播）
• 框架的中间价宕机产生异常，目前只支持Netty
• 框架服务治理产生异常，例如限流异常

业务系统只需要继承ProtocolEventInterceptor，实现onEvent方法，就可以捕获全部异常抛出

public class ServiceEventInterceptor extends ProtocolEventInterceptor {
    @Override
    protected void onEvent(ProtocolEvent event) {
        ApplicationType applicationType = event.getApplicationType();
        ActionType actionType = event.getActionType();
        ProtocolType protocolType = event.getProtocolType();
        ProtocolMessage protocolMessage = event.getProtocolMessage();
        System.out.println("--------------------收到异步事件通知--------------------");
        System.out.println("Application type=" + applicationType);
        System.out.println("Action type=" + actionType);
        System.out.println("Protocol type=" + protocolType);
        if (actionType != ActionType.SYSTEM) {
            System.out.println("Trace id=" + protocolMessage.getTraceId());
            System.out.println("Interface=" + protocolMessage.getInterface());
            System.out.println("Method=" + protocolMessage.getMethod());
            System.out.println("Parameter types=" + Arrays.asList(protocolMessage.getParameterTypes()));
            System.out.println("Parameters=" + Arrays.asList(protocolMessage.getParameters()));
        } else {
            System.out.println("From url=" + protocolMessage.getFromUrl());
            System.out.println("To url=" + protocolMessage.getToUrl());
        }
        System.out.println("Exception=" + ExceptionUtil.toExceptionString(protocolMessage.getException()));
        System.out.println("-------------------------------------------------------");
    }
}

 

框架接入比较简单，见下面的示例

<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:context="http://www.springframework.org/schema/context"
xmlns:aop="http://www.springframework.org/schema/aop"
xmlns:tx="http://www.springframework.org/schema/tx"
xmlns:mvc="http://www.springframework.org/schema/mvc"
xmlns:thunder="http://www.nepxion.com/schema/thunder"
xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-3.0.xsd
           http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context-3.0.xsd
           http://www.springframework.org/schema/aop http://www.springframework.org/schema/aop/spring-aop-3.0.xsd
           http://www.springframework.org/schema/tx http://www.springframework.org/schema/tx/spring-tx-3.0.xsd
           http://www.springframework.org/schema/mvc http://www.springframework.org/schema/mvc/spring-mvc-3.0.xsd
           http://www.nepxion.com/schema/thunder http://www.nepxion.com/schema/thunder/thunder-1.0.xsd">
   
    .......
    
    <!-- 异常的EventBus事件发布拦截 -->
    <bean id="eventInterceptor" class="com.nepxion.thunder.service.ServiceEventInterceptor"/>
</beans>