我正在使用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");
}
}
1条答案
按热度按时间frebpwbc1#
这是总线发送事件的过程。
new remoteapplicationevent()//创建事件
applicationcontext#publishevent//发布本地事件
总线自动配置#acceptlocal//总线接受2步发送的本地事件
servicematcher#isfromself//总线将判断事件是自发送的?如果是自身,则发送到外部通道(5步)
cloudbusoutboundchannel#发送
问题是4步判断失败,您可以使用busproperties#getid作为事件源服务,这是有效的