Spring Cloud Bus 消息总线 kafka 集成配置
集成阿里云kafka作为消息总线,参照https://github.com/AliwareMQ/aliware-kafka-demos/tree/master/kafka-spring-stream-demo进行配置配置topicspring.cloud.bus.destination=alikafka-xxxxxxxx禁止自动创建topicspring.cloud.stream.kafka.bi
集成阿里云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作为消息总线的实现就跑通了。
更多推荐
所有评论(0)