String subTopic = "cest/cst";
    String pubTopic = "cest/cst";
    String content = "Test Java";
    int qos = 2;
    String broker = "ws://emqx@10.168.18.239:8083";
    String clientId = "mqttx_c271be04";
    MemoryPersistence persistence = new MemoryPersistence();

    try {
        MqttClient client = new MqttClient(broker, clientId, persistence);

        // MQTT 连接选项
        MqttConnectOptions connOpts = new MqttConnectOptions();
        connOpts.setUserName("test");
        connOpts.setPassword("123456".toCharArray());
        // 保留会话
        connOpts.setCleanSession(true);
        // 设置回调
        client.setCallback(new OnMessageCallback());

        // 建立连接
        System.out.println("Connecting to broker: " + broker);
        client.connect(connOpts);

        System.out.println("Connected");
        System.out.println("Publishing message: " + content);

        // 订阅
        client.subscribe(subTopic,2);

        // 消息发布所需参数
        MqttMessage message = new MqttMessage(content.getBytes());
        message.setQos(qos);
        client.publish(pubTopic, message);
        System.out.println("Message published");

        client.disconnect();
        System.out.println("Disconnected");
        client.close();
        System.exit(0);
    } catch (MqttException me) {
        System.out.println("reason " + me.getReasonCode());
        System.out.println("msg " + me.getMessage());
        System.out.println("loc " + me.getLocalizedMessage());
        System.out.println("cause " + me.getCause());
        System.out.println("excep " + me);
        me.printStackTrace();
    }
}