spring集成mqtt客户端相关逻辑分析
Posted oldwangneverdie
tags:
篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了spring集成mqtt客户端相关逻辑分析相关的知识,希望对你有一定的参考价值。
概述
mqtt客户端有很多java实现,官网上列出了下面这些:
-
Xenqtt Includes a client library, mock broker for unit/integration testing, and applications to support enterprise needs like using a cluster of servers as a single client, an HTTP gateway, etc.
-
IA92 - deprecated IBM IA92 support pack, use Eclipse Paho GUI client instead. A useful MQTT Java swing GUI for publishing & subscribing. The Eclipse Paho GUI is identical but uses newer client code
-
vertx-mqtt-client is an open-source, high performance, non-blocking MQTT client built as a part of vert.x‘s jvm toolkit.
-
Qatja is a Java client library for MQTT 3.1.1 with specific implementation for android and Processing
网上查了下, spring集成mqtt使用了第一个包。
下面先分析下第一个包的 原生用法,然后分析下spring 对其的封装逻辑
Eclipse Paho Java原生使用方法
以官方的例子为例:
/******************************************************************************* * Copyright (c) 2009, 2014 IBM Corp. * * All rights reserved. This program and the accompanying materials * are made available under the terms of the Eclipse Public License v1.0 * and Eclipse Distribution License v1.0 which accompany this distribution. * * The Eclipse Public License is available at * http://www.eclipse.org/legal/epl-v10.html * and the Eclipse Distribution License is available at * http://www.eclipse.org/org/documents/edl-v10.php. * * Contributors: * Dave Locke - initial API and implementation and/or initial documentation */ package org.eclipse.paho.sample.mqttv3app; import java.io.IOException; import java.sql.Timestamp; import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken; import org.eclipse.paho.client.mqttv3.MqttCallback; import org.eclipse.paho.client.mqttv3.MqttClient; import org.eclipse.paho.client.mqttv3.MqttConnectOptions; import org.eclipse.paho.client.mqttv3.MqttException; import org.eclipse.paho.client.mqttv3.MqttMessage; import org.eclipse.paho.client.mqttv3.persist.MqttDefaultFilePersistence; /** * A sample application that demonstrates how to use the Paho MQTT v3.1 Client blocking API. * * It can be run from the command line in one of two modes: * - as a publisher, sending a single message to a topic on the server * - as a subscriber, listening for messages from the server * * There are three versions of the sample that implement the same features * but do so using using different programming styles: * <ol> * <li>Sample (this one) which uses the API which blocks until the operation completes</li> * <li>SampleAsyncWait shows how to use the asynchronous API with waiters that block until * an action completes</li> * <li>SampleAsyncCallBack shows how to use the asynchronous API where events are * used to notify the application when an action completes<li> * </ol> * * If the application is run with the -h parameter then info is displayed that * describes all of the options / parameters. */ public class Sample implements MqttCallback { /** * The main entry point of the sample. * * This method handles parsing of the arguments specified on the * command-line before performing the specified action. */ public static void main(String[] args) { // Default settings: boolean quietMode = false; String action = "publish"; String topic = ""; String message = "Message from blocking Paho MQTTv3 Java client sample"; int qos = 2; String broker = "m2m.eclipse.org"; int port = 1883; String clientId = null; String subTopic = "Sample/#"; String pubTopic = "Sample/Java/v3"; boolean cleanSession = true; // Non durable subscriptions boolean ssl = false; String password = null; String userName = null; // Parse the arguments - for (int i=0; i<args.length; i++) { // Check this is a valid argument if (args[i].length() == 2 && args[i].startsWith("-")) { char arg = args[i].charAt(1); // Handle arguments that take no-value switch(arg) { case ‘h‘: case ‘?‘: printHelp(); return; case ‘q‘: quietMode = true; continue; } // Now handle the arguments that take a value and // ensure one is specified if (i == args.length -1 || args[i+1].charAt(0) == ‘-‘) { System.out.println("Missing value for argument: "+args[i]); printHelp(); return; } switch(arg) { case ‘a‘: action = args[++i]; break; case ‘t‘: topic = args[++i]; break; case ‘m‘: message = args[++i]; break; case ‘s‘: qos = Integer.parseInt(args[++i]); break; case ‘b‘: broker = args[++i]; break; case ‘p‘: port = Integer.parseInt(args[++i]); break; case ‘i‘: clientId = args[++i]; break; case ‘c‘: cleanSession = Boolean.valueOf(args[++i]).booleanValue(); break; case ‘k‘: System.getProperties().put("javax.net.ssl.keyStore", args[++i]); break; case ‘w‘: System.getProperties().put("javax.net.ssl.keyStorePassword", args[++i]); break; case ‘r‘: System.getProperties().put("javax.net.ssl.trustStore", args[++i]); break; case ‘v‘: ssl = Boolean.valueOf(args[++i]).booleanValue(); break; case ‘u‘: userName = args[++i]; break; case ‘z‘: password = args[++i]; break; default: System.out.println("Unrecognised argument: "+args[i]); printHelp(); return; } } else { System.out.println("Unrecognised argument: "+args[i]); printHelp(); return; } } // Validate the provided arguments if (!action.equals("publish") && !action.equals("subscribe")) { System.out.println("Invalid action: "+action); printHelp(); return; } if (qos < 0 || qos > 2) { System.out.println("Invalid QoS: "+qos); printHelp(); return; } if (topic.equals("")) { // Set the default topic according to the specified action if (action.equals("publish")) { topic = pubTopic; } else { topic = subTopic; } } String protocol = "tcp://"; if (ssl) { protocol = "ssl://"; } String url = protocol + broker + ":" + port; if (clientId == null || clientId.equals("")) { clientId = "SampleJavaV3_"+action; } // With a valid set of arguments, the real work of // driving the client API can begin try { // Create an instance of this class Sample sampleClient = new Sample(url, clientId, cleanSession, quietMode,userName,password); // Perform the requested action if (action.equals("publish")) { sampleClient.publish(topic,qos,message.getBytes()); } else if (action.equals("subscribe")) { sampleClient.subscribe(topic,qos); } } catch(MqttException me) { // Display full details of any exception that occurs System.out.println("reason "+me.getReasonCode()); System.out.println("msg "+me.getMessage()); System.out.println("loc "+me.getLocalizedMessage()); System.out.println("cause "+me.getCause()); System.out.println("excep "+me); me.printStackTrace(); } } // Private instance variables private MqttClient client; private String brokerUrl; private boolean quietMode; private MqttConnectOptions conOpt; private boolean clean; private String password; private String userName; /** * Constructs an instance of the sample client wrapper * @param brokerUrl the url of the server to connect to * @param clientId the client id to connect with * @param cleanSession clear state at end of connection or not (durable or non-durable subscriptions) * @param quietMode whether debug should be printed to standard out * @param userName the username to connect with * @param password the password for the user * @throws MqttException */ public Sample(String brokerUrl, String clientId, boolean cleanSession, boolean quietMode, String userName, String password) throws MqttException { this.brokerUrl = brokerUrl; this.quietMode = quietMode; this.clean = cleanSession; this.password = password; this.userName = userName; //This sample stores in a temporary directory... where messages temporarily // stored until the message has been delivered to the server. //..a real application ought to store them somewhere // where they are not likely to get deleted or tampered with String tmpDir = System.getProperty("java.io.tmpdir"); MqttDefaultFilePersistence dataStore = new MqttDefaultFilePersistence(tmpDir); try { // Construct the connection options object that contains connection parameters // such as cleanSession and LWT conOpt = new MqttConnectOptions(); conOpt.setCleanSession(clean); if(password != null ) { conOpt.setPassword(this.password.toCharArray()); } if(userName != null) { conOpt.setUserName(this.userName); } // Construct an MQTT blocking mode client client = new MqttClient(this.brokerUrl,clientId, dataStore); // Set this wrapper as the callback handler client.setCallback(this); } catch (MqttException e) { e.printStackTrace(); log("Unable to set up client: "+e.toString()); System.exit(1); } } /** * Publish / send a message to an MQTT server * @param topicName the name of the topic to publish to * @param qos the quality of service to delivery the message at (0,1,2) * @param payload the set of bytes to send to the MQTT server * @throws MqttException */ public void publish(String topicName, int qos, byte[] payload) throws MqttException { // Connect to the MQTT server log("Connecting to "+brokerUrl + " with client ID "+client.getClientId()); client.connect(conOpt); log("Connected"); String time = new Timestamp(System.currentTimeMillis()).toString(); log("Publishing at: "+time+ " to topic \""+topicName+"\" qos "+qos); // Create and configure a message MqttMessage message = new MqttMessage(payload); message.setQos(qos); // Send the message to the server, control is not returned until // it has been delivered to the server meeting the specified // quality of service. client.publish(topicName, message); // Disconnect the client client.disconnect(); log("Disconnected"); } /** * Subscribe to a topic on an MQTT server * Once subscribed this method waits for the messages to arrive from the server * that match the subscription. It continues listening for messages until the enter key is * pressed. * @param topicName to subscribe to (can be wild carded) * @param qos the maximum quality of service to receive messages at for this subscription * @throws MqttException */ public void subscribe(String topicName, int qos) throws MqttException { // Connect to the MQTT server client.connect(conOpt); log("Connected to "+brokerUrl+" with client ID "+client.getClientId()); // Subscribe to the requested topic // The QoS specified is the maximum level that messages will be sent to the client at. // For instance if QoS 1 is specified, any messages originally published at QoS 2 will // be downgraded to 1 when delivering to the client but messages published at 1 and 0 // will be received at the same level they were published at. log("Subscribing to topic \""+topicName+"\" qos "+qos); client.subscribe(topicName, qos); // Continue waiting for messages until the Enter is pressed log("Press <Enter> to exit"); try { System.in.read(); } catch (IOException e) { //If we can‘t read we‘ll just exit } // Disconnect the client from the server client.disconnect(); log("Disconnected"); } /** * Utility method to handle logging. If ‘quietMode‘ is set, this method does nothing * @param message the message to log */ private void log(String message) { if (!quietMode) { System.out.println(message); } } /****************************************************************/ /* Methods to implement the MqttCallback interface */ /****************************************************************/ /** * @see MqttCallback#connectionLost(Throwable) */ public void connectionLost(Throwable cause) { // Called when the connection to the server has been lost. // An application may choose to implement reconnection // logic at this point. This sample simply exits. log("Connection to " + brokerUrl + " lost!" + cause); System.exit(1); } /** * @see MqttCallback#deliveryComplete(IMqttDeliveryToken) */ public void deliveryComplete(IMqttDeliveryToken token) { // Called when a message has been delivered to the // server. The token passed in here is the same one // that was passed to or returned from the original call to publish. // This allows applications to perform asynchronous // delivery without blocking until delivery completes. // // This sample demonstrates asynchronous deliver and // uses the token.waitForCompletion() call in the main thread which // blocks until the delivery has completed. // Additionally the deliveryComplete method will be called if // the callback is set on the client // // If the connection to the server breaks before delivery has completed // delivery of a message will complete after the client has re-connected. // The getPendingTokens method will provide tokens for any messages // that are still to be delivered. } /** * @see MqttCallback#messageArrived(String, MqttMessage) */ public void messageArrived(String topic, MqttMessage message) throws MqttException { // Called when a message arrives from the server that matches any // subscription made by the client String time = new Timestamp(System.currentTimeMillis()).toString(); System.out.println("Time:\t" +time + " Topic:\t" + topic + " Message:\t" + new String(message.getPayload()) + " QoS:\t" + message.getQos()); } /****************************************************************/ /* End of MqttCallback methods */ /****************************************************************/ static void printHelp() { System.out.println( "Syntax:\n\n" + " Sample [-h] [-a publish|subscribe] [-t <topic>] [-m <message text>]\n" + " [-s 0|1|2] -b <hostname|IP address>] [-p <brokerport>] [-i <clientID>]\n\n" + " -h Print this help text and quit\n" + " -q Quiet mode (default is false)\n" + " -a Perform the relevant action (default is publish)\n" + " -t Publish/subscribe to <topic> instead of the default\n" + " (publish: \"Sample/Java/v3\", subscribe: \"Sample/#\")\n" + " -m Use <message text> instead of the default\n" + " (\"Message from MQTTv3 Java client\")\n" + " -s Use this QoS instead of the default (2)\n" + " -b Use this name/IP address instead of the default (m2m.eclipse.org)\n" + " -p Use this port instead of the default (1883)\n\n" + " -i Use this client ID instead of SampleJavaV3_<action>\n" + " -c Connect to the server with a clean session (default is false)\n" + " \n\n Security Options \n" + " -u Username \n" + " -z Password \n" + " \n\n SSL Options \n" + " -v SSL enabled; true - (default is false) " + " -k Use this JKS format key store to verify the client\n" + " -w Passpharse to verify certificates in the keys store\n" + " -r Use this JKS format keystore to verify the server\n" + " If javax.net.ssl properties have been set only the -v flag needs to be set\n" + "Delimit strings containing spaces with \"\"\n\n" + "Publishers transmit a single message then disconnect from the server.\n" + "Subscribers remain connected to the server and receive appropriate\n" + "messages until <enter> is pressed.\n\n" ); } }
总结起来就是:
1、new一个MqttClient
client = new MqttClient(this.brokerUrl,clientId, dataStore);
2、set一个MqttCallback的实现类作为MqttClient的回调函数用来处理订阅的消息:
client.setCallback(this); //注意这里的this实现了MqttCallback接口
3、发布消息调用MqttClient的publish方法:
client.connect(conOpt);
MqttMessage message = new MqttMessage(payload);
message.setQos(qos);
client.publish(topicName, message);
4、订阅消息调用MqttClient的subscribe方法:
client.connect(conOpt); client.subscribe(topicName, qos);
Spring对mqtt的集成
maven配置
<dependencies> <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-integration</artifactId> </dependency> <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-integration-stream</artifactId> </dependency> <dependency> <groupId>org.springframework.integration</groupId> <artifactId>spring-integration-mqtt</artifactId> </dependency> </dependencies>
消息订阅处理的 官方例子
@SpringBootApplication public class MqttJavaApplication { public static void main(String[] args) { new SpringApplicationBuilder(MqttJavaApplication.class) .web(false) .run(args); } @Bean public MessageChannel mqttInputChannel() { return new DirectChannel(); } @Bean public MessageProducer inbound() { MqttPahoMessageDrivenChannelAdapter adapter = new MqttPahoMessageDrivenChannelAdapter("tcp://localhost:1883", "testClient", "topic1", "topic2"); adapter.setCompletionTimeout(5000); adapter.setConverter(new DefaultPahoMessageConverter()); adapter.setQos(1); adapter.setOutputChannel(mqttInputChannel()); return adapter; } @Bean @ServiceActivator(inputChannel = "mqttInputChannel") public MessageHandler handler() { return new MessageHandler() { @Override public void handleMessage(Message<?> message) throws MessagingException { System.out.println(message.getPayload()); } }; } }
MqttPahoMessageDrivenChannelAdapter的负责订阅消息
DirectChannel负责分发订阅到的消息
Messagehandler负责具体处理消息
MqttPahoMessageDrivenChannelAdapter和DireactChannel的代码逻辑很容易搞明白,下面主要梳理下MessageHandler是怎么被发现,然后用来处理消息的。
MessageHandler相关逻辑
spring.factoires中EnableAutoConfiguration的值包含org.springframework.boot.autoconfigure.integration.IntegrationAutoConfiguration,
在引入了spring-boot-starter-integration之后,@ConditionalOnClass(EnableIntegration.class)的条件满足
@Configuration @ConditionalOnClass(EnableIntegration.class) @EnableConfigurationProperties(IntegrationProperties.class) @AutoConfigureAfter(JmxAutoConfiguration.class) public class IntegrationAutoConfiguration { /** * Basic Spring Integration configuration. */ @Configuration @EnableIntegration protected static class IntegrationConfiguration { }
然后
@Target(ElementType.TYPE)
@Retention(RetentionPolicy.RUNTIME)
@Documented
@Import(IntegrationRegistrar.class)
public @interface EnableIntegration {
}
然后IntegrationRegistrar注册了一堆BeanDefination
public void registerBeanDefinitions(@Nullable AnnotationMetadata importingClassMetadata,
BeanDefinitionRegistry registry) {
registerImplicitChannelCreator(registry);
registerIntegrationConfigurationBeanFactoryPostProcessor(registry);
registerIntegrationEvaluationContext(registry);
registerIntegrationProperties(registry);
registerHeaderChannelRegistry(registry);
registerGlobalChannelInterceptorProcessor(registry);
registerBuiltInBeans(registry);
registerDefaultConfiguringBeanFactoryPostProcessor(registry);
registerDefaultDatatypeChannelMessageConverter(registry);
registerArgumentResolverMessageConverter(registry);
registerArgumentResolvers(registry);
registerListCapableArgumentResolvers(registry);
if (importingClassMetadata != null) {
registerMessagingAnnotationPostProcessors(importingClassMetadata, registry);
}
registerMessageBuilderFactory(registry);
IntegrationConfigUtils.registerRoleControllerDefinitionIfNecessary(registry);
}
然后注册了MessagingAnnotationPostProcessor这个BeanDefination
private void registerMessagingAnnotationPostProcessors(AnnotationMetadata meta, BeanDefinitionRegistry registry) {
if (!registry.containsBeanDefinition(IntegrationContextUtils.MESSAGING_ANNOTATION_POSTPROCESSOR_NAME)) {
BeanDefinitionBuilder builder =
BeanDefinitionBuilder.genericBeanDefinition(MessagingAnnotationPostProcessor.class)
.setRole(BeanDefinition.ROLE_INFRASTRUCTURE);
registry.registerBeanDefinition(IntegrationContextUtils.MESSAGING_ANNOTATION_POSTPROCESSOR_NAME,
builder.getBeanDefinition());
}
new PublisherRegistrar().registerBeanDefinitions(meta, registry);
}
主要关注MessagingAnnotationPostProcessor,因为是一个BeanPostProcessor,在处理被@ServiceActivator注解了的MessageHandler归属的类对应的Bean的时候的时候,会有一些逻辑:
public class MessagingAnnotationPostProcessor implements BeanPostProcessor, BeanFactoryAware, InitializingBean {
。。。。。。
@Override public void afterPropertiesSet() { Assert.notNull(this.beanFactory, "BeanFactory must not be null"); this.postProcessors.put(Filter.class, new FilterAnnotationPostProcessor(this.beanFactory)); this.postProcessors.put(Router.class, new RouterAnnotationPostProcessor(this.beanFactory)); this.postProcessors.put(Transformer.class, new TransformerAnnotationPostProcessor(this.beanFactory)); this.postProcessors.put(ServiceActivator.class, new ServiceActivatorAnnotationPostProcessor(this.beanFactory)); this.postProcessors.put(Splitter.class, new SplitterAnnotationPostProcessor(this.beanFactory)); this.postProcessors.put(Aggregator.class, new AggregatorAnnotationPostProcessor(this.beanFactory)); this.postProcessors.put(InboundChannelAdapter.class, new InboundChannelAdapterAnnotationPostProcessor(this.beanFactory)); this.postProcessors.put(BridgeFrom.class, new BridgeFromAnnotationPostProcessor(this.beanFactory)); this.postProcessors.put(BridgeTo.class, new BridgeToAnnotationPostProcessor(this.beanFactory)); Map<Class<? extends Annotation>, MethodAnnotationPostProcessor<?>> customPostProcessors = setupCustomPostProcessors(); if (!CollectionUtils.isEmpty(customPostProcessors)) { this.postProcessors.putAll(customPostProcessors); } }
@Override public Object postProcessAfterInitialization(final Object bean, final String beanName) throws BeansException { Assert.notNull(this.beanFactory, "BeanFactory must not be null"); Class<?> beanClass = AopUtils.getTargetClass(bean); // the set will hold records of prior class scans and indicate if no messaging annotations were found if (this.noAnnotationsCache.contains(beanClass)) { return bean; } ReflectionUtils.doWithMethods(beanClass, method -> { Map<Class<? extends Annotation>, List<Annotation>> annotationChains = new HashMap<>(); for (Class<? extends Annotation> annotationType : this.postProcessors.keySet()) { if (AnnotatedElementUtils.isAnnotated(method, annotationType.getName())) { List<Annotation> annotationChain = getAnnotationChain(method, annotationType); if (annotationChain.size() > 0) { annotationChains.put(annotationType, annotationChain); } } } if (StringUtils.hasText(MessagingAnnotationUtils.endpointIdValue(method)) && annotationChains.keySet().size() > 1) { throw new IllegalStateException("@EndpointId on " + method.toGenericString() + " can only have one EIP annotation, found: " + annotationChains.keySet().size()); } for (Entry<Class<? extends Annotation>, List<Annotation>> entry : annotationChains.entrySet()) { Class<? extends Annotation> annotationType = entry.getKey(); List<Annotation> annotations = entry.getValue(); processAnnotationTypeOnMethod(bean, beanName, method, annotationType, annotations); } if (annotationChains.size() == 0) { this.noAnnotationsCache.add(beanClass); } }, ReflectionUtils.USER_DECLARED_METHODS); return bean; } protected void processAnnotationTypeOnMethod(Object bean, String beanName, Method method, Class<? extends Annotation> annotationType, List<Annotation> annotations) { MethodAnnotationPostProcessor<?> postProcessor = MessagingAnnotationPostProcessor.this.postProcessors.get(annotationType); if (postProcessor != null && postProcessor.shouldCreateEndpoint(method, annotations)) { Method targetMethod = method; if (AopUtils.isJdkDynamicProxy(bean)) { try { targetMethod = bean.getClass().getMethod(method.getName(), method.getParameterTypes()); } catch (NoSuchMethodException e) { throw new IllegalArgumentException("Service methods must be extracted to the service " + "interface for JdkDynamicProxy. The affected bean is: ‘" + beanName + "‘ " + "and its method: ‘" + method + "‘", e); } } Object result = postProcessor.postProcess(bean, beanName, targetMethod, annotations); if (result != null && result instanceof AbstractEndpoint) { AbstractEndpoint endpoint = (AbstractEndpoint) result; String autoStartup = MessagingAnnotationUtils.resolveAttribute(annotations, "autoStartup", String.class); if (StringUtils.hasText(autoStartup)) { autoStartup = getBeanFactory().resolveEmbeddedValue(autoStartup); if (StringUtils.hasText(autoStartup)) { endpoint.setAutoStartup(Boolean.parseBoolean(autoStartup)); } } String phase = MessagingAnnotationUtils.resolveAttribute(annotations, "phase", String.class); if (StringUtils.hasText(phase)) { phase = getBeanFactory().resolveEmbeddedValue(phase); if (StringUtils.hasText(phase)) { endpoint.setPhase(Integer.parseInt(phase)); } } Role role = AnnotationUtils.findAnnotation(method, Role.class); if (role != null) { endpoint.setRole(role.value()); } String endpointBeanName = generateBeanName(beanName, method, annotationType); endpoint.setBeanName(endpointBeanName); getBeanFactory().registerSingleton(endpointBeanName, endpoint); getBeanFactory().initializeBean(endpoint, endpointBeanName); } } }
ServiceActivatorAnnotationPostProcessor
public Object postProcess(Object bean, String beanName, Method method, List<Annotation> annotations) { 。。。。。。 MessageHandler handler = createHandler(bean, method, annotations); 。。。。。。 AbstractEndpoint endpoint = createEndpoint(handler, method, annotations); if (endpoint != null) { return endpoint; } return handler; }
protected MessageHandler createHandler(Object bean, Method method, List<Annotation> annotations) {
AbstractReplyProducingMessageHandler serviceActivator;
if (AnnotatedElementUtils.isAnnotated(method, Bean.class.getName())) {
final Object target = this.resolveTargetBeanFromMethodWithBeanAnnotation(method);
serviceActivator = this.extractTypeIfPossible(target, AbstractReplyProducingMessageHandler.class);
if (serviceActivator == null) {
if (target instanceof MessageHandler) {
/*
* Return a reply-producing message handler so that we still get ‘produced no reply‘ messages
* and the super class will inject the advice chain to advise the handler method if needed.
*/
return new ReplyProducingMessageHandlerWrapper((MessageHandler) target);
}
protected AbstractEndpoint createEndpoint(MessageHandler handler, Method method, List<Annotation> annotations) { AbstractEndpoint endpoint = null; String inputChannelName = MessagingAnnotationUtils.resolveAttribute(annotations, getInputChannelAttribute(), String.class); if (StringUtils.hasText(inputChannelName)) { MessageChannel inputChannel; try { inputChannel = this.channelResolver.resolveDestination(inputChannelName); } catch (DestinationResolutionException e) { if (e.getCause() instanceof NoSuchBeanDefinitionException) { inputChannel = new DirectChannel(); this.beanFactory.registerSingleton(inputChannelName, inputChannel); inputChannel = (MessageChannel) this.beanFactory.initializeBean(inputChannel, inputChannelName); } else { throw e; } } Assert.notNull(inputChannel, "failed to resolve inputChannel ‘" + inputChannelName + "‘"); endpoint = doCreateEndpoint(handler, inputChannel, annotations); } return endpoint; } protected AbstractEndpoint doCreateEndpoint(MessageHandler handler, MessageChannel inputChannel, List<Annotation> annotations) { AbstractEndpoint endpoint; if (inputChannel instanceof PollableChannel) { PollingConsumer pollingConsumer = new PollingConsumer((PollableChannel) inputChannel, handler); configurePollingEndpoint(pollingConsumer, annotations); endpoint = pollingConsumer; } else { Poller[] pollers = MessagingAnnotationUtils.resolveAttribute(annotations, "poller", Poller[].class); Assert.state(ObjectUtils.isEmpty(pollers), "A ‘@Poller‘ should not be specified for Annotation-based " + "endpoint, since ‘" + inputChannel + "‘ is a SubscribableChannel (not pollable)."); if (inputChannel instanceof Publisher) { endpoint = new ReactiveStreamsConsumer(inputChannel, handler); } else { endpoint = new EventDrivenConsumer((SubscribableChannel) inputChannel, handler); } } return endpoint; }
创建的endPoint因为实现了SmartLifecycle接口
public class EventDrivenConsumer extends AbstractEndpoint implements IntegrationConsumer { private final SubscribableChannel inputChannel; private final MessageHandler handler; public EventDrivenConsumer(SubscribableChannel inputChannel, MessageHandler handler) { Assert.notNull(inputChannel, "inputChannel must not be null"); Assert.notNull(handler, "handler must not be null"); this.inputChannel = inputChannel; this.handler = handler; this.setPhase(Integer.MIN_VALUE); } @Override public MessageChannel getInputChannel() { return this.inputChannel; } @Override public MessageChannel getOutputChannel() { if (this.handler instanceof MessageProducer) { return ((MessageProducer) this.handler).getOutputChannel(); } else if (this.handler instanceof MessageRouter) { return ((MessageRouter) this.handler).getDefaultOutputChannel(); } else { return null; } } @Override public MessageHandler getHandler() { return this.handler; } @Override protected void doStart() { this.logComponentSubscriptionEvent(true); this.inputChannel.subscribe(this.handler); if (this.handler instanceof Lifecycle) { ((Lifecycle) this.handler).start(); } } @Override protected void doStop() { this.logComponentSubscriptionEvent(false); this.inputChannel.unsubscribe(this.handler); if (this.handler instanceof Lifecycle) { ((Lifecycle) this.handler).stop(); } } private void logComponentSubscriptionEvent(boolean add) { if (this.handler instanceof NamedComponent && this.inputChannel instanceof NamedComponent) { String channelName = ((NamedComponent) this.inputChannel).getComponentName(); String componentType = ((NamedComponent) this.handler).getComponentType(); componentType = StringUtils.hasText(componentType) ? componentType : ""; String componentName = getComponentName(); componentName = (StringUtils.hasText(componentName) && componentName.contains("#")) ? "" : ":" + componentName; StringBuffer buffer = new StringBuffer(); buffer.append("{" + componentType + componentName + "} as a subscriber to the ‘" + channelName + "‘ channel"); if (add) { buffer.insert(0, "Adding "); } else { buffer.insert(0, "Removing "); } logger.info(buffer.toString()); } } }
以上是关于spring集成mqtt客户端相关逻辑分析的主要内容,如果未能解决你的问题,请参考以下文章
Spring Boot 集成 WebSocket,轻松实现信息推送!
spring security oauth2认证中心 集成zuul网关的代码分析