Spring整合mqtt原理分析

代码示例

配置消息处理器

import java.util.UUID;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.integration.annotation.IntegrationComponentScan;
import org.springframework.integration.annotation.ServiceActivator;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.config.EnableIntegration;
import org.springframework.integration.core.MessageProducer;
import org.springframework.integration.mqtt.core.DefaultMqttPahoClientFactory;
import org.springframework.integration.mqtt.core.MqttPahoClientFactory;
import org.springframework.integration.mqtt.inbound.MqttPahoMessageDrivenChannelAdapter;
import org.springframework.integration.mqtt.outbound.MqttPahoMessageHandler;
import org.springframework.integration.mqtt.support.DefaultPahoMessageConverter;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageHandler;
 
@Configuration
@IntegrationComponentScan
@EnableIntegration
public class MqttClientConfig {
 
  /**
   * 连接工厂.
   */
  @Bean
  public MqttPahoClientFactory mqttClientFactory() {
    DefaultMqttPahoClientFactory factory = new DefaultMqttPahoClientFactory();
    MqttConnectOptions mqttConnectOptions = new MqttConnectOptions();
    mqttConnectOptions.setUserName("test1");
    mqttConnectOptions.setPassword("123456".toCharArray());
    mqttConnectOptions.setServerURIs(new String[]{"tcp://xxx:1883"});
    mqttConnectOptions.setKeepAliveInterval(2);
    mqttConnectOptions.setAutomaticReconnect(true);
    factory.setConnectionOptions(mqttConnectOptions);
    return factory;
  }
 
 
  private String createClientId() {
    return UUID.randomUUID().toString();
  }
 
  /**
   * 配置client,发布.
   */
  @Bean
  @ServiceActivator(inputChannel = "mqttOutboundChannel")
  public MessageHandler mqttOutbound() {
    MqttPahoMessageHandler messageHandler = new MqttPahoMessageHandler(
        createClientId(), mqttClientFactory());
    messageHandler.setAsync(true);
    messageHandler.setDefaultQos(2);
    messageHandler.setDefaultRetained(false); //不保留消息
    return messageHandler;
  }
 
  @Bean
  public MessageChannel mqttOutboundChannel() {
    return new DirectChannel();
  }
 
  //接收通道
  @Bean
  public MessageChannel mqttInputChannel() {
    return new DirectChannel();
  }
 
  /**
   * 配置client,监听的topic.
   */
  @Bean
  public MessageProducer inbound() {
    String[] topics = {"first_topic"};
    MqttPahoMessageDrivenChannelAdapter adapter =
        new MqttPahoMessageDrivenChannelAdapter(createClientId(),
            mqttClientFactory(), topics);
    adapter.setCompletionTimeout(3_000);
    adapter.setConverter(new DefaultPahoMessageConverter());
    adapter.setQos(2);
    adapter.setOutputChannel(mqttInputChannel());
    return adapter;
  }
 
  /**
   * 消息处理器
   */
  @Bean
  @ServiceActivator(inputChannel = "mqttInputChannel")
  public MessageHandler handler() {
    return (message -> {
      String topic = message.getHeaders().get("mqtt_receivedTopic").toString();
      String payload = message.getPayload().toString();
      System.out.println("消息主题：" + topic);
      System.out.println("消息内容：" + payload);
    });
  }
}

配置消息通道MessageChannel和消息处理器，消息通道有两个，一个用来发送消息，一个用来接收消息。

配置消息网关

import org.springframework.integration.annotation.MessagingGateway;
import org.springframework.integration.mqtt.support.MqttHeaders;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.stereotype.Component;
 
@Component
@MessagingGateway(defaultRequestChannel = "mqttOutboundChannel")
public interface MqttGateway {
 
  void sendToMqtt(String data, @Header(MqttHeaders.TOPIC) String topic);
}

加上@Component注解是为了IDE不报错，不加也不影响功能。

客户端使用

import org.springframework.context.annotation.AnnotationConfigApplicationContext;
 
public class TestSpringMqtt {
 
  public static void main(String[] args) {
    // 创建容器上下文
    AnnotationConfigApplicationContext context =
        new AnnotationConfigApplicationContext(MqttClientConfig.class);
    MqttGateway mqttGateway = (MqttGateway) context.getBean("mqttGateway");
    // 获取消息网关Bean对象并发送消息
    mqttGateway.sendToMqtt("hello", "first_topic");
  }
}

原理分析

@IntegrationComponentScan注解

先来分析@IntegrationComponentScan注解，它会导入IntegrationComponentScanRegistrar配置类。

public class IntegrationComponentScanRegistrar implements ImportBeanDefinitionRegistrar,
		ResourceLoaderAware, EnvironmentAware {
 
	private final Map componentRegistrars =
			new HashMap();
 
        // 过滤类型为@MessagingGateway注解，会交给MessagingGatewayRegistrar来处理
	public IntegrationComponentScanRegistrar() {
		this.componentRegistrars.put(new AnnotationTypeFilter(MessagingGateway.class, true),
				new MessagingGatewayRegistrar());
	}
	
	@Override
  public void registerBeanDefinitions(AnnotationMetadata importingClassMetadata, BeanDefinitionRegistry registry) {
  		Map componentScan =
  				importingClassMetadata.getAnnotationAttributes(IntegrationComponentScan.class.getName());
  
  		Collection basePackages = getBasePackages(importingClassMetadata, registry);
  
  		if (basePackages.isEmpty()) {
  			basePackages = Collections.singleton(ClassUtils.getPackageName(importingClassMetadata.getClassName()));
  		}
  
  		ClassPathScanningCandidateComponentProvider scanner =
  				new ClassPathScanningCandidateComponentProvider(false, this.environment) {
  
  					@Override
  					protected boolean isCandidateComponent(AnnotatedBeanDefinition beanDefinition) {
  						return beanDefinition.getMetadata().isIndependent()
  								&& !beanDefinition.getMetadata().isAnnotation();
  					}
  
  				};
  
  		filter(registry, componentScan, scanner); // NOSONAR - never null
  
  		scanner.setResourceLoader(this.resourceLoader);
  
                // 开始扫描
  		for (String basePackage : basePackages) {
  			Set candidateComponents = scanner.findCandidateComponents(basePackage);
  			for (BeanDefinition candidateComponent : candidateComponents) {
  				if (candidateComponent instanceof AnnotatedBeanDefinition) {
  					for (ImportBeanDefinitionRegistrar registrar : this.componentRegistrars.values()) {
                                                // 交给具体的注册器来处理，这里就是MessagingGatewayRegistrar
  						registrar.registerBeanDefinitions(((AnnotatedBeanDefinition) candidateComponent).getMetadata(),
  								registry);
  					}
  				}
  			}
  		}
  	}
	
}

此注解作用类似于@ComponentScan注解，会扫描包含@MessagingGateway注解的Class，并注册到容器中。继续分析MessagingGatewayRegistrar

public class MessagingGatewayRegistrar implements ImportBeanDefinitionRegistrar {
 
	@Override
	public void registerBeanDefinitions(AnnotationMetadata importingClassMetadata, BeanDefinitionRegistry registry) {
		if (importingClassMetadata != null && importingClassMetadata.isAnnotated(MessagingGateway.class.getName())) {
			Map annotationAttributes =
					importingClassMetadata.getAnnotationAttributes(MessagingGateway.class.getName());
			BeanDefinitionReaderUtils.registerBeanDefinition(this.parse(annotationAttributes), registry);
		}
	}
 
	public BeanDefinitionHolder parse(Map gatewayAttributes) {
                // 创建具体的BeanDefinition
		String defaultPayloadExpression = (String) gatewayAttributes.get("defaultPayloadExpression");
                // 核心，接口的具体实现类为GatewayProxyFactoryBean
		BeanDefinitionBuilder gatewayProxyBuilder =
				BeanDefinitionBuilder.genericBeanDefinition(GatewayProxyFactoryBean.class);
		AbstractBeanDefinition beanDefinition = gatewayProxyBuilder.getBeanDefinition();
		beanDefinition.addMetadataAttribute(new BeanMetadataAttribute(IntegrationConfigUtils.FACTORY_BEAN_OBJECT_TYPE,
				serviceInterface));
 
		return new BeanDefinitionHolder(beanDefinition, id);
	}
 
}
 

使用GatewayProxyFactoryBean类来作为接口的具体实现。

public class GatewayProxyFactoryBean extends AbstractEndpoint
		implements TrackableComponent, FactoryBean