前言: 本文作者张天,节选自笔者与其合著的《Spring Cloud微服务架构进阶》,即将在八月出版问世。本文将其中Spring Cloud Stream应用与自定义Rocketmq Binder的内容抽取出来,主要介绍实现Spring Cloud Stream 的RocketMQ绑定器。
在上一篇中,介绍了Spring Cloud Stream基本的概念及其编程模型。除此之外,Spring Cloud Stream提供了Binder接口来用于和外部消息队列进行绑定。本文将讲述Binder SPI的基本概念,主要组件和实现细节。 Binder SPI通过一系列的接口,工具类和检测机制提供了与外部消息队列绑定的绑定器机制。SPI的关键点是Binder接口,这个接口负责提供和外部消息队列进行绑定的具体实现。
1public interface Binder<T, C extends ConsumerProperties, P extends ProducerProperties> {
2 Binding<T> bindConsumer(String name, String group, T inboundBindTarget, C consumerProperties);
3 Binding<T> bindProducer(String name, T outboundBindTarget, P producerProperties);
4}
一个典型的自定义Binder组件实现应该包括以下几点:
1kafka:\
2org.springframework.cloud.stream.binder.kafka.config.KafkaBinderConfiguration
Spring Cloud Stream基于Binder SPI的实现来进行channel和消息队列的绑定任务。不同类型的消息队列中间件实现了不同的绑定器Binder。比如说:Spring-Cloud-Stream-Binder-Kafka是针对Kafka的Binder实现,而Spring-Cloud-Stream-Binder-Rabbit则是针对RabbitMQ的Binder实现。
Spring Cloud Stream依赖于Spring Boot的自动配置机制来配置Binder。如果一个Binder实现在项目的classpath中被发现,Spring Cloud Stream将会自动使用它。比如说,一个Spring Cloud Stream项目需要绑定RabbitMQ中间件的Binder,在pom文件中加入下面的依赖来轻松实现。
1<dependency>
2 <groupId>org.springframework.cloud</groupId>
3 <artifactId>spring-cloud-stream-binder-rabbit</artifactId>
4</dependency>
Spring Cloud Stream为接入不同的消息队列提供了一整套的自定义机制,通过为每个消息队里开发一个Binder来接入该消息队列。目前官方认定的Binder为rabbitmq binder和kafka binder。但是开发人员可以基于Stream Binder的机制来制定自己的Binder。下面我们就构建一个简单的RocketMQ的Binder。
需要在resources/META-INF/spring.binders文件中配置有关RocketMQ的Configuration类,该配置类会使用@Import来导入为RocketMQ制定的RocketMessageChannelBinderConfiguration
。
1rocket:\
2org.springframework.cloud.stream.binder.rocket.config.RocketServiceAutoConfiguration
RocketMessageChannelBinderConfiguration
将会提供两个极其重要的bean实例,分别为RocketMessageChannelBinder
和RocketExchangeQueueProvisioner
。RocketMessageChannelBinder
主要是用于channel和消息队列的绑定,而RocketExchangeQueueProvisioner
则封装了RocketMQ的相关API,可以用于创建消息队列的基础组件,比如说队列,交换器等。
1@Configuration
2public class RocketMessageChannelBinderConfiguration {
3 @Autowired
4 private ConnectionFactory rocketConnectionFactory;
5 @Autowired
6 private RocketProperties rocketProperties;
7 @Bean
8 RocketMessageChannelBinder rocketMessageChannelBinder() throws Exception {
9 RocketMessageChannelBinder binder = new RocketMessageChannelBinder(this.rocketConnectionFactory,
10 this.rocketProperties, provisioningProvider());
11 return binder;
12 }
13 @Bean
14 RocketExchangeQueueProvisioner provisioningProvider() {
15 return new RocketExchangeQueueProvisioner(this.rocketConnectionFactory);
16 }
17}
RocketMessageChannelBinder
继承了抽象类AbstractMessageChannelBinder
,并实现了#producerMessageHandler和#createConsumerEndpoint函数。
MessageHandler有向消息队列发送消息的能力,#createProducerMessageHandler函数就是为了创建MessageHandler对象,来将输出型Channel的消息发送到消息队列上。
1protected MessageHandler createProducerMessageHandler(
2 ProducerDestination destination,
3 ExtendedProducerProperties<RocketProducerProperties> producerProperties,
4 MessageChannel errorChannel)
5 throws Exception {
6 final AmqpOutboundEndpoint endpoint = new AmqpOutboundEndpoint(
7 buildRocketTemplate(producerProperties.getExtension(), errorChannel != null));
8 return endpoint;
9}
MessageProducer能够从消息队列接收消息,并将该消息发送输入型Channel。
1@Override
2protected MessageProducer createConsumerEndpoint(ConsumerDestination consumerDestination, String group,
3 ExtendedConsumerProperties<RocketConsumerProperties> properties) throws Exception {
4 SimpleRocketMessageListenerContainer listenerContainer = new SimpleRocketMessageListenerContainer();
5 RocketInboundChannelAdapter rocketInboundChannelAdapter = new RocketInboundChannelAdapter(listenerContainer);
6 return rocketInboundChannelAdapter;
7}
类似于RabbitMQ的Binder,需要实现下面一系列的类来实现从RocketMQ到对应MessageChannel的消息传递。
InnerConsumer实现的MessageListenerConcurrently接口是RocketMQ中用于并发接受异步消息的接口,该接口可以接收到RocketMQ发送过来的异步消息。而InnerConsumer在接受到消息之后,会将消息封装成RocketDelivery加入到阻塞队列中。
RocketBlockingQueueConsumer有一个阻塞队列来存储RocketMQ传递给RocketBlockingQueueConsumer.InnerConsumer的消息,而nextMessage函数可以从阻塞队列中拉取一个消息并返回。
SimpleRocketMessageListenerContainer.AsyncMessageProcessingConsumer是实现了Runnable接口,在run()接口中会无限循环地调用SimpleRocketMessageListenerContainer本身的receiveAndExecute。
1@Override
2public void run() {
3 if (!isActive()) {
4 return;
5 }
6 try {
7 //只要consumer的状态正常,就会一直循环
8 while (isActive(this.consumer) || this.consumer.hasDelivery() || !this.consumer.cancelled()) {
9 try {
10 boolean receivedOk = receiveAndExecute(this.consumer);
11 }
12 catch (ListenerExecutionFailedException ex) {
13 if (ex.getCause() instanceof NoSuchMethodException) {
14 throw new FatalListenerExecutionException("Invalid listener", ex);
15 }
16 }
17 catch (AmqpRejectAndDontRequeueException rejectEx) {
18 } catch (Throwable e) {
19 }
20 }
21 } catch (Exception e) {
22 }
23 finally {
24 if (getTransactionManager() != null) {
25 ConsumerChannelRegistry.unRegisterConsumerChannel();
26 }
27 }
28 this.start.countDown();
29 if (!isActive(this.consumer) || aborted) {
30 this.consumer.stop();
31 }
32 else {
33 restart(this.consumer);
34 }
35}
函数#receiveAndExecute最终的作用就是调用RocketBlockingQueueConsumer的nextMessage,然后再将消息调用messageListener.onMessage函数将消息传递出去。
SimpleRocketMessageListenerContainer的doStart函数会初始化RocketBlockingQueueConsumer并且启动SimpleRocketMessageListenerContainer的AsyncMessageProcessingConsumer会无限循环地从RocketBlockingQueueConsumer中获取RocketMQ传递过来的消息。
1private void doStart() {
2 synchronized (this.lifecycleMonitor) {
3 this.active = true;
4 this.running = true;
5 this.lifecycleMonitor.notifyAll();
6 }
7 synchronized (this.consumersMonitor) {
8 if (this.consumers != null) {
9 throw new IllegalStateException("A stopped container should not have consumers");
10 }
11 //初始化Consumer
12 int newConsumers = initializeConsumers();
13 if (this.consumers == null) {
14 return;
15 }
16 if (newConsumers <= 0) {
17 return;
18 }
19 Set<SimpleRocketMessageListenerContainer.AsyncMessageProcessingConsumer> processors =
20 new HashSet<>();
21 //对于每个RocketBlockingQueueConsumer启动一个
22 //AsyncMessageProcessingConsumer来执行任务
23 for (RocketBlockingQueueConsumer consumer : this.consumers) {
24 SimpleRocketMessageListenerContainer.AsyncMessageProcessingConsumer
25 processor = new SimpleRocketMessageListenerContainer.AsyncMessageProcessingConsumer(consumer);
26 processors.add(processor);
27 getTaskExecutor().execute(processor);
28 }
29 }
30}
RocketInboundChannelAdapter实现了MessageProducer接口。它主要将SimpleRocketMessageListenerContainer传递过来的消息经过MessageTemplate传递给MessageChannel。
接下来则是RocketInboundChannelAdapter.Listener的实现,它就是RocketBlockingQueueConsumer.nextMessage函数中的messageListener。
1public class Listener implements ChannelAwareMessageListener, RetryListener {
2 public void onMessage(Message message, Channel channel) throws Exception {
3 try {
4 this.createAndSend(message, channel);
5 } catch (RuntimeException var7) {
6 if (RocketInboundChannelAdapter.this.getErrorChannel() == null) {
7 throw var7;
8 }
9 RocketInboundChannelAdapter.this.getMessagingTemplate().send(RocketInboundChannelAdapter.this.getErrorChannel(), RocketInboundChannelAdapter.this.buildErrorMessage((org.springframework.messaging.Message)null, new ListenerExecutionFailedException("Message conversion failed", var7, message)));
10 }
11 }
12 private void createAndSend(Message message, Channel channel) {
13 org.springframework.messaging.Message<Object> messagingMessage = this.createMessage(message, channel);
14 RocketInboundChannelAdapter.this.sendMessage(messagingMessage);
15 }
16 private org.springframework.messaging.Message<Object> createMessage(Message message, Channel channel) {
17 Object payload = RocketInboundChannelAdapter.this.messageConverter.fromMessage(message);
18 org.springframework.messaging.Message<Object> messagingMessage = RocketInboundChannelAdapter.this.getMessageBuilderFactory().withPayload(payload).build();
19 return messagingMessage;
20 }
21}
RocketProvisioningProvider实现了ProvisioningProvider接口,它有两个函数:provisionProducerDestination和provisionConsumerDestination,分别用于创建ProducerDestination和ConsumerDestination。RocketProvisioningProvider的实现类似于RabbitProvisioningProvider。只不过在声明队列,交换器和绑定时使用了RocketAdmin所实现的RocketMQ的相关API。
本文概要介绍了Spring Cloud Stream的Rocketmq绑定器的实现,限于篇幅不展开具体的代码讲解。读者感兴趣,可以关注GitHub上的代码。根据Spring Cloud Stream抽象的接口,我们可以自由地实现各种消息队列的绑定器。
项目GitHub地址:https://github.com/ztelur/spring-cloud-stream-binder-rocket