Spring Integration Dispatcher 没有频道订阅者
声明：本页面是StackOverFlow热门问题的中英对照翻译，遵循CC BY-SA 4.0协议，如果您需要使用它，必须同样遵循CC BY-SA许可，注明原文地址和作者信息，同时你必须将它归于原作者（不是我）：StackOverFlow
原文地址: http://stackoverflow.com/questions/41239553/
Warning: these are provided under cc-by-sa 4.0 license. You are free to use/share it, But you must attribute it to the original authors (not me):
StackOverFlow
Spring Integration Dispatcher has no subscribers for channel
提问by Angelo Immediata
I'm using spring integration and its support for MQTT; I saw the spring integration documentation and my simple test case is to publish a message on a MQTT topic. The Spring documentation is located here: http://docs.spring.io/spring-integration/reference/html/mqtt.html#_configuring_with_java_configuration_15
我正在使用 spring 集成及其对 MQTT 的支持；我看到了 spring 集成文档，我的简单测试用例是在 MQTT 主题上发布消息。Spring 文档位于此处：http: //docs.spring.io/spring-integration/reference/html/mqtt.html#_configuring_with_java_configuration_15
I'm using these versions:
我正在使用这些版本：
- spring 4.3.4
- spring integration 4.3.5
- 弹簧 4.3.4
- 弹簧集成 4.3.5
I built this simple configuration class:
我构建了这个简单的配置类：
@Configuration
@IntegrationComponentScan
public class CommunicationServerApplication
{
@Bean
public MqttPahoClientFactory mqttClientFactory()
{
DefaultMqttPahoClientFactory factory = new DefaultMqttPahoClientFactory();
factory.setServerURIs(mqttServerUris);
if (StringUtils.hasText(mqttUsername) && StringUtils.hasText(mqttPassword))
{
factory.setUserName(mqttUsername);
factory.setPassword(mqttPassword);
}
factory.setConnectionTimeout(mqttConnectionTimeout);
factory.setKeepAliveInterval(mqttKeepAliveInterval);
factory.setPersistence(new MqttDefaultFilePersistence(mqttPersistenceFileDirectory));
return factory;
}
@Bean
@ServiceActivator(inputChannel = "mqttOutboundChannel", autoStartup="true")
public MessageHandler mqttOutbound()
{
String clientId = mqttClientId;
if( !StringUtils.hasText(clientId) )
{
clientId = UUID.randomUUID().toString();
}
MqttPahoMessageHandler messageHandler = new MqttPahoMessageHandler(clientId, mqttClientFactory());
messageHandler.setAsync(true);
messageHandler.setDefaultTopic(mqttTopic);
if( mqttQos >= 0 && mqttQos <=2 )
{
messageHandler.setDefaultQos(mqttQos);
}
return messageHandler;
}
@Bean
public MessageChannel mqttOutboundChannel()
{
DirectChannel dc = new DirectChannel();
return dc;
}
@MessagingGateway(defaultRequestChannel = "mqttOutboundChannel")
public interface MqttMsgproducer
{
void sendToMqtt(String data);
}
}
And then I used this simple test case:
然后我使用了这个简单的测试用例：
@ContextConfiguration(value ={ "classpath:app-ctx.xml"})
@RunWith(SpringJUnit4ClassRunner.class)
public class SimpleMqttTestSuite
{
private static final Logger logger = LoggerFactory.getLogger(SimpleMqttTestSuite.class.getName());
@Autowired
private MqttMsgproducer sender;
@Test
public void startServerTest()
{
try
{
sender.sendToMqtt("Hello");
}
catch (Exception e)
{
logger.error("Error", e);
}
}
}
My app-ctx.xml is:
我的 app-ctx.xml 是：
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:context="http://www.springframework.org/schema/context"
xmlns:p="http://www.springframework.org/schema/p" xmlns:tx="http://www.springframework.org/schema/tx"
xmlns:mvc="http://www.springframework.org/schema/mvc" xmlns:oxm="http://www.springframework.org/schema/oxm"
xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context.xsd
http://www.springframework.org/schema/tx http://www.springframework.org/schema/tx/spring-tx.xsd
http://www.springframework.org/schema/oxm http://www.springframework.org/schema/oxm/spring-oxm.xsd
http://www.springframework.org/schema/mvc http://www.springframework.org/schema/mvc/spring-mvc.xsd">
<context:component-scan base-package="it.olegna.test.integration" />
<context:property-placeholder location="classpath:configuration.properties"
order="0" ignore-resource-not-found="true" ignore-unresolvable="true" />
</beans>
Executing the simple test, I'm having this error:
执行简单的测试，我有这个错误：
2016-12-20 10:46:33,889 49967 [nioEventLoopGroup-3-1] ERROR - Errore
org.springframework.messaging.MessageDeliveryException: Dispatcher has no subscribers for channel 'org.springframework.context.support.GenericApplicationContext@2e6a8155.mqttOutboundChannel'.; nested exception is org.springframework.integration.MessageDispatchingException: Dispatcher has no subscribers
at org.springframework.integration.channel.AbstractSubscribableChannel.doSend(AbstractSubscribableChannel.java:81) ~[spring-integration-core-4.3.5.RELEASE.jar:4.3.5.RELEASE]
at org.springframework.integration.channel.AbstractMessageChannel.send(AbstractMessageChannel.java:423) ~[spring-integration-core-4.3.5.RELEASE.jar:4.3.5.RELEASE]
at org.springframework.integration.channel.AbstractMessageChannel.send(AbstractMessageChannel.java:373) ~[spring-integration-core-4.3.5.RELEASE.jar:4.3.5.RELEASE]
at org.springframework.messaging.core.GenericMessagingTemplate.doSend(GenericMessagingTemplate.java:115) ~[spring-messaging-4.3.4.RELEASE.jar:4.3.4.RELEASE]
at org.springframework.messaging.core.GenericMessagingTemplate.doSend(GenericMessagingTemplate.java:45) ~[spring-messaging-4.3.4.RELEASE.jar:4.3.4.RELEASE]
at org.springframework.messaging.core.AbstractMessageSendingTemplate.send(AbstractMessageSendingTemplate.java:105) ~[spring-messaging-4.3.4.RELEASE.jar:4.3.4.RELEASE]
at org.springframework.messaging.core.AbstractMessageSendingTemplate.convertAndSend(AbstractMessageSendingTemplate.java:143) ~[spring-messaging-4.3.4.RELEASE.jar:4.3.4.RELEASE]
at org.springframework.messaging.core.AbstractMessageSendingTemplate.convertAndSend(AbstractMessageSendingTemplate.java:135) ~[spring-messaging-4.3.4.RELEASE.jar:4.3.4.RELEASE]
at org.springframework.integration.gateway.MessagingGatewaySupport.send(MessagingGatewaySupport.java:375) ~[spring-integration-core-4.3.5.RELEASE.jar:4.3.5.RELEASE]
at org.springframework.integration.gateway.GatewayProxyFactoryBean.invokeGatewayMethod(GatewayProxyFactoryBean.java:477) ~[spring-integration-core-4.3.5.RELEASE.jar:4.3.5.RELEASE]
at org.springframework.integration.gateway.GatewayProxyFactoryBean.doInvoke(GatewayProxyFactoryBean.java:429) ~[spring-integration-core-4.3.5.RELEASE.jar:4.3.5.RELEASE]
at org.springframework.integration.gateway.GatewayProxyFactoryBean.invoke(GatewayProxyFactoryBean.java:420) ~[spring-integration-core-4.3.5.RELEASE.jar:4.3.5.RELEASE]
at org.springframework.integration.gateway.GatewayCompletableFutureProxyFactoryBean.invoke(GatewayCompletableFutureProxyFactoryBean.java:65) ~[spring-integration-core-4.3.5.RELEASE.jar:4.3.5.RELEASE]
at org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:179) ~[spring-aop-4.3.4.RELEASE.jar:4.3.4.RELEASE]
at org.springframework.aop.framework.JdkDynamicAopProxy.invoke(JdkDynamicAopProxy.java:213) ~[spring-aop-4.3.4.RELEASE.jar:4.3.4.RELEASE]
I can't figure what I'm missing in the configuration. Can anybody give to me a tip?
我无法弄清楚配置中缺少什么。有人可以给我一个提示吗？
thank you
谢谢你
Angelo
安杰洛
回答by Angelo Immediata
I solved my issue
我解决了我的问题
It was related to the fact that I built the Channel but now handler have been subscribed
这与我建立了频道但现在处理程序已被订阅的事实有关
In my application class I did the following:
在我的应用程序类中，我执行了以下操作：
@Bean
public MessageChannel mqttOutboundChannel()
{
DirectChannel dc = new DirectChannel();
dc.subscribe(mqttOutbound());
return dc;
}
As you can see now I manually add subscribe the bean mqttOutbound (the message handler) to the Channel
正如您现在看到的，我手动将订阅 bean mqttOutbound（消息处理程序）添加到 Channel
By doing in this way all works
通过这样做所有的工作
I hope this can help
我希望这可以帮助
Angelo
安杰洛
UPDATE AFTER Gary Russell ANSWER
加里罗素回答后更新
As suggested by Gary Russell I didn't subscribe to the Channel
正如 Gary Russell 所建议的，我没有订阅频道
I added the annotation @EnableIntegration
我添加了注释 @EnableIntegration
So my Application class now is the following:
所以我的 Application 类现在如下：
@Configuration
@IntegrationComponentScan
@EnableIntegration
public class CommunicationServerApplication
{
@Bean
public MqttPahoClientFactory mqttClientFactory()
{
DefaultMqttPahoClientFactory factory = new DefaultMqttPahoClientFactory();
factory.setServerURIs(mqttServerUris);
if (StringUtils.hasText(mqttUsername) && StringUtils.hasText(mqttPassword))
{
factory.setUserName(mqttUsername);
factory.setPassword(mqttPassword);
}
factory.setConnectionTimeout(mqttConnectionTimeout);
factory.setKeepAliveInterval(mqttKeepAliveInterval);
factory.setPersistence(new MqttDefaultFilePersistence(mqttPersistenceFileDirectory));
return factory;
}
@Bean
@ServiceActivator(inputChannel = "mqttOutboundChannel", autoStartup="true")
public MessageHandler mqttOutbound()
{
String clientId = mqttClientId;
if( !StringUtils.hasText(clientId) )
{
clientId = UUID.randomUUID().toString();
}
MqttPahoMessageHandler messageHandler = new MqttPahoMessageHandler(clientId, mqttClientFactory());
messageHandler.setAsync(true);
messageHandler.setDefaultTopic(mqttTopic);
if( mqttQos >= 0 && mqttQos <=2 )
{
messageHandler.setDefaultQos(mqttQos);
}
return messageHandler;
}
@Bean
public MessageChannel mqttOutboundChannel()
{
DirectChannel dc = new DirectChannel();
return dc;
}
@MessagingGateway(defaultRequestChannel = "mqttOutboundChannel")
public interface MqttMsgproducer
{
void sendToMqtt(String data);
}
}
回答by Gary Russell
Your solution is incorrect - you must notsubscribe within a channel bean definition. I believe your problem is that you are missing @EnableIntegrationon the class.
您的解决方案不正确 - 您不得在频道 bean 定义中订阅。我相信你的问题是你缺课@EnableIntegration了。
