# spring-cloud-demo

**Repository Path**: achuanle/spring-cloud-demo

## Basic Information

- **Project Name**: spring-cloud-demo
- **Description**: Spring Cloud Demo
- **Primary Language**: Java
- **License**: Not specified
- **Default Branch**: master
- **Homepage**: None
- **GVP Project**: No

## Statistics

- **Stars**: 0
- **Forks**: 0
- **Created**: 2018-11-26
- **Last Updated**: 2020-12-19

## Categories & Tags

**Categories**: Uncategorized

**Tags**: None

## README


### 模块简介

* service-api是服务API基础包。
* service-discovery 注册中心服务端（服务治理），即Eureka Server。
* service-hello是服务提供者，consumer-hello是服务消费者。含有调用端负载均衡Ribbon例子。
* product-service是服务提供者，mall-web是服务消费者。含有Rest客户端Feign例子，微服务容错保护Hystrix例子，分布式链路跟踪Sleuth例子。
* hystrix-dashboard是Hystrix（微服务容错保护）监控大盘服务。mall-web中有Feign结合Hystrix的例子，也有单独使用Hystrix的例子（在ApplicationTwo和ProductTwoController中）。
* zuul-server是API服务网关。
* zipkin-server是收集分布式服务的时间数据的分布式跟踪系统。在product-service和mall-web中只需要引入相应依赖包，然后在application.properties中做相应的配置即可。
* 实现消息驱动（Stream）开发。改造分为下面三步：
  - 安装Kafka服务器;
  - 改造product-service商品微服务，实现商品消息的发送;
  - 改造mall-web微服务，实现商品消息的监听。
  - 测试
    + 依次启动Kafka服务器、service-discovery、product-service、mall-web等服务。
    + product-service服务的消息测试端点: `curl http://127.0.0.1:2100/products/item-2 -X POST`。
* Spring Cloud Bus示例
  - 安装Kafka服务器;
  - 改造product-service商品微服务，实现商品事件的发送;
    - 改造mall-web微服务，实现商品事件的监听。
    - 测试
      + 依次启动Kafka服务器、service-discovery、product-service、mall-web等服务。
      + product-service服务的消息测试端点: `curl http://127.0.0.1:2100/products/item-2 -X POST`。

![基于消息驱动开发(Stream)架构图](./基于消息驱动开发(Stream)架构图.png)

### Spring Cloud简介

Spring Cloud是一系列框架的集合，其基于Spring Boot的开发便利性巧妙地简化了分布式系统基础设施的开发，构建了服务治理(发现注册)、配置中心、消息总线、负载均衡、断路器、数据监控、分布式会话和集群状态管理等功能，为我们提供一整套企业级分布式云应用的完美解决方案。

Spring Cloud包含了多个子项目（针对分布式系统中涉及的多个不同开源产品），比如：Spring Cloud Config、Spring Cloud Netflix、Spring Cloud CloudFoundry、Spring Cloud AWS、Spring Cloud Security、Spring Cloud Commons、Spring Cloud Zookeeper、Spring Cloud CLI等项目。这些项目是Spring将目前各家公司开发的比较成熟、经得起实际考验的服务框架组合起来，通过Spring Boot风格进行再封装屏蔽掉了复杂的配置和实现原理，最终给我们开发者留出了一套简单易懂、易部署和易维护的分布式系统开发工具包。

### Spring Cloud 具有的特性

Spring Cloud 具有特性，以及适用于哪些场景等包含：

* 基于版本的分布式配置管理
* 服务注册与发现
* 路由
* 服务之间调用（依赖）
* 负载均衡
* 断路器
* 全局锁（分布式锁）
* 选主以及集群状态管理
* 分布式消息服务

Spring Cloud的核心是服务治理。而服务治理主要通过整合Netflix的相关产品来实现这方面的功能，也就是Spring Cloud Netflix，在该项目中包括用于服务注册和发现的Eureka，调用断路器Hystrix，调用端负载均衡Ribbon，Rest客户端Feign，智能服务路由Zuul，用于监控数据收集和展示的Spectator、Servo、Atlas，用于配置读取的Archaius和提供Controller层Reactive封装的RxJava。除此之外，针对Feign和RxJava并不是Netiflix的产品，但也被整合到了Spring Cloud Netflix中。

### Spring Boot项目配置加载顺序

* 命令行参数
* SPRING_APPLICATION_JSON 参数
* 从 java:comp/env 加载 JNDI 属性
* Java 系统属性 （System.getProperties()）
* 操作系统环境变量
* 如果有使用 random.* 属性配置，则使用 RandomValuePropertySource 产生
* 外部特定应用配置文件 例如：application-{profile}.properties 或者 YAML variants
* 内部特定应用配置文件 例如：application-{profile}.properties 或者 YAML variants
* 外部应用配置文件 例如：application.properties 或者 YAML variants
* 内部应用配置文件 例如：application.properties 或者 YAML variants
* 加载 @Configuration 类的 @PropertySource 或者 @ConfigurationProperties 指向的配置文件
* 默认配置，通过 SpringApplication.setDefaultProperties 设置

Spring Boot 配置文件。有些配置不能够配置在application.properties文件中而要配置在bootstrap.properties文件中，因为在Spring Boot启动时有`引导上下文`和`应用上下文`的概念，只有将配置服务器信息定义在引导上下文中，才能够获取到配置信息。否则，服务启动时会报找不到变量定义的错误。

### Spring Boot 启动时指定参数

* 指定配置文件。java -jar xxx.jar --spring.config.location=classpath:/default.properties,classpath:/override.properties

* 指定端口。java -jar xxx.jar --server.port=2300

### Spring Boot 修改默认端口号

* 修改application.properties中的server.port=8080。ServerProperties类读取Spring Boot的默认配置文件application.properties的值注入到bean里。

* 在启动类中修改。

* 使用命令行参数。java -jar xxx.jar --server.port=8080。

* 使用虚拟机参数。java Application -Dserver.port=8080。

### 使用 Profile 区分环境

Spring Boot 可以在配置文件、Java代码类、日志配置中来配置 profile 区分不同环境执行不同的结果。启动时通过 --spring.profiles.active=xxx 命令行参数指定要加载的配置。

#### 使用配置文件

以 application.properties 为例，通过文件名来区分环境 application-{profile}.properties。

application.properties

    ```
    app.name=MyApp
    server.port=8080
    spring.profiles.active=dev
    ```

application-dev.properties

    ```
    server.port=8081
    ```

application-test.properties

    ```
    server.port=8082
    ```

在启动程序的时候通过添加`–spring.profiles.active={profile}`来指定具体使用的配置。例如，执行`java -jar demo.jar --spring.profiles.active=dev`那么上面3个文件中的内容将被如何应用？Spring Boot会先加载默认的配置文件，然后使用具体指定的profile中的配置去覆盖默认配置。

* app.name 只存在于默认配置文件 application.properties 中，因为指定环境中不存在同样的配置，所以该值不会被覆盖。
* server.port 默认为8080，但是我们指定了环境后，将会被覆盖。如果指定test环境，server.port 则为 8082。
* spring.profiles.active 默认指定dev环境，如果我们在运行时指定 --spring.profiles.active=test 那么将应用 test 环境，最终 server.port 的值为 8082。

#### Spring Boot 可配置选项

https://docs.spring.io/spring-boot/docs/2.0.6.RELEASE/reference/htmlsingle/#common-application-properties

各个 jar 提供的可配置选项在 META-INF/spring-configuration-metadata.json 中。

#### Java 类中使用 @Profile 注解

下面2个不同的类实现了同一个接口，@Profile注解指定了具体环境。

```
public interface SendMessage {
    public void send();// 发送短信方法定义
}

// Dev 环境实现类
@Component
@Profile("dev")
public class DevSendMessage implements SendMessage {
    @Override
    public void send() {
        System.out.println(">>>>>>>>Dev Send()<<<<<<<<");
    }
}

// Test 环境实现类
@Component
@Profile("test")
public class StgSendMessage implements SendMessage {
    @Override
    public void send() {
        System.out.println(">>>>>>>>Stg Send()<<<<<<<<");
    }
}

// 启动类
@SpringBootApplication
public class Application {
    @Value("${app.name}")
    private String name;

    @Autowired
    private SendMessage sendMessage;

    @PostConstruct
    public void init(){
        sendMessage.send();// 会根据 profile 指定的环境实例化对应的类
    }
}
```

#### logback-spring.xml 也支持用节点来支持区分

注意：文件名不要用 logback.xml，请使用 logback-spring.xml。

```
<?xml version="1.0" encoding="UTF-8"?>
<configuration>
    <include resource="org/springframework/boot/logging/logback/base.xml" />
    <logger name="org.springframework.web" level="INFO"/>

    <springProfile name="default">
        <logger name="org.springboot.sample" level="TRACE" />
    </springProfile>

    <springProfile name="dev">
        <logger name="org.springboot.sample" level="DEBUG" />
    </springProfile>

    <springProfile name="staging">
        <logger name="org.springboot.sample" level="INFO" />
    </springProfile>
</configuration>
```

### Spring Boot Shell 脚本

* start.sh

```
#!/bin/sh

rm -f tpid
nohup java -jar /data/app/myapp.jar --spring.profiles.active=pro > /dev/null 2>&1 &
echo $! > tpid
```

* stop.sh

```
#!/bin/sh

tpid=`cat tpid | awk '{print $1}'`
tpid=`ps -aef | grep $tpid | awk '{print $2}' |grep $tpid`
if [ ${tpid} ]; then
    kill -9 $tpid
fi
```

* check.sh

```
#!/bin/sh

tpid=`cat tpid | awk '{print $1}'`
tpid=`ps -aef | grep $tpid | awk '{print $2}' |grep $tpid`
if [ ${tpid} ]; then
    echo App is Running.
else
    echo App is NOT Running.
fi
```

* kill.sh

```
#!/bin/sh

# kill -9 `ps -ef|grep 项目名称|awk '{print $2}'`
kill -9 `ps -ef|grep demo|awk '{print $2}'`
```


### Spring Boot Embed Tomcat Server 常用配置

```
# 项目contextPath，一般在正式发布版本中，我们不配置
server.context-path=/myspringboot

# 错误页，指定发生错误时，跳转的URL。请查看BasicErrorController源码便知
server.error.path=/error

# 服务端口
server.port=9090

# session最大超时时间(分钟)，默认为30
server.session-timeout=60

# 该服务绑定IP地址，启动服务器时如本机不是该IP地址则抛出异常启动失败，只有特殊需求的情况下才配置
server.address=192.168.16.11

# tomcat最大线程数，默认为200
server.tomcat.max-threads=500

# tomcat的URI编码
server.tomcat.uri-encoding=UTF-8

# 存放Tomcat的日志、Dump等文件的临时文件夹，默认为系统的tmp文件夹（如：C:\Users\achuan\AppData\Local\Temp）
server.tomcat.basedir=/data/logs

# 打开Tomcat的Access日志，并可以设置日志格式的方法：
server.tomcat.access-log-enabled=true

#server.tomcat.access-log-pattern=

# accesslog目录，默认在basedir/logs
#server.tomcat.accesslog.directory=

# 日志文件目录
logging.path=H:/springboot-tomcat-tmp

# 日志文件名称，默认为spring.log
logging.file=myapp.log
```

### 分布式链路跟踪(Sleuth)

当我们进行微服务架构开发时，通常会根据业务来划分微服务，各业务之间通过REST进行调用。一个用户操作，可能需要很多微服务的协同才能完成，如果在业务调用链路上任何一个微服务出现问题或者网络超时，都会导致功能失败。随着业务越来越多，对于微服务之间的调用链的分析会越来越复杂。

Spring Cloud Sleuth为服务之间调用提供链路追踪。通过Sleuth可以很清楚的了解到一个服务请求经过了哪些服务，每个服务处理花费了多长。从而让我们可以很方便的理清各微服务间的调用关系。此外Sleuth可以帮助我们：

* 耗时分析: 通过Sleuth可以很方便的了解到每个采样请求的耗时，从而分析出哪些服务调用比较耗时;
* 可视化错误: 对于程序未捕捉的异常，可以通过集成Zipkin服务界面上看到;
* 链路优化: 对于调用比较频繁的服务，可以针对这些服务实施一些优化措施。

### 基于消息驱动开发(Spring Cloud Stream)

基于消息驱动的开发几乎成了微服务架构下必备开发方式之一。这是因为：
第一原来传统单体架构开发中的接口调用开发已经在微服务架构下不存在；
第二微服务架构的开发要求降低各微服务直接的依赖耦合，一旦我们在某个微服务中直接调用另外一个微服务，那么这两个微服务就会通过依赖产生了强耦合；
第三微服务的自治原则也强烈要求各微服务之间不能够互相调用。因此，在微服务架构开发中基于消息驱动的开发成为了一种必然趋势。

#### Spring Cloud Stream简介

从Spring Cloud Stream核心原理上来说，Stream提供了一个与消息中间件进行消息收发的抽象层，这个也是Spring所擅长的。通过该抽象层剥离了业务中消息收发与实际所使用中间件直接的耦合，从而使得我们可以轻松与各种消息中间件对接，也可以很简单的就可以实现所使用的消息中间件的更换。

#### Stream应用模型及核心概念

![Stream应用模型](./Stream应用模型.png)

在该模型图上有如下几个核心概念:

* Source: 当需要发送消息时，我们就需要通过Source，Source将会把我们所要发送的消息(POJO对象)进行序列化（默认转换成JSON格式字符串），然后将这些数据发送到Channel中；

* Sink: 当我们需要监听消息时就需要通过Sink来，Sink负责从消息通道中获取消息，并将消息反序列化成消息对象(POJO对象)，然后交给具体的消息监听处理进行业务处理；

* Channel: 消息通道是Stream的抽象之一。通常我们向消息中间件发送消息或者监听消息时需要指定主题（Topic）/消息队列名称，但这样一旦我们需要变更主题名称的时候需要修改消息发送或者消息监听的代码，但是通过Channel抽象，我们的业务代码只需要对接Channel就可以了，具体这个Channel对应的是那个主题，就可以在配置文件中来指定，这样当主题变更的时候我们就不用对代码做任何修改，从而实现了与具体消息中间件的解耦；

* Binder: Stream中另外一个抽象层。通过不同的Binder可以实现与不同消息中间件的整合，比如上面的示例我们所使用的就是针对Kafka的Binder，通过Binder提供统一的消息收发接口，从而使得我们可以根据实际需要部署不同的消息中间件，或者根据实际生产中所部署的消息中间件来调整我们的配置。

#### Stream消息发送与监听

Stream提供了下面三个注解:

* @Input: 创建一个消息输入通道，用于消息监听;
* @Output: 创建一个消息输出通道，用于消息发送;
* @EnableBinding: 建立与消息通道的绑定。

在使用时可以通过@Input和@Output创建多个通道，使用这两个注解创建通道非常简单，只需要将它们分别注解到接口的相应方法上即可，而不需要具体来实现该注解。当启动Stream框架时，就会根据这两个注解通过动态代码生成技术生成相应的实现，并注入到Spring应用上下文中，这样我们就可以在代码中直接使用。

##### Output注解

对于@Output注解来说，所注解的方法的返回值必须是MessageChannel，MessageChannel也就是具体消息发送的通道。比如下面的代码：

```
public interface ProductSource {
    @Output
    MessageChannel hotProducts();

    @Output
    MessageChannel selectedProducts();
}
```

这样，就可以通过ProductSource所创建的消息通道来发送消息了。

##### Input注解

对于@Input注解来说，所注解的方法的返回值必须是SubscribableChannel，SubscribableChannel也就是消息监听的通道。比如下面的代码：

```
public interface ProductSink {
    @Input
    SubscribableChannel productOrders();
}
```

这样，就可以通过ProductSink所创建的消息通道来监听消息了。

##### 开箱即用的Source和Sink接口

在代码中使用了Source、Sink，那么这两个接口和Output、Input注解什么关系呢？让我们来看一下这两个接口的源码：

```
// Source源码
public interface Source {
    String OUTPUT = "output";

    @Output(Source.OUTPUT)
    MessageChannel output();
}

// Sink源码
public interface Sink {
    String INPUT = "input";

    @Input(Sink.INPUT)
    SubscribableChannel input();
}
```

@Input和@Output是Stream核心应用的注解，而Source和Sink只不过是Stream为我们所提供开箱即用的两个接口而已，有没有这两个接口我们都可以正常使用Stream。

此外，Stream还提供了一个开箱即用的接口Processor：

```
public interface Processor extends Source, Sink {
}
```

也就是说Processor同时可以作为消息发送和消息监听，这种接口在我们开发消息管道类型应用时会非常有用。

#### 自定义消息通道名称

消息通道是Stream的一个抽象，通过该抽象可以避免与消息中间件具体的主题耦合，那么到底是怎么一回事呢？从Source和Sink源码中可以看到，所注解的@Output和@Input注解中都有一个参数，分别为output和input，为Stream中所使用的消息通道名称。再观察一下product-service和mall-web中的配置：

```
# product-service中的配置
spring.cloud.stream.bindings.output.destination=product-topic
spring.cloud.stream.bindings.output.content-type=application/json

# mall-web中的配置
spring.cloud.stream.bindings.input.destination=product-topic
spring.cloud.stream.bindings.input.content-type=application/json
spring.cloud.stream.bindings.input.group=mallWebGroup
```

从配置中可以看到destination属性的配置，分别指定了output和inout，也就是Stream中所使用的消息通道名称。因此，我们可以通过这两个注解来分别设置消息通道的名称，比如：

```
public interface ProductProcessor {
    @Output("pmsoutput")
    MessageChannel productOutput();

    @Input("pmsinput")
    SubscribableChannel input();
}
```

这样，当我们使用ProductProcessor接口来实现消息发送和监听的时就需要在配置文件中配置如下：

```
# 消息发送
spring.cloud.stream.bindings.pmsoutput.destination=product-topic
spring.cloud.stream.bindings.pmsoutput.content-type=application/json

# 消息监听
spring.cloud.stream.bindings.pmsinput.destination=product-topic
spring.cloud.stream.bindings.pmsinput.content-type=application/json
spring.cloud.stream.bindings.pmsinput.group=mallWebGroup
```

#### 将消息通道绑定到消息中间件

消息发送通道和监听通道都创建好了，那么将它们对接到具体的消息中间件就可以完成消息的发送和监听功能了，而@EnableBinding注解就是用来实现该功能。具体使用方式如下：

```
// 实现发送的绑定
@EnableBinding(Source.class)
public class Application {

}

// 实现监听的绑定
@EnableBinding(Sink.class)
public class ProductMsgListener {

}
```

需要说明的是，@EnableBinding可以同时绑定多个接口，例如：@EnableBinding(value={ProductSource.class, ProductSink.class})。

#### 直接使用通道

一个消息发送的代码：

```
private org.springframework.cloud.stream.messaging.Source source;

protected void sendMsg(String action, String itemCode) {// action为消息类型，即增删改等动作
    ProductMsg productMsg = new ProductMsg(action, itemCode);
    logger.debug("发送商品消息:{} ", productMsg);

    // 发送消息
    source.output().send(MessageBuilder.withPayload(productMsg).build());
}
```

既然@Output所提供的MessageChannel才是最终消息发送时使用的，那么我们是否可以直接使用呢？的确是可以的，上面的代码我们可以更改成如下：

```
@Service
public class ProductService {
    protected Logger logger = LoggerFactory.getLogger(ProductService.class);

    private MessageChannel output;
    private List<ProductDto> productList;

    @Autowired
    public ProductService(MessageChannel output) {
        this.output = output;
        this.productList = buildProducts();
    }

    /**
     * 具体消息发送的实现
     * @param action 消息类型，即增删改等动作
     * @param itemCode 商品货号
     */
    protected void sendMsg(String action, String itemCode) {
        ProductMsg productMsg = new ProductMsg(action, itemCode);
        logger.debug("发送商品消息:{} ", productMsg);

        // 发送消息
        output.send(MessageBuilder.withPayload(productMsg).build());
    }
}
```

默认使用MessageChannel的名称做为ID，但是如果我们在@Output注解中增加了名称定义，例如：

```
public interface ProductSource {
    @Output("pmsoutput")
    MessageChannel output();
}
```

那么这个时候Stream会使用pmsoutput作为Bean的Id，而我们的代码也需要为如下：

```
@Autowired
public ProductService(@Qualifier("pmsoutput") MessageChannel output) {
    this.output = output;
    this.productList = buildProducts();
}
```

### ESB(Enterprise Service Bus)

企业服务总线（Enterprise Service Bus，ESB）的概念是从面向服务的架构(Service Oriented Architecture， SOA)发展而来。SOA描述了一种IT基础设施的应用集成模型；其中的构件集是以一种定义清晰的层次化结构来相互耦合。一个ESB是一个预先组装的SOA实现，它包含了实现SOA分层目标所必需的基础功能部件。

在企业计算领域，企业服务总线是指由中间件基础设施产品技术实现的、通过事件驱动和基于XML消息引擎，为更复杂的面向服务的架构提供的软件架构的构造物。企业服务总线通常在企业消息系统上提供一个抽象层，使得集成架构师能够不用编码而是利用消息的价值完成集成工作。

企业服务总线提供可靠消息传输，服务接入，协议转换，数据格式转换，基于内容的路由等功能，屏蔽了服务的物理位置，协议和数据格式。

其中，最重要的一句就是：`企业服务总线通常在企业消息系统上提供一个抽象层，使得集成架构师能够不用编码而是利用消息的价值完成集成工作`。 通俗一点来讲就是`企业服务总线是架构在消息中间件之上的另外一个抽象层`，使得我们可以不用关心消息相关的处理就可以完成业务逻辑的处理。

到这里是不是有点突然明白Spring Cloud Bus 和 Spring Cloud Stream之间的关系了，刚开始接触这两个组件时，大部分都会迷惑到底这两者有什么区别？它们又有什么联系？

Stream通过对消息中间件进行抽象封装，提供一个统一的接口供我们发送和监听消息，而Bus则是在Stream基础之上再次进行抽象封装，使得我们可以在不用理解消息发送、监听等概念的基础上使用消息来完成业务逻辑的处理。那么Spring Cloud Bus是如何为我们实现的呢？一句话概括就是事件机制。

### Spring的事件机制

在Spring框架中有一个事件机制，该机制是一个观察者模式的实现。观察者模式建立一种对象与对象之间的依赖关系，当一个对象(称之为：被观察目标)发生改变时将自动通知其它对象(称之为：观察者)，这些观察者将做出相应的反应。一个被观察目标可以对应多个观察者，而且这些观察者之间没有相互联系，可以根据需要增加和删除观察者，使得系统更易于扩展。通过Spring事件机制可以达到如下目的：

* 应用模块之间的解耦;
* 对同一种事件可以根据需要定义多种处理方式;
* 对主线应用不干扰，是一个极佳的开闭原则(OCP)实践。

当我们在应用中引入事件机制时需要借助Spring中以下接口或抽象类：

* ApplicationEventPublisher: 这是一个接口，用来发布一个事件;
* ApplicationEvent: 这是一个抽象类，用来定义一个事件;
* ApplicationListener<E extends ApplicationEvent>: 这是一个接口，实现事件的监听。

其中Spring应用的上下文ApplicationContext默认是实现了ApplicationEventPublisher接口，因此在发布事件时我们可以直接使用ApplicationContext.publishEvent()方法来发送。

### 一个典型的Spring事件发送与监听示例

#### 定义事件

比如，我们定义一个用户事件。

```
import cn.achuan.spring.springcloud.demo.user.User;
import com.google.common.base.MoreObjects;
import org.springframework.context.ApplicationEvent;

/**
 * 用户事件
 */
public class UserEvent extends ApplicationEvent {
    /** 消息类型：更新用户，值为: {@value} */
    public static final String U_UPDATE = "update";

    private String action;
    private User user;

    public UserEvent(User user) {
        super(user);
        this.user = user;
    }

    public UserEvent(User user, String action) {
        super(user);
        this.action = action;
        this.user = user;
    }

    @Override
    public String toString() {
        return MoreObjects.toStringHelper(this)
                .add("action", this.getAction())
                .add("user", this.getUser()).toString();
    }

    public String getAction() {
        return action;
    }
    public void setAction(String action) {
        this.action = action;
    }

    public User getUser() {
        return user;
    }
    public void setUser(User user) {
        this.user = user;
    }
}
```

#### 定义监听

定义一个用户事件监听器，当用户变更时做相应处理。用户事件监听比较简单，只需要实现ApplicationListener接口，进行相应处理即可。

```
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.context.ApplicationListener;
import org.springframework.stereotype.Component;

/**
 * 用户事件监听
 */
@Component
public class UserEventListener implements ApplicationListener<UserEvent> {
    private static final Logger logger = LoggerFactory.getLogger(UserEventListener.class);

    @Override
    public void onApplicationEvent(UserEvent userEvent) {
        logger.debug("收到用户事件:{} ", userEvent);
        // 实现具体的业务处理
    }
}
```

#### 发送消息

发送消息比较简单，我们也可以直接在Event中实现，比如我们将上面UserEvent更改为如下。

```
import cn.achuan.spring.demo.ext.ApplicationContextHolder;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationEvent;

public class UserEvent extends ApplicationEvent {
    /**
     * 发布事件
     */
    public void fire() {
        ApplicationContext context = ApplicationContextHolder.getApplicationContext();
        if(context != null) {
            logger.debug("发布事件：{}", this);
            context.publishEvent(this);
        }else{
            logger.warn("未获取到当前Spring上下文信息，无法发布事件");
        }
    }
}
```

那么我们就可以在需要的地方通过下面的代码来发布事件了：

```
new UserEvent(user, UserEvent.U_UPDATE).fire();
```

#### ApplicationContextHolder代码

```
import org.springframework.beans.BeansException;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.context.ApplicationListener;
import org.springframework.context.event.ContextRefreshedEvent;

public class ApplicationContextHolder implements ApplicationContextAware, ApplicationListener<ContextRefreshedEvent> {

    private static ApplicationContext applicationContext;
    private static volatile boolean refreshed;

    @Override
    public void onApplicationEvent(ContextRefreshedEvent event) {
        refreshed = true;
    }

    public static boolean hasApplicationContext() {
        return (refreshed && applicationContext != null);
    }

    public static ApplicationContext getApplicationContext() {
        return applicationContext;
    }

    @Override
    public void setApplicationContext(ApplicationContext context) throws BeansException {
        applicationContext = context;
    }

    public static boolean isRefreshed() {
        return refreshed;
    }
}
```

### Spring Cloud Bus机制

了解了Spring的事件机制，那么Spring Cloud Bus又是如何将事件机制和Stream结合在一起的呢？总起来说机制如下：

1. 在需要发布或者监听事件的应用中增加@RemoteApplicationEventScan注解，通过该注解就可以启动Stream中所说的消息通道的绑定；
2. 对于事件发布，则需要继承ApplicationEvent的扩展类RemoteApplicationEvent，当通过ApplicationContext.publishEvent()发布此种类型的事件时，Spring Cloud Bus就会对所要发布的事件进行包装，形成一个我们所熟知的消息，然后通过默认的springCloudBus消息通道发送到消息中间件；
3. 对于事件监听者则不需要进行任何变更，仍旧按照上面的方式就可以实现消息的监听。但，需要注意的一点就是在消费的微服务工程中也必须定义第2步所定义的事件，并且需要保障全类名一致(如果不一致，则需要做一点工作)。

就是这么简单。通过Bus我们就可以像编写单体架构应用一样进行开发，而不需要关系什么消息中间件、主题、消息、通道呀等等一大堆概念。

#### 构建事件

```
RemoteApplicationEvent(Object source, String originService, String destinationService)
```

在构建一个事件时需要指定originService和destinationService。对于事件发布者来说originService就是自己，而destinationService则是指将事件发布到那些微服务实例。destinationService配置的格式为：{serviceId}:{appContextId}，在配置时serviceId和appContextId可以使用通配符，如果这两个变量都使用通配符的话`(*:**)`，则事件将发布到所有的微服务实例。如只省略appContextId，则事件只会发布给指定微服务的所有实例，如：userservice:**，则只会将事件发布给userservice微服务。

```
/**
 * 商品事件。
 * 当商品配置变更时，如：修改、删除等，就需要构建一个商品事件。
 */
public class ProductEvent extends RemoteApplicationEvent {
    /** 消息类型：更新商品，值为: {@value} */
    public static final String P_UPDATE = "update";
    /** 消息类型：删除商品，值为: {@value} */
    public static final String P_DELETE = "delete";

    private String action;
    private String itemCode;

    public ProductEvent() {
        super();
    }

    public ProductEvent(Object source, String originService, String destinationService, String action, String itemCode) {
        super(source, originService, destinationService);
        this.action = action;
        this.itemCode = itemCode;
    }

    @Override
    public String toString() {
        return MoreObjects.toStringHelper(this)
                .add("action", this.getAction())
                .add("itemCode", this.getItemCode()).toString();
    }

    public String getAction() {
        return action;
    }
    public void setAction(String action) {
        this.action = action;
    }

    public String getItemCode() {
        return itemCode;
    }
    public void setItemCode(String itemCode) {
        this.itemCode = itemCode;
    }
}
```

#### 事件发布

```
@Service
public class ProductService {
    /**
     * 保存或更新商品信息
     * @param product
     * @return
     */
    public Product save(Product product) {
        // 实现商品保存处理
        // 发布商品消息
        fireEvent(ProductEvent.P_UPDATE, product);
        return product;
    }

    /**
     * 发布商品事件
     * @param eventAction
     * @param product
     */
    protected void fireEvent(String eventAction, Product product) {
        ProductEvent productEvent = new ProductEvent(
                product,
                ApplicationContextHolder.getApplicationContext().getId(),
                "*:**",
                eventAction,
                product.getItemCode());

        // 发布事件
        RemoteApplicationEventPublisher.publishEvent(productEvent);
    }
}
```

#### RemoteApplicationEventPublisher代码

```
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.cloud.bus.event.RemoteApplicationEvent;
import org.springframework.context.ApplicationContext;


/**
 * 远程事件发布者
 */
public class RemoteApplicationEventPublisher {
    private static Logger logger = LoggerFactory.getLogger(RemoteApplicationEventPublisher.class);

    /**
     * 发布一个事件
     * @param event
     */
    public static void publishEvent(RemoteApplicationEvent event){
        ApplicationContext context = ApplicationContextHolder.getApplicationContext();
        if(context != null) {
            context.publishEvent(event);
            logger.debug("已发布事件:{}", event);
        }else{
            logger.warn("未获取到当前Spring上下文信息，无法发布事件");
        }
    }
}
```

#### 开启远程事件扫描

启动类添加@RemoteApplicationEventScan注解。

>注意： 远程事件必须定义在@RemoteApplicationEventScan注解所注解类的子包中，否则无法实现远程事件发布。

#### 事件监听处理

```
/**
 * 远程事件监听
 */
@Component
public class ProductEventListener implements ApplicationListener<ProductEvent> {
    @Autowired
    private ProductService productService;

    @Override
    public void onApplicationEvent(ProductEvent productEvent) {
        if (ProductEvent.P_UPDATE.equalsIgnoreCase(productEvent.getAction())) {
            logger.debug("Mall-Web微服务收到商品更新事件: {}", productEvent);
            // 重新加载商品信息
        } else if (ProductEvent.P_DELETE.equalsIgnoreCase(productEvent.getAction())) {
            logger.debug("Mall-Web微服务收到商品删除事件");
        } else {
            logger.debug("Mall-Web微服务收到未知商品事件");
        }
    }
}
```

### Spring Cloud Stream 与 Spring Cloud Bus 的选择

从代码来说的确使用Bus会更容易理解，也更容易上手。这对于当使用场合比较简单会非常好，比如：广播。典型的应用就是Config中的配置刷新，当在项目中同时引入了Config和Bus时，就可以通过/bus/refresh端点实现配置更改的广播，从而让相应的微服务重新加载配置数据。

当然，Bus简便性的另外一层含义就是不够灵活，因此具体是在项目中使用Bug还是直接使用Stream就看实际的需求了，总起来一句就是：够用就好。