如何向mqtt代理注册springxd源mqtt

Question

我用spring创建了流，比如-

stream create mqtttestfile --definition "mqtt --url='tcp://localhost:1883' --topics='helloTopic' | file" --deploy

创建并部署新流“mqtttestfile”。我还检查了localhost:9393/admin-ui，成功创建和部署了流。

我的MQTT代理运行在localhost上:1883。但是，当我检查/tmp/xd/output文件目录时，mqtttestfile.out文件就丢失了。

我需要就我假设的以下几点作出澄清：

1. 我认为MQTT客户端已经配置在spring源mqtt模块中。因此，当我们创建流时，它将自动订阅代理上的特定主题。
2. 我还尝试运行两个单独的python脚本，一个用于订阅，另一个用于在单独的终端上发布，而且运行良好。所以mqtt经纪人没问题。

这是我从spring控制台得到的日志：

将部署到新容器的时间安排在15000 ms内。

2017-03-04T12:07:51+0530 1.3.0.信息配置-路径儿童cache -0 container.DeploymentListener -路径缓存事件:container.DeploymentListener type=CHILD_ADDED

moduleName信息部署模块[ModuleDescriptor@7e658391 container.DeploymentListener = 'file'，moduleLabel =‘moduleLabel’，group = 'mqtttestfile'，sourceChannelName =空，sinkChannelName =空，索引= 1，类型=接收器，参数= map[empty]，list[empty]]

2017-03-04T12:07:54+0530 1.3.0

moduleName信息部署模块[ModuleDescriptor@5a7d8e37 container.DeploymentListener = 'mqtt'，moduleLabel = 'mqtttestfile‘组= 'mqtttestfile'，sourceChannelName = null，sinkChannelName =，index = 0，type =源，参数=’->‘helloTopic，'url’->：/->：1883，主机:1883，儿童= list[empty]]

2017-03-04T12:07:56+0530 1.3.0 Deployment信息DeploymentSupervisor-0 zk.ZKStreamDeploymentHandler -流‘mqtttestfile’的部署状态: DeploymentStatus{state=deployed}

在spring 1.3.1中，问题仍未解决，这是我在日志中看到的错误消息-

2017-03-05T01:15:06+0530 1.3.1.RELEASE INFO LeaderSelector-1 zk.DeploymentSupervisor -由于线程中断而取消领导

Rec错误MQTT Rec: xd.mqtt.client.id.src inbound.MqttPahoMessageDrivenChannelAdapter -失去连接:连接丢失；重试

谢谢。

Answer 1

我刚刚测试了你的溪流对我来说很好..。

$ cat /tmp/xd/output/mqtttestfile.out 
foo
bar

(在我向队列中添加消息foo和bar之后)。

为org.springframework.integration启用了调试日志记录(在容器的logback.grooy文件中)，我看到.

2017-03-04T08:02:59-0500 1.3.1.RELEASE INFO DeploymentsPathChildrenCache-0 endpoint.EventDrivenConsumer - started outbound.mqtttestfile.0
2017-03-04T08:02:59-0500 1.3.1.RELEASE DEBUG DeploymentsPathChildrenCache-0 inbound.MqttPahoMessageDrivenChannelAdapter - Connected and subscribed to [helloTopic]
2017-03-04T08:02:59-0500 1.3.1.RELEASE INFO DeploymentsPathChildrenCache-0 inbound.MqttPahoMessageDrivenChannelAdapter - started mqttInbound
2017-03-04T08:02:59-0500 1.3.1.RELEASE INFO DeploymentSupervisor-0 zk.ZKStreamDeploymentHandler - Deployment status for stream 'mqtttestfile': DeploymentStatus{state=deployed}

奇怪的是，您没有看到...started...消息(信息)。

你能试试1.3.1版吗？