spring云总线事件(remoteapplicationevent)未发布到kafka

uinbv5nw  于 2021-06-04  发布在  Kafka
关注(0)|答案(1)|浏览(678)

我正在使用SpringCloudBus将事件发布到kafka,以便另一个示例可以侦听相同的事件。事件正在触发,但未发布到Kafka。我用的是Spring Cloud巴士和Spring Cloud流。
版本:spring boot:2.0,spring cloud bus:2.0.0,spring cloud stream:2.0.1
应用程序.yml:

server:
  port: 7711
spring:
  application:
    index: ${random.uuid}
  cloud:
    bus:
      enabled: true
    stream:
      kafka:
        binder:
          brokers: localhost:9092
      bindings:
        input:
          destination: EMPLOYEE-TOPIC-DEMO-R1-P1
          group: ali

pom.xml文件

<dependency>
    <groupId>org.springframework.cloud</groupId>
    <artifactId>spring-cloud-stream</artifactId>
</dependency>
<dependency>
    <groupId>org.springframework.cloud</groupId>
    <artifactId>spring-cloud-stream-binder-kafka</artifactId>
</dependency>

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

发布事件:

@Autowired
private ApplicationContext context;

@StreamListener(ConsumerStream.INPUT)
public void messageConsumer(@Payload String jsonValue, @Headers MessageHeaders header) {

    try {
        log.info("Enter in Consumer->messageConsumer()");
        final String myUniqueId = context.getId();
        context.publishEvent(new MessagingEventBus(this,myUniqueId,header));
    } catch (Exception e) {
        log.error("Exception caught while processing the request :", e);
    }
}

事件类别:

@Slf4j
public class MessagingEventBus extends RemoteApplicationEvent {

    private MessageHeaders header;

    // Must supply a default constructor and getters/setters for deserialization
    public MessagingEventBus() {
    }

    public MessagingEventBus(Object source, String originService, MessageHeaders header) {
        // source is the object that is publishing the event
        // originService is the unique context ID of the publisher
        super(source, originService);
        this.header = header;
    }

}

事件侦听器:

@Component
@Slf4j
public class MessagingEventBusListener implements ApplicationListener<MessagingEventBus> {

    @Override
    public void onApplicationEvent(MessagingEventBus messagingEventBus) {
       log.info("Messaging Event Bus Listener called");
    }
}
frebpwbc

frebpwbc1#

这是总线发送事件的过程。
new remoteapplicationevent()//创建事件
applicationcontext#publishevent//发布本地事件
总线自动配置#acceptlocal//总线接受2步发送的本地事件
servicematcher#isfromself//总线将判断事件是自发送的?如果是自身,则发送到外部通道(5步)
cloudbusoutboundchannel#发送
问题是4步判断失败,您可以使用busproperties#getid作为事件源服务,这是有效的

@Autowired
BusProperties busProperties;

public void fire(){
    new RemoteApplicationEvent(this, busProperties.getId().......
}

相关问题