我已经用spring-Xd创建了流,例如- stream create mqtttestfile --definition "mqtt --url='tcp://localhost:1883' --topics='helloTopic' | file" --deploy 创建并部署了新的流“ mqtttestfile”。我还检查了localhost:9393 / admin-ui,成功创建并部署了流。
stream create mqtttestfile --definition "mqtt --url='tcp://localhost:1883' --topics='helloTopic' | file" --deploy
我的MQTT代理在localhost:1883上运行。但是当我检查/ tmp / xd / output文件目录时,缺少mqtttestfile.out文件。
我需要对我的假设的以下几点进行澄清:
我认为MQTT客户端已在spring-xd源mqtt模块中配置。因此,当我们创建流时,它将自动订阅代理上的特定主题。
我还尝试了运行两个单独的python脚本,一个用于订阅,另一个用于在单独的终端上的发布者,并且运行良好。所以mqtt经纪人没问题。
这是我得到的spring-xd控制台的日志:
istener-在15000毫秒内将部署调度到新容器
2017-03-04T12:07:51 + 0530 1.3.0.RELEASE INFO DeploymentsPathChildrenCache-0容器.DeploymentListener-路径缓存事件:path = / deployments / modules / allocated / d634d310-12b4-4a83-baea-c1c98dfb7bba / mqtttestfile.sink .file.1,类型= CHILD_ADDED
2017-03-04T12:07:51 + 0530 1.3.0.RELEASE INFO DeploymentsPathChildrenCache-0容器.DeploymentListener-为流’mqtttestfile’部署模块’file’2017-03-04T12:07:52 + 0530 1.3.0.RELEASE INFO DeploymentsPathChildrenCache-0容器。DeploymentListener-部署模块[ModuleDescriptor @ 7e658391 moduleName =‘file’,moduleLabel =‘file’,group =‘mqtttestfile’,sourceChannelName = [null],sinkChannelName = [null],索引= 1,类型=接收器,参数=映射[[空]],子代=列表[[空]]]
2017-03-04T12:07:54 + 0530 1.3.0.RELEASE INFO DeploymentsPathChildrenCache-0容器.DeploymentListener-路径缓存事件:path = / deployments / modules / allocated / d634d310-12b4-4a83-baea-c1c98dfb7bba / mqtttestfile.source .mqtt.1,类型= CHILD_ADDED
2017-03-04T12:07:54 + 0530 1.3.0.RELEASE INFO DeploymentsPathChildrenCache-0容器.DeploymentListener-为流’mqtttestfile’部署模块’mqtt’2017-03-04T12:07:54 + 0530 1.3.0.RELEASE INFO DeploymentsPathChildrenCache-0容器。DeploymentListener-部署模块[ModuleDescriptor @ 5a7d8e37 moduleName =‘mqtt’,moduleLabel =‘mqtt’,group =‘mqtttestfile’,sourceChannelName = [null],sinkChannelName = [null],索引= 0,类型=源,参数= map [‘topics’->‘helloTopic’,‘url’->‘tcp:// localhost:1883’],子代= list [[空]]]
2017-03-04T12:07:56 + 0530 1.3.0.RELEASE INFO DeploymentSupervisor-0 zk.ZKStreamDeploymentHandler-流’mqtttestfile’的部署状态:DeploymentStatus {state = deployed}
使用spring-xd 1.3.1仍无法解决问题,这是我在日志中看到的错误消息,
2017-03-05T01:15:06 + 0530 1.3.1.RELEASE INFO LeaderSelector-1 zk.DeploymentSupervisor-由于线程中断而取消了领导权
2017-03-05T01:15:06 + 0530 1.3.1。释放错误MQTT记录:xd.mqtt.client.id.src inbound.MqttPahoMessageDrivenChannelAdapter-丢失的连接:连接丢失; 正在重试…
谢谢。