SpringCloud Bus实现自定义发布、消费事件

布满荆棘的人生 2022-12-23 09:37 249阅读 0赞

## 1. spring cloud bus介绍 ##

Spring cloud bus通过轻量消息代理连接各个分布的节点。这会用在广播状态的变化（例如配置变化）或者其他的消息指令。Spring cloud bus的一个核心思想是通过分布式的启动器对spring boot应用进行扩展，也可以用来建立一个多个应用之间的通信频道。目前唯一实现的方式是用AMQP消息代理作为通道，同样特性的设置（有些取决于通道的设置）在更多通道的文档中。看源码会发现spring cloud bus是在spring cloud stream的基础上做的封装。

## 2. 创建项目 ##

我们创建两个项目A、B使用rabbitmq作为中间件，项目A用来发布事件，项目B用来消费事件，两个项目都需要引入依赖如下：

<dependency>
        <groupId>org.springframework.cloud</groupId>
        <artifactId>spring-cloud-starter-bus-amqp</artifactId>
    </dependency>

bootstrap.yml中添加连接rabbitmq的配置：

spring:
      rabbitmq:
        host: 127.0.0.1
        port: 5672
        username: test
        password: test123

## 3. 两个项目的公共代码 ##

#### 3.1 创建事件 ####

事件需要继承RemoteApplicationEvent类：

import lombok.Getter;
    import org.springframework.cloud.bus.event.RemoteApplicationEvent;
    
    /** * @author mazhen * @className RouteEvent * @Description TODO * @date 2020/11/20 15:54 */
    public class RouteEvent extends RemoteApplicationEvent { 
        @Getter
        private String message;
        public RouteEvent() { super();}
        public RouteEvent(Object body,String originService, String destinationService, String message) { 
            super(body,originService,destinationService);
            this.message = message;
        }
        public RouteEvent(Object body,String originService, String message) { 
            super(body,originService);
            this.message = message;
        }
    }

#### 3.2 添加配置类 ####

import org.springframework.cloud.bus.jackson.RemoteApplicationEventScan;
    import org.springframework.context.annotation.Configuration;
    
    /** * @author mazhen * @className BusExtConfiguration * @Description TODO * @date 2020/11/20 16:15 */
    @Configuration
    @RemoteApplicationEventScan(basePackageClasses = RouteEvent.class)
    public class BusExtConfiguration { 
    }

@RemoteApplicationEventScan(basePackageClasses = CustomEvent.class)用来告诉bus自己实现的事件在哪个包下，也可直接写在启动类上。

## 4. 项目A的事件发布代码 ##

#### 4.1 事件发布service ####

public interface BusEventService { 
        Boolean publishRouteEvent(String route);
    }

#### 4.2 事件发布serviceImpl ####

import lombok.extern.log4j.Log4j2;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.cloud.bus.BusProperties;
    import org.springframework.context.ApplicationEventPublisher;
    import org.springframework.stereotype.Service;
    
    /** * @author mazhen * @className BusEventServiceImpl * @Description TODO * @date 2020/11/20 16:32 */
    @Log4j2
    @Service
    public class BusEventServiceImpl implements BusEventService { 
    
        @Autowired
        private ApplicationEventPublisher applicationEventPublisher;
    
        @Autowired
        private BusProperties busProperties;
    
        @Override
        public Boolean publishRouteEvent(String route) { 
            RouteEvent routeEvent = new RouteEvent(this,busProperties.getId(),"GATEWAY-PORTAL",route);
            try { 
                log.info("Bus start publishing event");
                applicationEventPublisher.publishEvent(routeEvent);
                log.info("Bus publish event success");
                return true;
            } catch (Exception e) { 
                log.error("Bus publish event fail:",e);
            }
            return false;
        }
     }

#### 4.3 事件发布controller ####

import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.web.bind.annotation.GetMapping;
    import org.springframework.web.bind.annotation.RequestMapping;
    import org.springframework.web.bind.annotation.RequestParam;
    import org.springframework.web.bind.annotation.RestController;
    
    /** * @author mazhen * @className GatewayRoutesApiController * @Description * @date 2020/11/20 19:42 */
    @Slf4j
    @RestController
    @RequestMapping("/bus/publish")
    public class GatewayRoutesApiController { 
        @Autowired
        private BusEventService busEventService;
    
        @GetMapping("/test")
        public Boolean testGateway(@RequestParam("routeNo") String routeNo){ 
            return busEventService.publishRouteEvent(routeNo);
        }
    }

## 5. 项目B的事件消费代码 ##

项目B的事件消费代码即事件监听器：

import lombok.extern.log4j.Log4j2;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.context.ApplicationListener;
    import org.springframework.stereotype.Component;
    
    /** * @author mazhen * @className BusRoutesRefreshEventListener * @Description TODO * @date 2020/11/20 17:00 */
    @Log4j2
    @Component
    public class BusRoutesRefreshEventListener implements ApplicationListener<RouteEvent> { 
    
        @Autowired
        private RefreshRoutesEventProcessor processor;
    
        @Override
        public void onApplicationEvent(RouteEvent routeEvent) { 
            String message = routeEvent.getMessage();
            log.info("接受数据,message:{}",message);
            //具体的业务逻辑
        }
    }

## 6. 测试 ##

#### 6.1 调用项目A的发布事件controller ####

执行http://localhost:8080/bus/publish/test?routeNo=this is a test data  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L21hbWVuZzE5ODg_size_16_color_FFFFFF_t_70_pic_center]  
看日志，项目A发布事件成功。

#### 6.2 项目B的消费事件 ####

![在这里插入图片描述][20201124112703283.png_pic_center]  
看日志，项目B消费事件成功。

参考：[https://blog.csdn.net/weixin\_38312502/article/details/105975490][https_blog.csdn.net_weixin_38312502_article_details_105975490]

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L21hbWVuZzE5ODg_size_16_color_FFFFFF_t_70_pic_center]: /images/20221120/e1b4073c24854708b9f7d67a21570ec6.png
[20201124112703283.png_pic_center]: /images/20221120/6dd54a418d0e42f995d4657667acf705.png
[https_blog.csdn.net_weixin_38312502_article_details_105975490]: https://blog.csdn.net/weixin_38312502/article/details/105975490