{"id":2983,"date":"2023-07-14T17:56:42","date_gmt":"2023-07-14T09:56:42","guid":{"rendered":"https:\/\/xinchewhd.com.cn\/?p=2983"},"modified":"2023-07-14T17:56:43","modified_gmt":"2023-07-14T09:56:43","slug":"%e9%80%9a%e8%bf%87%e5%b7%a5%e5%8e%82%e6%a8%a1%e5%bc%8f%e5%ae%9e%e7%8e%b0springbootmqtt-%e8%ae%a2%e9%98%85%e4%b8%8e%e6%b6%88%e8%b4%b9","status":"publish","type":"post","link":"https:\/\/xinchewhd.com.cn\/index.php\/spring-boot\/%e9%80%9a%e8%bf%87%e5%b7%a5%e5%8e%82%e6%a8%a1%e5%bc%8f%e5%ae%9e%e7%8e%b0springbootmqtt-%e8%ae%a2%e9%98%85%e4%b8%8e%e6%b6%88%e8%b4%b9\/","title":{"rendered":"\u901a\u8fc7\u5de5\u5382\u6a21\u5f0f\u5b9e\u73b0SpringBoot+MQTT-\u8ba2\u9605\u4e0e\u6d88\u8d39"},"content":{"rendered":"

1.pom.xml<\/span><\/h3>\n
<!--        \u8fde\u63a5MQTT--><\/span><\/span>\n        <!--mqtt--><\/span><\/span>\n        <<\/span>dependency<\/span>><\/span><\/span>\n            <<\/span>groupId<\/span>><\/span>org.springframework.boot<\/<\/span>groupId<\/span>><\/span><\/span>\n            <<\/span>artifactId<\/span>><\/span>spring-boot-starter-integration<\/<\/span>artifactId<\/span>><\/span><\/span>\n        <\/<\/span>dependency<\/span>><\/span><\/span>\n        <<\/span>dependency<\/span>><\/span><\/span>\n            <<\/span>groupId<\/span>><\/span>org.springframework.integration<\/<\/span>groupId<\/span>><\/span><\/span>\n            <<\/span>artifactId<\/span>><\/span>spring-integration-stream<\/<\/span>artifactId<\/span>><\/span><\/span>\n        <\/<\/span>dependency<\/span>><\/span><\/span>\n        <<\/span>dependency<\/span>><\/span><\/span>\n            <<\/span>groupId<\/span>><\/span>org.springframework.integration<\/<\/span>groupId<\/span>><\/span><\/span>\n            <<\/span>artifactId<\/span>><\/span>spring-integration-mqtt<\/<\/span>artifactId<\/span>><\/span><\/span>\n        <\/<\/span>dependency<\/span>><\/span><\/span><\/pre>\n
2.yml\u914d\u7f6e<\/span><\/h3>\n
mqtt<\/span>:<\/span><\/span>\n  config<\/span>:<\/span><\/span>\n    username<\/span>: <\/span>admin<\/span>\n    password<\/span>: <\/span>admin # \u5bc6\u7801<\/span><\/span>\n    hostUrl<\/span>: <\/span>tcp:<\/span>\/\/xx.xx.xxx.xx:<\/span>1883 # tcp:\/\/ip:\u7aef\u53e3<\/span><\/span>\n    clientId<\/span>: <\/span>emq-ss-client # \u5ba2\u6237\u7aefid<\/span><\/span>\n    defaultTopic<\/span>: <\/span>topic_default # \u8ba2\u9605\u4e3b\u9898<\/span><\/span>\n    timeout<\/span>: <\/span>1000 # \u8d85\u65f6\u65f6\u95f4 \uff08\u5355\u4f4d\uff1a\u79d2\uff09<\/span><\/span>\n    keepalive<\/span>: <\/span>60 # \u5fc3\u8df3 \uff08\u5355\u4f4d\uff1a\u79d2\uff09<\/span><\/span>\n    enabled<\/span>: <\/span>false # \u662f\u5426\u4f7f\u80fdmqtt\u529f\u80fd<\/span><\/span><\/pre>\n
3.MqttProperties<\/span><\/h3>\n
package<\/span> com<\/span>.rfos<\/span>.assistsilkworm<\/span>.mqtt<\/span>.properties<\/span>;<\/span>\n<\/span>\nimport<\/span> io<\/span>.swagger<\/span>.annotations<\/span>.ApiModelProperty<\/span>;<\/span>\nimport<\/span> lombok<\/span>.Data<\/span>;<\/span>\nimport<\/span> org<\/span>.springframework<\/span>.boot<\/span>.context<\/span>.properties<\/span>.ConfigurationProperties<\/span>;<\/span>\nimport<\/span> org<\/span>.springframework<\/span>.stereotype<\/span>.Component<\/span>;<\/span>\n<\/span>\n\/**<\/span><\/span>\n * @Author Michale<\/span><\/span>\n * @CreateDate 2022\/9\/4<\/span><\/span>\n * @Describe \u8bfb\u53d6mqtt\u914d\u7f6e\u4fe1\u606f<\/span><\/span>\n *\/<\/span><\/span>\n@Data<\/span><\/span>\n@Component<\/span><\/span>\n@ConfigurationProperties<\/span>(prefix<\/span> =<\/span> \"mqtt.config\"<\/span>)<\/span>\npublic<\/span> class<\/span> MqttProperties<\/span> {<\/span>\n<\/span>\n    @ApiModelProperty<\/span>(\"\u7528\u6237\u540d\"<\/span>)<\/span>\n    private<\/span> String<\/span> username<\/span>;<\/span>\n<\/span>\n    @ApiModelProperty<\/span>(\"\u5bc6\u7801\"<\/span>)<\/span>\n    private<\/span> String<\/span> password<\/span>;<\/span>\n<\/span>\n    @ApiModelProperty<\/span>(\"\u5730\u5740\"<\/span>)<\/span>\n    private<\/span> String<\/span> hostUrl<\/span>;<\/span>\n<\/span>\n    @ApiModelProperty<\/span>(\"\u5ba2\u6237\u7aefid\"<\/span>)<\/span>\n    private<\/span> String<\/span> clientId<\/span>;<\/span>\n<\/span>\n    @ApiModelProperty<\/span>(\"\u8ba2\u9605\u4e3b\u9898\"<\/span>)<\/span>\n    private<\/span> String<\/span> defaultTopic<\/span>;<\/span>\n<\/span>\n    @ApiModelProperty<\/span>(\"\u8d85\u65f6\u65f6\u95f4\"<\/span>)<\/span>\n    private<\/span> int<\/span> timeout<\/span>;<\/span>\n<\/span>\n    @ApiModelProperty<\/span>(\"\u5fc3\u8df3\"<\/span>)<\/span>\n    private<\/span> int<\/span> keepalive<\/span>;<\/span>\n<\/span>\n    @ApiModelProperty<\/span>(\"mqtt\u5f00\u5173\"<\/span>)<\/span>\n    private<\/span> boolean<\/span> enabled<\/span>;<\/span>\n}<\/span>\n<\/span><\/pre>\n
\n
4.VO\u4f20\u8f93\u5b9e\u4f53\u7c7b<\/span><\/h3>\n
package<\/span> com<\/span>.rfos<\/span>.assistsilkworm<\/span>.mqtt<\/span>.vo<\/span>;<\/span>\n<\/span>\nimport<\/span> io<\/span>.swagger<\/span>.annotations<\/span>.ApiModelProperty<\/span>;<\/span>\nimport<\/span> lombok<\/span>.Data<\/span>;<\/span>\n<\/span>\n\/**<\/span><\/span>\n * @Author rfos<\/span><\/span>\n * @Date 2023\/2\/20 17:08<\/span><\/span>\n * @Description TODO \u6d88\u606f\u4f53<\/span><\/span>\n *\/<\/span><\/span>\n@Data<\/span><\/span>\npublic<\/span> class<\/span> MqttVO<\/span> {<\/span>\n    @ApiModelProperty<\/span>(\"\u8ba2\u9605\u7684\u4e3b\u9898\"<\/span>)<\/span>\n    public<\/span>   String<\/span> topic<\/span> ;<\/span>\n<\/span>\n    @ApiModelProperty<\/span>(\"\u53d1\u9001\u7684\u5185\u5bb9\"<\/span>)<\/span>\n    public<\/span>   String<\/span> payload<\/span> ;<\/span>\n}<\/span>\n<\/span><\/pre>\n
5.MqttReceiveHandle<\/span><\/h3>\n
\n
package<\/span> com<\/span>.rfos<\/span>.assistsilkworm<\/span>.mqtt<\/span>.handler<\/span>;<\/span>\n<\/span>\nimport<\/span> lombok<\/span>.extern<\/span>.slf4j<\/span>.Slf4j<\/span>;<\/span>\nimport<\/span> org<\/span>.springframework<\/span>.integration<\/span>.annotation<\/span>.ServiceActivator<\/span>;<\/span>\nimport<\/span> org<\/span>.springframework<\/span>.integration<\/span>.mqtt<\/span>.support<\/span>.MqttHeaders<\/span>;<\/span>\nimport<\/span> org<\/span>.springframework<\/span>.messaging<\/span>.Message<\/span>;<\/span>\nimport<\/span> org<\/span>.springframework<\/span>.messaging<\/span>.MessageHandler<\/span>;<\/span>\nimport<\/span> org<\/span>.springframework<\/span>.messaging<\/span>.MessagingException<\/span>;<\/span>\nimport<\/span> org<\/span>.springframework<\/span>.stereotype<\/span>.Component<\/span>;<\/span>\n<\/span>\n\/**<\/span><\/span>\n * mqtt\u5ba2\u6237\u7aef\u6d88\u606f\u5904\u7406\u7c7b<\/span><\/span>\n **\/<\/span><\/span>\n@Slf4j<\/span><\/span>\n@Component<\/span><\/span>\npublic<\/span> class<\/span> MqttReceiveHandler<\/span> implements<\/span> MessageHandler<\/span> {<\/span>\n<\/span>\n<\/span>\n<\/span>\n    @Override<\/span><\/span>\n    @ServiceActivator<\/span>(inputChannel<\/span> =<\/span> \"MQTT_INPUT_CHANNEL\"<\/span>)<\/span>\n    public<\/span> void<\/span> handleMessage<\/span>(Message<\/span><?><\/span> message<\/span>) throws<\/span> MessagingException<\/span> {<\/span>\n        log<\/span>.info<\/span>(\"\u6536\u5230\u8ba2\u9605\u6d88\u606f: {}\"<\/span>, message<\/span>);<\/span>\n        String<\/span> topic<\/span> =<\/span> message<\/span>.getHeaders<\/span>().get<\/span>(MqttHeaders<\/span>.RECEIVED_TOPIC<\/span>).toString<\/span>();<\/span>\n        log<\/span>.info<\/span>(\"\u6d88\u606f\u4e3b\u9898\uff1a{}\"<\/span>, topic<\/span>);<\/span>\n    }<\/span>\n}<\/span><\/pre>\n
\n
6.MqttConfig<\/span><\/h3>\n
package<\/span> com<\/span>.rfos<\/span>.assistsilkworm<\/span>.mqtt<\/span>.config<\/span>;<\/span>\n<\/span>\nimport<\/span> com<\/span>.rfos<\/span>.assistsilkworm<\/span>.mqtt<\/span>.handler<\/span>.MqttReceiveHandler<\/span>;<\/span>\nimport<\/span> com<\/span>.rfos<\/span>.assistsilkworm<\/span>.mqtt<\/span>.properties<\/span>.MqttProperties<\/span>;<\/span>\nimport<\/span> lombok<\/span>.extern<\/span>.slf4j<\/span>.Slf4j<\/span>;<\/span>\nimport<\/span> org<\/span>.eclipse<\/span>.paho<\/span>.client<\/span>.mqttv3<\/span>.MqttConnectOptions<\/span>;<\/span>\nimport<\/span> org<\/span>.springframework<\/span>.beans<\/span>.factory<\/span>.annotation<\/span>.Autowired<\/span>;<\/span>\nimport<\/span> org<\/span>.springframework<\/span>.context<\/span>.annotation<\/span>.Bean<\/span>;<\/span>\nimport<\/span> org<\/span>.springframework<\/span>.context<\/span>.annotation<\/span>.Configuration<\/span>;<\/span>\nimport<\/span> org<\/span>.springframework<\/span>.integration<\/span>.annotation<\/span>.IntegrationComponentScan<\/span>;<\/span>\nimport<\/span> org<\/span>.springframework<\/span>.integration<\/span>.annotation<\/span>.ServiceActivator<\/span>;<\/span>\nimport<\/span> org<\/span>.springframework<\/span>.integration<\/span>.channel<\/span>.DirectChannel<\/span>;<\/span>\nimport<\/span> org<\/span>.springframework<\/span>.integration<\/span>.mqtt<\/span>.core<\/span>.DefaultMqttPahoClientFactory<\/span>;<\/span>\nimport<\/span> org<\/span>.springframework<\/span>.integration<\/span>.mqtt<\/span>.core<\/span>.MqttPahoClientFactory<\/span>;<\/span>\nimport<\/span> org<\/span>.springframework<\/span>.integration<\/span>.mqtt<\/span>.inbound<\/span>.MqttPahoMessageDrivenChannelAdapter<\/span>;<\/span>\nimport<\/span> org<\/span>.springframework<\/span>.integration<\/span>.mqtt<\/span>.outbound<\/span>.MqttPahoMessageHandler<\/span>;<\/span>\nimport<\/span> org<\/span>.springframework<\/span>.integration<\/span>.mqtt<\/span>.support<\/span>.DefaultPahoMessageConverter<\/span>;<\/span>\nimport<\/span> org<\/span>.springframework<\/span>.messaging<\/span>.Message<\/span>;<\/span>\nimport<\/span> org<\/span>.springframework<\/span>.messaging<\/span>.MessageChannel<\/span>;<\/span>\nimport<\/span> org<\/span>.springframework<\/span>.messaging<\/span>.MessageHandler<\/span>;<\/span>\nimport<\/span> org<\/span>.springframework<\/span>.messaging<\/span>.MessagingException<\/span>;<\/span>\n<\/span>\nimport<\/span> java<\/span>.util<\/span>.Arrays<\/span>;<\/span>\nimport<\/span> java<\/span>.util<\/span>.List<\/span>;<\/span>\n<\/span>\n\/**<\/span><\/span>\n * mqtt \u63a8\u9001and\u63a5\u6536 \u6d88\u606f\u7c7b<\/span><\/span>\n **\/<\/span><\/span>\n@Configuration<\/span><\/span>\n@IntegrationComponentScan<\/span><\/span>\n@Slf4j<\/span><\/span>\npublic<\/span> class<\/span> MqttConfig<\/span> {<\/span>\n<\/span>\n    @Autowired<\/span><\/span>\n    private<\/span> MqttProperties<\/span> mqttProperties<\/span>;<\/span>\n    @Autowired<\/span><\/span>\n    private<\/span> MqttReceiveHandler<\/span> mqttReceiveHandle<\/span>;<\/span>\n<\/span>\n    \/**<\/span><\/span>\n     * MQTT\u8fde\u63a5\u5668\u9009\u9879<\/span><\/span>\n     **\/<\/span><\/span>\n    @Bean<\/span>(value<\/span> =<\/span> \"getMqttConnectOptions\"<\/span>)<\/span>\n    public<\/span> MqttConnectOptions<\/span> getMqttConnectOptions<\/span>() {<\/span>\n        MqttConnectOptions<\/span> mqttConnectOptions<\/span> =<\/span> new<\/span> MqttConnectOptions<\/span>();<\/span>\n        \/\/ \u8bbe\u7f6e\u662f\u5426\u6e05\u7a7asession,\u8fd9\u91cc\u5982\u679c\u8bbe\u7f6e\u4e3afalse\u8868\u793a\u670d\u52a1\u5668\u4f1a\u4fdd\u7559\u5ba2\u6237\u7aef\u7684\u8fde\u63a5\u8bb0\u5f55\uff0c\u8fd9\u91cc\u8bbe\u7f6e\u4e3atrue\u8868\u793a\u6bcf\u6b21\u8fde\u63a5\u5230\u670d\u52a1\u5668\u90fd\u4ee5\u65b0\u7684\u8eab\u4efd\u8fde\u63a5<\/span><\/span>\n        mqttConnectOptions<\/span>.setCleanSession<\/span>(true<\/span>);<\/span>\n        \/\/ \u8bbe\u7f6e\u8d85\u65f6\u65f6\u95f4 \u5355\u4f4d\u4e3a\u79d2<\/span><\/span>\n        mqttConnectOptions<\/span>.setConnectionTimeout<\/span>(10<\/span>);<\/span>\n        mqttConnectOptions<\/span>.setAutomaticReconnect<\/span>(true<\/span>);<\/span>\n        mqttConnectOptions<\/span>.setUserName<\/span>(mqttProperties<\/span>.getUsername<\/span>());<\/span>\n        mqttConnectOptions<\/span>.setPassword<\/span>(mqttProperties<\/span>.getPassword<\/span>().toCharArray<\/span>());<\/span>\n        mqttConnectOptions<\/span>.setServerURIs<\/span>(new<\/span> String<\/span>[]{mqttProperties<\/span>.getHostUrl<\/span>()});<\/span>\n        \/\/ \u8bbe\u7f6e\u4f1a\u8bdd\u5fc3\u8df3\u65f6\u95f4 \u5355\u4f4d\u4e3a\u79d2 \u670d\u52a1\u5668\u4f1a\u6bcf\u96941.5*20\u79d2\u7684\u65f6\u95f4\u5411\u5ba2\u6237\u7aef\u53d1\u9001\u5fc3\u8df3\u5224\u65ad\u5ba2\u6237\u7aef\u662f\u5426\u5728\u7ebf\uff0c\u4f46\u8fd9\u4e2a\u65b9\u6cd5\u5e76\u6ca1\u6709\u91cd\u8fde\u7684\u673a\u5236<\/span><\/span>\n        mqttConnectOptions<\/span>.setKeepAliveInterval<\/span>(10<\/span>);<\/span>\n        \/\/ \u8bbe\u7f6e\u201c\u9057\u5631\u201d\u6d88\u606f\u7684\u8bdd\u9898\uff0c\u82e5\u5ba2\u6237\u7aef\u4e0e\u670d\u52a1\u5668\u4e4b\u95f4\u7684\u8fde\u63a5\u610f\u5916\u4e2d\u65ad\uff0c\u670d\u52a1\u5668\u5c06\u53d1\u5e03\u5ba2\u6237\u7aef\u7684\u201c\u9057\u5631\u201d\u6d88\u606f\u3002<\/span><\/span>\n        \/\/mqttConnectOptions.setWill(\"willTopic\", WILL_DATA, 2, false);<\/span><\/span>\n        return<\/span> mqttConnectOptions<\/span>;<\/span>\n    }<\/span>\n<\/span>\n    \/**<\/span><\/span>\n     * MQTT\u5de5\u5382<\/span><\/span>\n     **\/<\/span><\/span>\n    @Bean<\/span>(\"mqttClientFactor\"<\/span>)<\/span>\n    public<\/span> MqttPahoClientFactory<\/span> mqttClientFactory<\/span>() {<\/span>\n        DefaultMqttPahoClientFactory<\/span> factory<\/span> =<\/span> new<\/span> DefaultMqttPahoClientFactory<\/span>();<\/span>\n        factory<\/span>.setConnectionOptions<\/span>(getMqttConnectOptions<\/span>());<\/span>\n        return<\/span> factory<\/span>;<\/span>\n    }<\/span>\n<\/span>\n<\/span>\n<\/span>\n    \/**<\/span><\/span>\n     * MQTT\u4fe1\u606f\u901a\u9053\uff08\u751f\u4ea7\u8005\uff09<\/span><\/span>\n     **\/<\/span><\/span>\n    @Bean<\/span>(\"MQTT_OUT_BOUND_CHANNEL\"<\/span>)<\/span>\n    public<\/span> MessageChannel<\/span> mqttOutboundChannel<\/span>() {<\/span>\n        return<\/span> new<\/span> DirectChannel<\/span>();<\/span>\n    }<\/span>\n<\/span>\n    \/**<\/span><\/span>\n     * MQTT\u6d88\u606f\u5904\u7406\u5668\uff08\u751f\u4ea7\u8005\uff09<\/span><\/span>\n     **\/<\/span><\/span>\n    @Bean<\/span><\/span>\n    @ServiceActivator<\/span>(inputChannel<\/span> =<\/span> \"MQTT_OUT_BOUND_CHANNEL\"<\/span>)<\/span>\n    public<\/span> MqttPahoMessageHandler<\/span> mqttOutbound<\/span>() {<\/span>\n        MqttPahoMessageHandler<\/span> messageHandler<\/span> =<\/span> new<\/span> MqttPahoMessageHandler<\/span>(mqttProperties<\/span>.getClientId<\/span>() +<\/span> \"_producer\"<\/span>, mqttClientFactory<\/span>());<\/span>\n        messageHandler<\/span>.setAsync<\/span>(true<\/span>);<\/span>\n        messageHandler<\/span>.setDefaultTopic<\/span>(mqttProperties<\/span>.getDefaultTopic<\/span>());<\/span>\n        messageHandler<\/span>.setAsyncEvents<\/span>(true<\/span>); \/\/ \u6d88\u606f\u53d1\u9001\u548c\u4f20\u8f93\u5b8c\u6210\u4f1a\u6709\u5f02\u6b65\u7684\u901a\u77e5\u56de\u8c03<\/span><\/span>\n        \/\/\u8bbe\u7f6e\u8f6c\u6362\u5668 \u53d1\u9001bytes\u6570\u636e<\/span><\/span>\n        DefaultPahoMessageConverter<\/span> converter<\/span> =<\/span> new<\/span> DefaultPahoMessageConverter<\/span>();<\/span>\n        converter<\/span>.setPayloadAsBytes<\/span>(true<\/span>);<\/span>\n        return<\/span> messageHandler<\/span>;<\/span>\n    }<\/span>\n<\/span>\n    \/**<\/span><\/span>\n     * \u914d\u7f6eclient,\u76d1\u542c\u7684topic<\/span><\/span>\n     * MQTT\u6d88\u606f\u8ba2\u9605\u7ed1\u5b9a\uff08\u6d88\u8d39\u8005\uff09<\/span><\/span>\n     **\/<\/span><\/span>\n    @Bean<\/span>(\"inbound\"<\/span>)<\/span>\n    public<\/span> MqttPahoMessageDrivenChannelAdapter<\/span> inbound<\/span>() {<\/span>\n        List<\/span><<\/span>String<\/span>><\/span> topicList<\/span> =<\/span> Arrays<\/span>.asList<\/span>(mqttProperties<\/span>.getDefaultTopic<\/span>().trim<\/span>().split<\/span>(\",\"<\/span>));<\/span>\n        String<\/span>[] topics<\/span> =<\/span> new<\/span> String<\/span>[topicList<\/span>.size<\/span>()];<\/span>\n        topicList<\/span>.toArray<\/span>(topics<\/span>);<\/span>\n        MqttPahoMessageDrivenChannelAdapter<\/span> adapter<\/span> =<\/span><\/span>\n                new<\/span> MqttPahoMessageDrivenChannelAdapter<\/span>(mqttProperties<\/span>.getClientId<\/span>() +<\/span> \"_consumer\"<\/span>, mqttClientFactory<\/span>(), topics<\/span>);<\/span>\n        adapter<\/span>.setCompletionTimeout<\/span>(mqttProperties<\/span>.getTimeout<\/span>());<\/span>\n        DefaultPahoMessageConverter<\/span> converter<\/span> =<\/span> new<\/span> DefaultPahoMessageConverter<\/span>();<\/span>\n        converter<\/span>.setPayloadAsBytes<\/span>(true<\/span>);<\/span>\n        adapter<\/span>.setConverter<\/span>(converter<\/span>);<\/span>\n        adapter<\/span>.setQos<\/span>(2<\/span>);<\/span>\n\/\/        adapter.addTopic(\"TOPIC1\");<\/span><\/span>\n        \/\/\u8bbe\u7f6e\u8ba2\u9605\u901a\u9053<\/span><\/span>\n        adapter<\/span>.setOutputChannel<\/span>(mqttInputChannel<\/span>());<\/span>\n        return<\/span> adapter<\/span>;<\/span>\n    }<\/span>\n<\/span>\n\/\/    \/**<\/span><\/span>\n\/\/     * \u83b7\u53d6MQTT\u5ba2\u6237\u7aef<\/span><\/span>\n\/\/     * \u81ea\u5b9a\u4e49\u8ba2\u9605\u6d88\u606f<\/span><\/span>\n\/\/     *\/<\/span><\/span>\n\/\/    @Value(\"mqttClient\")<\/span><\/span>\n\/\/    public IMqttClient getMqttClient() throws MqttException {<\/span><\/span>\n\/\/        IMqttClient mqttClient = mqttClientFactory().getClientInstance(mqttProperties.getHostUrl(), mqttProperties.getClientId());<\/span><\/span>\n\/\/        mqttClient.connect(getMqttConnectOptions());<\/span><\/span>\n\/\/        return mqttClient;<\/span><\/span>\n\/\/    }<\/span><\/span>\n<\/span>\n    \/**<\/span><\/span>\n     * MQTT\u4fe1\u606f\u901a\u9053\uff08\u6d88\u8d39\u8005\uff09<\/span><\/span>\n     **\/<\/span><\/span>\n    @Bean<\/span>(\"MQTT_INPUT_CHANNEL\"<\/span>)<\/span>\n    public<\/span> MessageChannel<\/span> mqttInputChannel<\/span>() {<\/span>\n        return<\/span> new<\/span> DirectChannel<\/span>();<\/span>\n    }<\/span>\n<\/span>\n    \/**<\/span><\/span>\n     * MQTT\u6d88\u606f\u5904\u7406\u5668\uff08\u6d88\u8d39\u8005\uff09<\/span><\/span>\n     **\/<\/span><\/span>\n    @Bean<\/span><\/span>\n    @ServiceActivator<\/span>(inputChannel<\/span> =<\/span> \"MQTT_INPUT_CHANNEL\"<\/span>)<\/span>\n    public<\/span> MessageHandler<\/span> handler<\/span>() {<\/span>\n        return<\/span> new<\/span> MessageHandler<\/span>() {<\/span>\n            @Override<\/span><\/span>\n            public<\/span> void<\/span> handleMessage<\/span>(Message<\/span><?><\/span> message<\/span>) throws<\/span> MessagingException<\/span> {<\/span>\n                \/\/\u5904\u7406\u63a5\u6536\u6d88\u606f<\/span><\/span>\n                mqttReceiveHandle<\/span>.handleMessage<\/span>(message<\/span>);<\/span>\n            }<\/span>\n        };<\/span>\n    }<\/span>\n}<\/span><\/pre>\n
\n
7.MqttPublishGateway<\/span><\/h3>\n
package<\/span> com<\/span>.rfos<\/span>.assistsilkworm<\/span>.mqtt<\/span>.gateway<\/span>;<\/span>\n<\/span>\nimport<\/span> org<\/span>.springframework<\/span>.integration<\/span>.annotation<\/span>.MessageEndpoint<\/span>;<\/span>\nimport<\/span> org<\/span>.springframework<\/span>.integration<\/span>.annotation<\/span>.MessagingGateway<\/span>;<\/span>\nimport<\/span> org<\/span>.springframework<\/span>.integration<\/span>.mqtt<\/span>.support<\/span>.MqttHeaders<\/span>;<\/span>\nimport<\/span> org<\/span>.springframework<\/span>.messaging<\/span>.handler<\/span>.annotation<\/span>.Header<\/span>;<\/span>\nimport<\/span> org<\/span>.springframework<\/span>.stereotype<\/span>.Service<\/span>;<\/span>\n<\/span>\n\/**<\/span><\/span>\n * mqtt\u53d1\u9001\u6d88\u606f<\/span><\/span>\n * \uff08defaultRequestChannel = \"mqttOutboundChannel\" \u5bf9\u5e94config\u914d\u7f6e\uff09<\/span><\/span>\n * **\/<\/span><\/span>\n@Service<\/span>(\"mqttGateway\"<\/span>)<\/span>\n@MessagingGateway<\/span>(defaultRequestChannel<\/span> =<\/span> \"MQTT_OUT_BOUND_CHANNEL\"<\/span>)<\/span>\npublic<\/span> interface<\/span> MqttPublishGateway<\/span> {<\/span>\n<\/span>\n    \/**<\/span><\/span>\n     * \u53d1\u9001\u4fe1\u606f\u5230MQTT\u670d\u52a1\u5668<\/span><\/span>\n     *<\/span><\/span>\n     * @param<\/span><\/span>\n     *\/<\/span><\/span>\n    void<\/span> sendToMqttObject<\/span>(@Header<\/span>(MqttHeaders<\/span>.TOPIC<\/span>) String<\/span> topic<\/span>, byte<\/span>[] payload<\/span>);<\/span>\n<\/span>\n    \/**<\/span><\/span>\n     * \u53d1\u9001\u4fe1\u606f\u5230MQTT\u670d\u52a1\u5668<\/span><\/span>\n     *<\/span><\/span>\n     * @param topic \u4e3b\u9898<\/span><\/span>\n     * @param payload \u6d88\u606f\u4e3b\u4f53<\/span><\/span>\n     *\/<\/span><\/span>\n    void<\/span> sendToMqtt<\/span>(@Header<\/span>(MqttHeaders<\/span>.TOPIC<\/span>) String<\/span> topic<\/span>, String<\/span> payload<\/span>);<\/span>\n<\/span>\n    \/**<\/span><\/span>\n     * \u53d1\u9001\u4fe1\u606f\u5230MQTT\u670d\u52a1\u5668<\/span><\/span>\n     *<\/span><\/span>\n     * @param topic \u4e3b\u9898<\/span><\/span>\n     * @param qos \u5bf9\u6d88\u606f\u5904\u7406\u7684\u51e0\u79cd\u673a\u5236\u3002<\/span><\/span>\n     * 0 \u8868\u793a\u7684\u662f\u8ba2\u9605\u8005\u6ca1\u6536\u5230\u6d88\u606f\u4e0d\u4f1a\u518d\u6b21\u53d1\u9001\uff0c\u6d88\u606f\u4f1a\u4e22\u5931\u3002<\/span><\/span>\n     * 1 \u8868\u793a\u7684\u662f\u4f1a\u5c1d\u8bd5\u91cd\u8bd5\uff0c\u4e00\u76f4\u5230\u63a5\u6536\u5230\u6d88\u606f\uff0c\u4f46\u8fd9\u79cd\u60c5\u51b5\u53ef\u80fd\u5bfc\u81f4\u8ba2\u9605\u8005\u6536\u5230\u591a\u6b21\u91cd\u590d\u6d88\u606f\u3002<\/span><\/span>\n     * 2 \u591a\u4e86\u4e00\u6b21\u53bb\u91cd\u7684\u52a8\u4f5c\uff0c\u786e\u4fdd\u8ba2\u9605\u8005\u6536\u5230\u7684\u6d88\u606f\u6709\u4e00\u6b21\u3002<\/span><\/span>\n     * @param payload \u6d88\u606f\u4e3b\u4f53<\/span><\/span>\n     *\/<\/span><\/span>\n    void<\/span> sendToMqtt<\/span>(@Header<\/span>(MqttHeaders<\/span>.TOPIC<\/span>) String<\/span> topic<\/span>, @Header<\/span>(MqttHeaders<\/span>.QOS<\/span>) int<\/span> qos<\/span>, String<\/span> payload<\/span>);<\/span>\n<\/span>\n    \/**<\/span><\/span>\n     * \u53d1\u9001\u4fe1\u606f\u5230MQTT\u670d\u52a1\u5668<\/span><\/span>\n     *<\/span><\/span>\n     * @param topic \u4e3b\u9898<\/span><\/span>\n     * @param payload \u6d88\u606f\u4e3b\u4f53<\/span><\/span>\n     *\/<\/span><\/span>\n    void<\/span> sendToMqtt<\/span>(@Header<\/span>(MqttHeaders<\/span>.TOPIC<\/span>) String<\/span> topic<\/span>, Object<\/span> payload<\/span>);<\/span>\n<\/span>\n    \/**<\/span><\/span>\n     * \u53d1\u9001\u4fe1\u606f\u5230MQTT\u670d\u52a1\u5668<\/span><\/span>\n     *<\/span><\/span>\n     * @param topic \u4e3b\u9898<\/span><\/span>\n     * @param payload \u6d88\u606f\u4e3b\u4f53<\/span><\/span>\n     *\/<\/span><\/span>\n    void<\/span> sendToMqtt<\/span>(@Header<\/span>(MqttHeaders<\/span>.TOPIC<\/span>) String<\/span> topic<\/span>, byte<\/span>[] payload<\/span>);<\/span>\n}<\/span><\/pre>\n
\n
8.MqttListener<\/span><\/h3>\n
package<\/span> com<\/span>.rfos<\/span>.assistsilkworm<\/span>.mqtt<\/span>.listenter<\/span>;<\/span>\n<\/span>\nimport<\/span> lombok<\/span>.extern<\/span>.slf4j<\/span>.Slf4j<\/span>;<\/span>\nimport<\/span> org<\/span>.springframework<\/span>.context<\/span>.event<\/span>.EventListener<\/span>;<\/span>\nimport<\/span> org<\/span>.springframework<\/span>.integration<\/span>.mqtt<\/span>.event<\/span>.MqttConnectionFailedEvent<\/span>;<\/span>\nimport<\/span> org<\/span>.springframework<\/span>.integration<\/span>.mqtt<\/span>.event<\/span>.MqttMessageDeliveredEvent<\/span>;<\/span>\nimport<\/span> org<\/span>.springframework<\/span>.integration<\/span>.mqtt<\/span>.event<\/span>.MqttMessageSentEvent<\/span>;<\/span>\nimport<\/span> org<\/span>.springframework<\/span>.integration<\/span>.mqtt<\/span>.event<\/span>.MqttSubscribedEvent<\/span>;<\/span>\nimport<\/span> org<\/span>.springframework<\/span>.stereotype<\/span>.Component<\/span>;<\/span>\n<\/span>\n@Slf4j<\/span><\/span>\n@Component<\/span><\/span>\npublic<\/span> class<\/span> MqttListener<\/span> {<\/span>\n    \/**<\/span><\/span>\n     * \u8fde\u63a5\u5931\u8d25\u7684\u4e8b\u4ef6\u901a\u77e5<\/span><\/span>\n     * @param mqttConnectionFailedEvent<\/span><\/span>\n     *\/<\/span><\/span>\n    @EventListener<\/span>(classes<\/span> =<\/span> MqttConnectionFailedEvent<\/span>.class<\/span>)<\/span>\n    public<\/span> void<\/span> listenerAction<\/span>(MqttConnectionFailedEvent<\/span> mqttConnectionFailedEvent<\/span>) {<\/span>\n        log<\/span>.info<\/span>(\"\u8fde\u63a5\u5931\u8d25\u7684\u4e8b\u4ef6\u901a\u77e5\"<\/span>);<\/span>\n    }<\/span>\n<\/span>\n    \/**<\/span><\/span>\n     * \u5df2\u53d1\u9001\u7684\u4e8b\u4ef6\u901a\u77e5<\/span><\/span>\n     * @param mqttMessageSentEvent<\/span><\/span>\n     *\/<\/span><\/span>\n    @EventListener<\/span>(classes<\/span> =<\/span> MqttMessageSentEvent<\/span>.class<\/span>)<\/span>\n    public<\/span> void<\/span> listenerAction<\/span>(MqttMessageSentEvent<\/span> mqttMessageSentEvent<\/span>) {<\/span>\n        log<\/span>.info<\/span>(\"\u5df2\u53d1\u9001\u7684\u4e8b\u4ef6\u901a\u77e5[{}]\"<\/span>,mqttMessageSentEvent<\/span>.toString<\/span>());<\/span>\n    }<\/span>\n<\/span>\n    \/**<\/span><\/span>\n     * \u5df2\u4f20\u8f93\u5b8c\u6210\u7684\u4e8b\u4ef6\u901a\u77e5<\/span><\/span>\n     * 1.QOS == 0,\u53d1\u9001\u6d88\u606f\u540e\u4f1a\u5373\u53ef\u8fdb\u884c\u6b64\u4e8b\u4ef6\u56de\u8c03\uff0c\u56e0\u4e3a\u4e0d\u9700\u8981\u7b49\u5f85\u56de\u6267<\/span><\/span>\n     * 2.QOS == 1\uff0c\u53d1\u9001\u6d88\u606f\u540e\u4f1a\u7b49\u5f85ACK\u56de\u6267\uff0cACK\u56de\u6267\u540e\u4f1a\u8fdb\u884c\u6b64\u4e8b\u4ef6\u901a\u77e5<\/span><\/span>\n     * 3.QOS == 2\uff0c\u53d1\u9001\u6d88\u606f\u540e\u4f1a\u7b49\u5f85PubRECV\u56de\u6267\uff0c\u77e5\u9053\u6536\u5230PubCOMP\u540e\u4f1a\u8fdb\u884c\u6b64\u4e8b\u4ef6\u901a\u77e5<\/span><\/span>\n     * @param mqttMessageDeliveredEvent<\/span><\/span>\n     *\/<\/span><\/span>\n    @EventListener<\/span>(classes<\/span> =<\/span> MqttMessageDeliveredEvent<\/span>.class<\/span>)<\/span>\n    public<\/span> void<\/span> listenerAction<\/span>(MqttMessageDeliveredEvent<\/span> mqttMessageDeliveredEvent<\/span>) {<\/span>\n        log<\/span>.info<\/span>(\"\u5df2\u4f20\u8f93\u5b8c\u6210\u7684\u4e8b\u4ef6\u901a\u77e5\"<\/span>);<\/span>\n    }<\/span>\n<\/span>\n    \/**<\/span><\/span>\n     * \u6d88\u606f\u8ba2\u9605\u7684\u4e8b\u4ef6\u901a\u77e5<\/span><\/span>\n     * @param mqttSubscribedEvent<\/span><\/span>\n     *\/<\/span><\/span>\n    @EventListener<\/span>(classes<\/span> =<\/span> MqttSubscribedEvent<\/span>.class<\/span>)<\/span>\n    public<\/span> void<\/span> listenerAction<\/span>(MqttSubscribedEvent<\/span> mqttSubscribedEvent<\/span>) {<\/span>\n        log<\/span>.info<\/span>(\"\u6d88\u606f\u8ba2\u9605\u7684\u4e8b\u4ef6\u901a\u77e5\"<\/span>);<\/span>\n        log<\/span>.info<\/span>(\"\u6210\u529f\u8ba2\u9605\u5230\u4e3b\u9898:  info={}\"<\/span>, mqttSubscribedEvent<\/span>.toString<\/span>());<\/span>\n    }<\/span>\n}<\/span><\/pre>\n
\n
9.MqttClientController<\/span><\/h3>\n
package com.rfos.assistsilkworm.mqtt.client;\n\nimport com.rfos.assistsilkworm.common.CommonResult;\nimport com.rfos.assistsilkworm.mqtt.config.MqttConfig;\nimport com.rfos.assistsilkworm.mqtt.gateway.MqttPublishGateway;\nimport org.eclipse.paho.client.mqttv3.IMqttClient;\nimport org.eclipse.paho.client.mqttv3.MqttClient;\nimport org.eclipse.paho.client.mqttv3.MqttException;\nimport org.springframework.beans.factory.annotation.Autowired;\nimport org.springframework.beans.factory.annotation.Qualifier;\nimport org.springframework.integration.annotation.ServiceActivator;\nimport org.springframework.integration.core.MessageProducer;\nimport org.springframework.integration.mqtt.core.MqttPahoClientFactory;\nimport org.springframework.integration.mqtt.inbound.MqttPahoMessageDrivenChannelAdapter;\nimport org.springframework.web.bind.annotation.GetMapping;\nimport org.springframework.web.bind.annotation.RequestMapping;\nimport org.springframework.web.bind.annotation.RequestParam;\nimport org.springframework.web.bind.annotation.RestController;\n\n\/**\n * @Author rfos\n * @Date 2023\/2\/20 19:01\n * @Description TODO MQTT\u5ba2\u6237\u7aef\u5904\u7406\u53d1\u9001\u6570\u636e\n *\/\n@RestController\n@RequestMapping(\"\/mqtt\")\npublic class MqttClientController {\n    \/**\n     * \u53d1\u9001\u7f51\u5173\n     *\/\n    @Autowired\n    private MqttPublishGateway mqttPublishGateway;\n\n    @Autowired\n    @Qualifier(\"inbound\")\n    private MqttPahoMessageDrivenChannelAdapter messageProducer;\n\n\n\n    \/**\n     * \u6d4b\u8bd5\uff1a\u53d1\u5e03\u6d88\u606f\n     *\/\n    @GetMapping(\"\/testPublish\")\n    public CommonResult sentTest(@RequestParam(\"topic\") String topic,\n                                 @RequestParam(\"payload\") String payload){\n        mqttPublishGateway.sendToMqtt(topic,1,payload);\n        return CommonResult.success();\n    }\n\n    \/**\n     * \u6d4b\u8bd5\u8ba2\u9605\u4fe1\u606f\n     *\/\n    @GetMapping(\"\/testSubscribe\")\n    @ServiceActivator(inputChannel = \"MQTT_INPUT_CHANNEL\")\n    public CommonResult testSubscribe() throws MqttException {\n        messageProducer.addTopic(\"TOPICS\");\n        return CommonResult.success();\n    }\n\n\n\n\n}\n<\/pre>\n
\n
<\/span><\/p>\n
\n <\/mp-style-type><\/p>\n
\u672c\u7bc7\u6587\u7ae0\u6765\u6e90\u4e8e\u5fae\u4fe1\u516c\u4f17\u53f7\uff1a\u4f5c\u8005\uff1a\u539f\u521b rfos \u5c0f\u7f16\u9a91\u9c7c\u4e4b\u5b66\u70b9\u77e5\u8bc6 \u5c0f\u7f16\u9a91\u9c7c\u4e4b\u5b66\u70b9\u77e5\u8bc6 \u5fae\u4fe1\u53f7 rfospublic \u529f\u80fd\u4ecb\u7ecd \u77e5\u8bc6\u662f\u751f\u6d3b\u7684\u660e\u706f,\u66f4\u662f\u4e00\u79cd\u5feb\u4e50\u3002\u597d\u5947\u5219\u662f\u77e5\u8bc6\u7684\u840c\u82bd\uff0c\u800c\u8fd9\u91cc\u6b63\u662f\u4f60\u4eec\u7684\u597d\u5947\u4e4b\u5904\uff0c\u4e5f\u662f\u4f60\u4eec\u7684\u83b7\u89e3\u4e4b\u5730\u3002\u9676\u884c\u77e5\u8bf4\u8fc7:\u201c\u667a\u6167\u662f\u751f\u6210\u7684\uff0c\u77e5\u8bc6\u662f\u5b66\u6765\u7684\u3002\u5f88\u5e78\u8fd0\uff0c\u80fd\u628a\u6240\u5b66\uff0c\u6240\u609f\uff0c\u6240\u77e5\uff0c\u6240\u5f97\u7684\u4e1c\u897f\u5206\u4eab\u7ed9\u5927\u5bb6\u3002\u5e0c\u671b\u5927\u5bb6\u5173\u6ce8\uff0c\u8c22\u8c22 ! \u53d1\u8868\u4e8e
\n\u8f6c\u8f7d\u5730\u5740\uff1ahttps:\/\/mp.weixin.qq.com\/s\/Fwhle56yhHES9x3bcHBUuQ<\/p>\n","raw":"","protected":false},"excerpt":{"rendered":"
1.pom.xml <!–       \u8fde\u63a5MQTT–>     …<\/p>\n","protected":false},"author":1,"featured_media":0,"comment_status":"open","ping_status":"open","sticky":false,"template":"","format":"standard","meta":{"emotion":"","emotion_color":"","title_style":"","license":"","footnotes":""},"categories":[7],"tags":[],"class_list":["post-2983","post","type-post","status-publish","format-standard","hentry","category-spring-boot"],"post_thumbnail_image":"https:\/\/wp-moto-1258805347.cos.ap-shanghai.myqcloud.com\/default\/nine\/2023-07-14\/17-56-09\/20230714175609323229873.png","content_first_image":"https:\/\/wp-moto-1258805347.cos.ap-shanghai.myqcloud.com\/default\/nine\/2023-07-14\/17-56-09\/20230714175609323229873.png","post_medium_image_300":"https:\/\/wp-moto-1258805347.cos.ap-shanghai.myqcloud.com\/default\/nine\/2023-07-14\/17-56-09\/20230714175609323229873.png","post_thumbnail_image_624":"https:\/\/wp-moto-1258805347.cos.ap-shanghai.myqcloud.com\/default\/nine\/2023-07-14\/17-56-09\/20230714175609323229873.png","post_frist_image":"https:\/\/wp-moto-1258805347.cos.ap-shanghai.myqcloud.com\/default\/nine\/2023-07-14\/17-56-09\/20230714175609323229873.png","post_medium_image":"https:\/\/wp-moto-1258805347.cos.ap-shanghai.myqcloud.com\/default\/nine\/2023-07-14\/17-56-09\/20230714175609323229873.png","post_large_image":"https:\/\/wp-moto-1258805347.cos.ap-shanghai.myqcloud.com\/default\/nine\/2023-07-14\/17-56-09\/20230714175609323229873.png","post_full_image":"https:\/\/wp-moto-1258805347.cos.ap-shanghai.myqcloud.com\/default\/nine\/2023-07-14\/17-56-09\/20230714175609323229873.png","post_all_images":[{"imagesurl":"https:\/\/wp-moto-1258805347.cos.ap-shanghai.myqcloud.com\/default\/nine\/2023-07-14\/17-56-09\/20230714175609323229873.png","id":"image0"},{"imagesurl":"https:\/\/wp-moto-1258805347.cos.ap-shanghai.myqcloud.com\/default\/nine\/2023-07-14\/17-56-09\/20230714175609323229873.png","id":"image1"},{"imagesurl":"https:\/\/wp-moto-1258805347.cos.ap-shanghai.myqcloud.com\/default\/nine\/2023-07-14\/17-56-09\/20230714175609e6e6f721d.png","id":"image2"},{"imagesurl":"https:\/\/wp-moto-1258805347.cos.ap-shanghai.myqcloud.com\/default\/nine\/2023-07-14\/17-56-09\/20230714175609e6e6f721d.png","id":"image3"}],"videoAdId":"","listAd":"0","listAdId":"","listAdEvery":6,"total_comments":0,"category_name":"Spring Boot","post_date":"2023-07-14","like_count":"0","praiseWord":"\u9f13\u52b1","copyright_state":"","excitationAd":"0","rewardedVideoAdId":"","detailAdId":"","detailAd":"0","enterpriseMinapp":"0","audios":[],"postImageUrl":"https:\/\/wp-moto-1258805347.cos.ap-shanghai.myqcloud.com\/2023\/05\/20230519082947553.jpg","avatarurls":[],"related_posts":null,"pageviews":525,"next_post_id":2985,"next_post_title":"\u9762\u8bd5\u5b98\u95ee\uff1aSpringBoot\u4e2d\u6761\u4ef6\u6ce8\u89e3\u5e95\u5c42\u5982\u4f55\u5b9e\u73b0\u7684\uff1f","previous_post_id":2971,"previous_post_title":"6 \u79cd\u65b9\u5f0f\u8bfb\u53d6 Springboot \u7684\u914d\u7f6e\uff08\u539f\u7406+\u5b9e\u6218\uff09","_links":{"self":[{"href":"https:\/\/xinchewhd.com.cn\/index.php\/wp-json\/wp\/v2\/posts\/2983","targetHints":{"allow":["GET"]}}],"collection":[{"href":"https:\/\/xinchewhd.com.cn\/index.php\/wp-json\/wp\/v2\/posts"}],"about":[{"href":"https:\/\/xinchewhd.com.cn\/index.php\/wp-json\/wp\/v2\/types\/post"}],"author":[{"embeddable":true,"href":"https:\/\/xinchewhd.com.cn\/index.php\/wp-json\/wp\/v2\/users\/1"}],"replies":[{"embeddable":true,"href":"https:\/\/xinchewhd.com.cn\/index.php\/wp-json\/wp\/v2\/comments?post=2983"}],"version-history":[{"count":0,"href":"https:\/\/xinchewhd.com.cn\/index.php\/wp-json\/wp\/v2\/posts\/2983\/revisions"}],"wp:attachment":[{"href":"https:\/\/xinchewhd.com.cn\/index.php\/wp-json\/wp\/v2\/media?parent=2983"}],"wp:term":[{"taxonomy":"category","embeddable":true,"href":"https:\/\/xinchewhd.com.cn\/index.php\/wp-json\/wp\/v2\/categories?post=2983"},{"taxonomy":"post_tag","embeddable":true,"href":"https:\/\/xinchewhd.com.cn\/index.php\/wp-json\/wp\/v2\/tags?post=2983"}],"curies":[{"name":"wp","href":"https:\/\/api.w.org\/{rel}","templated":true}]}}