集成阿里云kafka作为消息总线,参照https://github.com/AliwareMQ/aliware-kafka-demos/tree/master/kafka-spring-stream-demo进行配置

配置topic

spring.cloud.bus.destination=alikafka-xxxxxxxx

禁止自动创建topic

spring.cloud.stream.kafka.binder.autoCreateTopics=false

配置接入点

spring.cloud.stream.kafka.binder.brokers=SASL_SSL://kafka-cn-internet.aliyun.com:8080

根据源代码,可以看到spring clound bus的输入输出通道如下:

/**
 * @author Dave Syer
 *
 */
public interface SpringCloudBusClient {

	String INPUT = "springCloudBusInput";

	String OUTPUT = "springCloudBusOutput";

	@Output(SpringCloudBusClient.OUTPUT)
	MessageChannel springCloudBusOutput();

	@Input(SpringCloudBusClient.INPUT)
	SubscribableChannel springCloudBusInput();
}

输出通道配置

spring.cloud.stream.bindings.springCloudBusOutput.

输入通道配置

spring.cloud.stream.bindings.springCloudBusInput.

配置消费组

spring.cloud.stream.bindings.springCloudBusInput.group=CID_alikafka-xxxxxxxxx

又踩了些坑,因为我已经用到2.0的spring cloud了,但例子里是1.5.x。

2.0的spring cloud bus 的kafka client升级到了1.0.1,而阿里云上的kafka版本还是0.10.0

按上述配置万之后,启动正常,但是发送消息时,会报出一个“Magic v1 does not support record headers”的异常。

这是由于client和broker版本不一致的关系。

参考 https://github.com/spring-cloud/spring-cloud-stream-binder-kafka/issues/290 ,可以通过headerMode配置来解决

在客户端和服务端都增加如下配置:

spring.cloud.stream.default.producer.headerMode=embeddedHeaders
spring.cloud.stream.default.consumer.headerMode=embeddedHeaders

配置中心使用kafka作为消息总线的实现就跑通了。

Logo

Kafka开源项目指南提供详尽教程,助开发者掌握其架构、配置和使用,实现高效数据流管理和实时处理。它高性能、可扩展,适合日志收集和实时数据处理,通过持久化保障数据安全,是企业大数据生态系统的核心。

更多推荐