# spring-cloud-stream-eventbus_1

**Repository Path**: wangzongbo/spring-cloud-stream-eventbus_1

## Basic Information

- **Project Name**: spring-cloud-stream-eventbus_1
- **Description**: No description available
- **Primary Language**: Unknown
- **License**: Apache-2.0
- **Default Branch**: main
- **Homepage**: None
- **GVP Project**: No

## Statistics

- **Stars**: 0
- **Forks**: 3
- **Created**: 2023-01-30
- **Last Updated**: 2023-01-30

## Categories & Tags

**Categories**: Uncategorized

**Tags**: None

## README

## 目的

SpringCloudStream 使用函数编程后配置太繁琐了, 简直难受. 本项目旨在简化其配置. 使用几个注解就能定义个事件生产和消费.

## 事件总线的要求

- 事件创建者 ok
- 事件总线 ok
- 事件存储（Event Store）no
- 订阅者 ok
- 事件观察者（Event Watcher）no

## 支持

- RabbitMQ
- Apache Kafka
- Apache RocketMQ

## example

1. 定义事件

```
//指定eventKey: 在rabbitmq中为routingKey, 在kafka中为topic, 在rocketmq中为topic
//指定binder 指定那个binder
@Event(eventKey = "device.create.event",binder = "local_rabbit")
public class DeviceCreateEvent extends AbstractEvent {

    private String device;

    private long ts;

    private String createBy;
}
```

2. 定义事件消费

```
//group 在rabbitmq中为queue, 在kafka中为group, 在rocketmq中为queue(消费组概念)
@Component("c1")
@EventSubscribe(group = "q1", consumer = 11, binder = "local_rabbit")
public class ConsumerBean implements Consumer<DeviceCreateEvent> {

    @Override
    public void accept(DeviceCreateEvent deviceCreateEvent) {

        System.out.println("ConsumerBean xxxxx:"+deviceCreateEvent);
    }
}
```

3. 发送事件
    1. 使用stream模型生产消息
       ```
       //binder 代表发送到那个binder 
       @Bean
       @EventPublish(binder = "local_rabbit")
       public Supplier<DeviceCreateEvent> deviceCreateEventSource() {
           return () -> {
               DeviceCreateEvent deviceCreateEvent = new DeviceCreateEvent();
               deviceCreateEvent.setDevice(UUID.randomUUID().toString());
               deviceCreateEvent.setCreateBy(String.valueOf(new Random().nextInt(100)));
               deviceCreateEvent.setOnline(DeviceOnlineStatus.ONLINE);
               deviceCreateEvent.setTs(System.currentTimeMillis());
               return deviceCreateEvent;
           };
       }       
       ```
    2. 使用api发布事件
      ```
        @Autowired
        public EventPublisher eventPublisher;
    
        @Override
        public void run(String... args) {
            DeviceCreateEvent deviceCreateEvent=new DeviceCreateEvent();
            deviceCreateEvent.setDevice("C20110");
            deviceCreateEvent.setTs(System.currentTimeMillis());
            deviceCreateEvent.setCreateBy("666666666666666666666");
            eventPublisher.sendEvent(deviceCreateEvent);
        }
      ```
4. 配置eventbus

```
// 扫描event, consumer,producer的package
@EnableEventBus(basePackages = {
        "cn.nvr.eventbus.example.consumer",
        "cn.nvr.eventbus.example.event",
        "cn.nvr.eventbus.example.producer"})
@SpringBootApplication
public class EventbusApp {
```

## 总结

可以省略springCloudStream的大部分配置. 只对SpringCloudStream做的增强. 目前测试了rabbitmq,kafka,rocketmq