springboot + eclipse paho 实现mqtt client 消费者

• 依赖
• 配置
• 回调类
• 创建消费者
• 启动

		
            org.springframework.boot
            spring-boot-starter-integration
        

        
            org.springframework.integration
            spring-integration-stream
        

        
            org.springframework.integration
            spring-integration-mqtt
        

yml文件

  mqtt:
    username: 用户名
    password: 密码
    host: mqtt服务器地址
    client-id: 唯一的client-id
    topic: 要订阅的主题
    connection-timeout: 30
    keep-alive-interval: 10
    qos: 0

property类

@Component
@ConfigurationProperties(prefix = "mqtt")
@Data
public class MqttProperty {

    private String username;

    private String password;

    private String host;

    private String clientId;

    private String topic;

    private Integer connectionTimeout;

    private Integer keepAliveInterval;

    private Integer qos;

}

@Configuration
@Slf4j
public class MyCallBack implements MqttCallback {
    @Override
    public void connectionLost(Throwable throwable) {
        log.error("mqtt链接丢失");
    }

    @Override
    public void messageArrived(String s, MqttMessage mqttMessage) throws Exception {
        log.info("topic:" + s);
        String data = new String(mqttMessage.getPayload());
        log.info("message:" + data);
        
    }

    @Override
    public void deliveryComplete(IMqttDeliveryToken iMqttDeliveryToken) {
        log.debug("deliveryComplete:" + iMqttDeliveryToken.isComplete());
    }
}

@Configuration
@Slf4j
public class ClientBuilder {
    @Resource
    MqttProperty mqttProperty;

    @Resource
    MyCallBack myCallBack;

    @Bean
    public MqttClient init() throws MqttException {
        MqttClient ret = new MqttClient(mqttProperty.getHost(), mqttProperty.getClientId());
        MqttConnectOptions options = new MqttConnectOptions();
        options.setUserName(mqttProperty.getUsername());
        options.setPassword(mqttProperty.getPassword().toCharArray());
        options.setCleanSession(false);
        options.setConnectionTimeout(mqttProperty.getConnectionTimeout());
        options.setKeepAliveInterval(mqttProperty.getKeepAliveInterval());
        options.setAutomaticReconnect(true);
        ret.setCallback(myCallBack);
        ret.connect(options);
        if (!ret.isConnected()){
            log.info("mqtt链接"+ mqttProperty.getHost() +"失败");
        }
        else {
            log.info("mqtt链接"+ mqttProperty.getHost() +"成功");
            ret.subscribe(mqttProperty.getTopic(), mqttProperty.getQos());
        }
        return ret;
    }
}

启动项目后会自动订订阅配置的主题