使用kotlin+springboot集成mqtt

在物联网平台搭建方面，最基础的通讯设施 mqtt 是必不可少的，之前用 java 集成，现在换成 kotiln

一、引入 maven 依赖

<dependency>
            <groupId>org.springframework.integration</groupId>
            <artifactId>spring-integration-mqtt</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-integration</artifactId>
        </dependency>

二、配置 mqtt

• 需要在 springboot 的 application.yml 配置如下

spring:
  mqtt:
    url: tcp://localhost:1883 # mqtt的host和端口
    client.id: ${spring.application.name}${server.port}
    subscibe.topic: lot-pi   # 要订阅的主题
    send.topic: lot-admin  # 要发送的主题
    username: mqtt   # 用户名(未生效)
    password: 123456 # 密码(未生效)

• 基于 bean 的 mqtt 配置

/**
 * mqtt配置
 *
 * @author liucheng
 */
@Configuration
@IntegrationComponentScan
class MqttConfiguration @Autowired
constructor(private var mqttHandler: MqttHandler) {

    @Value("\${spring.mqtt.username}")
    private val mqttUserName: String? = null
    @Value("\${spring.mqtt.password}")
    private val mqttPassword: String? = null
    @Value("\${spring.mqtt.url}")
    private val hostUrl: String? = null
    @Value("\${spring.mqtt.client.id}")
    private val clientId: String? = null
    @Value("\${spring.mqtt.send.topic}")
    private val sendTopic: String? = null
    @Value("\${spring.mqtt.subscibe.topic}")
    private val subscibeTopic: String? = null


    @Bean
    fun mqttClientFactory(): MqttPahoClientFactory {
        val options = MqttConnectOptions()
        options.serverURIs = arrayOf(hostUrl)
        options.userName = mqttUserName
        options.password = mqttPassword!!.toCharArray()
        val factory = DefaultMqttPahoClientFactory()
        factory.connectionOptions = options
        return factory
    }

    @Bean
    fun mqttInFlow(): IntegrationFlow {
        return IntegrationFlows.from(mqttInbound())
                .transform<Any, Any> { p: Any? -> p }
                .handle(mqttHandler.handler())
                .get()
    }

    @Bean
    fun mqttInbound(): MessageProducerSupport {
        val adapter = MqttPahoMessageDrivenChannelAdapter(clientId + time + "customer",
                mqttClientFactory(), subscibeTopic)
        adapter.setConverter(DefaultPahoMessageConverter())
        adapter.setCompletionTimeout(5000)
        adapter.setQos(1)
        return adapter
    }

    @Bean
    fun mqttOutFlow(): IntegrationFlow {
        return IntegrationFlows.from(mqttInputChannel())
                .handle(mqttOutbound())
                .get()
    }

    @Bean
    fun mqttInputChannel(): MessageChannel {
        return DirectChannel()
    }

    @Bean
    fun mqttOutbound(): MessageHandler {
        val messageHandler = MqttPahoMessageHandler(clientId + time + "publisher", mqttClientFactory())
        messageHandler.setAsync(true)
        messageHandler.setDefaultTopic(sendTopic)
        return messageHandler
    }

    companion object {
        var time = System.currentTimeMillis()
    }
}

三、创建一个用来引用到服务中发送消息的业务接口

MqttService

@MessagingGateway(defaultRequestChannel = "mqttInputChannel")
interface MqttService {
    fun sendToMqtt(data: String)
}

四、创建一个接收到消息的处理器

MqttHandler

@Component
class MqttHandler @Autowired
constructor(private val applicationEventPublisher: ApplicationEventPublisher) {
    private val logger = LoggerFactory.getLogger(javaClass)
    fun handler(): MessageHandler {
        return MessageHandler { message: Message<*> ->
            logger.info("***********MqttHandler**********:{}", message)
            applicationEventPublisher.publishEvent(MqttEvent(receiveRecord))
        }
    }
}

我这里将收到的消息打印出来并使用事件发送出去，如果不使用事件直接在这里处理即可。

五、发送消息

@Component
class MqttSendTest @Autowired
constructor(private val mqttService: MqttService) {
    fun test() {
        # 这里发送内容是字符串
        mqttService.sendToMqtt("ok")
    }
}