Azure Java SDK v12 未异步下载文件

Question

我正在编写一个使用 Java 12 Azure 存储 SDK 从 Azure Blob 存储下载图像的快速概念验证。当我将其转换为同步时，以下代码可以正常工作。但是，尽管代码底部有 subscribe()，但我只看到订阅消息。成功和错误处理程序没有触发。如有任何建议或想法，我将不胜感激。

感谢您的时间和帮助。

private fun azureReactorDownload() {

    var startTime = 0L
    var accountName = "abcd"
    var key = "09sd0908sd08f0s&&6^%"
    var endpoint = "https://${accountName}.blob.core.windows.net/$accountName
    var containerName = "mycontainer"
    var blobName = "animage.jpg"

    // Get the Blob Service client, so we can use it to access blobs, containers, etc.
    BlobServiceClientBuilder()
        // Container URL
        .endpoint(endpoint)
        .credential(
            SharedKeyCredential(
                accountName,
                key
            )
        )
        .buildAsyncClient()

        // Get the container client so we can work with our container and its blobs.
        .getContainerAsyncClient(containerName)

        // Get the block blob client so we can access individual blobs and include the path
        // within the container as part of the filename.
        .getBlockBlobAsyncClient(blobName)

        // Initiate the download of the desired blob.
        .download()
        .map { response ->
            // Drill down to the ByteBuffer.
            response.value()
        }
        .doOnSubscribe {
            println(">>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>> Subscription arrived.")
            startTime = System.currentTimeMillis()
        }
        .doOnSuccess { data ->
            data.map { byteBuffer ->
                println(">>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>> READY TO WRITE TO THE FILE")
                byteBuffer.writeToFile("/tmp/azrxblobdownload.jpg")
                val elapsedTime = System.currentTimeMillis() - startTime
                println(">>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>> Finished downloading blob in $elapsedTime ms.")
            }
        }
        .doOnError {
            println(">>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>> Failed to download blob: ${it.localizedMessage}")
        }

        .subscribe()
}

fun ByteBuffer.writeToFile(path: String) {
    val fc = FileOutputStream(path).channel
    fc.write(this)
    fc.close()
}

Answer 1

我看到有人在 4 个月前问了同​​样的问题，但没有得到答案：

Azure Blob Storage Java SDK: Why isn't asynchronous working?

我将推测 JDK 的这一部分现在无法正常工作。我不建议使用 Azure 的 Java 版本。

您应该能够以另一种方式完成它，也许是以下答案之一： Downloading Multiple Files Parallelly or Asynchronously in Java

Answer 2

我曾与 Microsoft 合作，并在以下链接中有一个记录在案的解决方案：https://github.com/Azure/azure-sdk-for-java/issues/5071。和我一起工作的人提供了非常好的背景信息，所以它不仅仅是一些工作代码。

我在 Azure Java SDK v12 中针对 downloadToFile() 方法向 Microsoft 打开了一个类似的查询，该方法在保存到文件时会引发异常。

这是该帖子的工作代码：

private fun azureReactorDownloadMS() {

    var startTime = 0L

    val chunkCounter = AtomicInteger(0)

    // Get the Blob Service client, so we can use it to access blobs, containers, etc.
    val aa = BlobServiceClientBuilder()
        // Container URL
        .endpoint(kEndpoint)
        .credential(
            SharedKeyCredential(
                kAccountName,
                kAccountKey
            )
        )
        .buildAsyncClient()

        // Get the container client so we can work with our container and its blobs.
        .getContainerAsyncClient(kContainerName)

        // Get the block blob client so we can access individual blobs and include the path
        // within the container as part of the filename.
        .getBlockBlobAsyncClient(kBlobName)
        .download()
        // Response<Flux<ByteBuffer>> to Flux<ByteBuffer>
        .flatMapMany { response ->
            response.value()
        }
        .doOnSubscribe {
            println(">>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>> Subscription arrived.")
            startTime = System.currentTimeMillis()
        }
        .doOnNext { byteBuffer ->
            println(">>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>> CHUNK ${chunkCounter.incrementAndGet()} FROM BLOB ARRIVED...")
        }
        .doOnComplete {
            val elapsedTime = System.currentTimeMillis() - startTime
            println(">>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>> Finished downloading ${chunkCounter.incrementAndGet()} chunks of data for the blob in $elapsedTime ms.")
        }
        .doOnError {
            println(">>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>> Failed to download blob: ${it.localizedMessage}")
        }

        .blockLast()
}