diff --git a/Sources/General/KingfisherManager.swift b/Sources/General/KingfisherManager.swift
index 1979a8f70..42eff0ebd 100644
--- a/Sources/General/KingfisherManager.swift
+++ b/Sources/General/KingfisherManager.swift
@@ -708,6 +708,113 @@ public class KingfisherManager {
     }
 }
 
+// Concurrency
+extension KingfisherManager {
+    /// Gets an image from a given resource.
+    /// - Parameters:
+    ///   - resource: The `Resource` object defines data information like key or URL.
+    ///   - options: Options to use when creating the image.
+    ///   - progressBlock: Called when the image downloading progress gets updated. If the response does not contain an
+    ///                    `expectedContentLength`, this block will not be called. `progressBlock` is always called in
+    ///                    main queue.
+    ///
+    /// - Note: This method will first check whether the requested `resource` is already in cache or not. If cached,
+    ///    it returns the value after the cached image retrieved. Otherwise, it
+    ///    will download the `resource`, store it in cache, then returns the value.
+    public func retrieveImage(
+        with resource: Resource,
+        options: KingfisherOptionsInfo? = nil,
+        progressBlock: DownloadProgressBlock? = nil
+    ) async throws -> RetrieveImageResult
+    {
+        try await retrieveImage(
+            with: resource.convertToSource(),
+            options: options,
+            progressBlock: progressBlock
+        )
+    }
+    
+    /// Gets an image from a given resource.
+    ///
+    /// - Parameters:
+    ///   - source: The `Source` object defines data information from network or a data provider.
+    ///   - options: Options to use when creating the image.
+    ///   - progressBlock: Called when the image downloading progress gets updated. If the response does not contain an
+    ///                    `expectedContentLength`, this block will not be called. `progressBlock` is always called in
+    ///                    main queue.
+    ///
+    /// - Note: This method will first check whether the requested `resource` is already in cache or not. If cached,
+    ///    it returns the value after the cached image retrieved. Otherwise, it
+    ///    will download the `resource`, store it in cache, then returns the value.
+    ///
+    public func retrieveImage(
+        with source: Source,
+        options: KingfisherOptionsInfo? = nil,
+        progressBlock: DownloadProgressBlock? = nil
+    ) async throws -> RetrieveImageResult
+    {
+        let options = currentDefaultOptions + (options ?? .empty)
+        let info = KingfisherParsedOptionsInfo(options)
+        return try await retrieveImage(
+            with: source,
+            options: info,
+            progressBlock: progressBlock
+        )
+    }
+    
+    func retrieveImage(
+        with source: Source,
+        options: KingfisherParsedOptionsInfo,
+        progressBlock: DownloadProgressBlock? = nil
+    ) async throws -> RetrieveImageResult
+    {
+        var info = options
+        if let block = progressBlock {
+            info.onDataReceived = (info.onDataReceived ?? []) + [ImageLoadingProgressSideEffect(block)]
+        }
+        return try await retrieveImage(
+            with: source,
+            options: info,
+            progressiveImageSetter: nil
+        )
+    }
+    
+    func retrieveImage(
+        with source: Source,
+        options: KingfisherParsedOptionsInfo,
+        progressiveImageSetter: ((KFCrossPlatformImage?) -> Void)? = nil,
+        referenceTaskIdentifierChecker: (() -> Bool)? = nil
+    ) async throws -> RetrieveImageResult
+    {
+        let task = CancellationDownloadTask()
+        return try await withTaskCancellationHandler {
+            try await withCheckedThrowingContinuation { continuation in
+                let downloadTask = retrieveImage(
+                    with: source,
+                    options: options,
+                    downloadTaskUpdated: { newTask in
+                        Task {
+                            await task.setTask(newTask)
+                        }
+                    },
+                    progressiveImageSetter: progressiveImageSetter,
+                    referenceTaskIdentifierChecker: referenceTaskIdentifierChecker,
+                    completionHandler: { result in
+                        continuation.resume(with: result)
+                    }
+                )
+                Task {
+                    await task.setTask(downloadTask)
+                }
+            }
+        } onCancel: {
+            Task {
+                await task.task?.cancel()
+            }
+        }
+    }
+}
+
 class RetrievingContext {
 
     var options: KingfisherParsedOptionsInfo
diff --git a/Sources/Networking/ImageDownloader.swift b/Sources/Networking/ImageDownloader.swift
index 7078d885d..68eac53e6 100644
--- a/Sources/Networking/ImageDownloader.swift
+++ b/Sources/Networking/ImageDownloader.swift
@@ -87,6 +87,13 @@ public struct DownloadTask {
     }
 }
 
+actor CancellationDownloadTask {
+    var task: DownloadTask?
+    func setTask(_ task: DownloadTask?) {
+        self.task = task
+    }
+}
+
 extension DownloadTask {
     enum WrappedTask {
         case download(DownloadTask)
diff --git a/Tests/KingfisherTests/KingfisherManagerTests.swift b/Tests/KingfisherTests/KingfisherManagerTests.swift
index 17803d7cb..30b365d07 100644
--- a/Tests/KingfisherTests/KingfisherManagerTests.swift
+++ b/Tests/KingfisherTests/KingfisherManagerTests.swift
@@ -89,6 +89,32 @@ class KingfisherManagerTests: XCTestCase {
         waitForExpectations(timeout: 3, handler: nil)
     }
     
+    func testRetrieveImageAsync() async throws {
+        let url = testURLs[0]
+        stub(url, data: testImageData)
+
+        let manager = self.manager!
+        
+        var result = try await manager.retrieveImage(with: url)
+        XCTAssertNotNil(result.image)
+        XCTAssertEqual(result.cacheType, .none)
+        
+        result = try await manager.retrieveImage(with: url)
+        XCTAssertNotNil(result.image)
+        XCTAssertEqual(result.cacheType, .memory)
+        
+        manager.cache.clearMemoryCache()
+        result = try await manager.retrieveImage(with: url)
+        XCTAssertNotNil(result.image)
+        XCTAssertEqual(result.cacheType, .disk)
+        
+        manager.cache.clearMemoryCache()
+        await manager.cache.clearDiskCache()
+        result = try await manager.retrieveImage(with: url)
+        XCTAssertNotNil(result.image)
+        XCTAssertEqual(result.cacheType, .none)
+    }
+    
     func testRetrieveImageWithProcessor() {
         let exp = expectation(description: #function)
         let url = testURLs[0]
@@ -149,6 +175,53 @@ class KingfisherManagerTests: XCTestCase {
         waitForExpectations(timeout: 3, handler: nil)
     }
     
+    func testRetrieveImageCancel() {
+        let exp = expectation(description: #function)
+        let url = testURLs[0]
+        let stub = delayedStub(url, data: testImageData, length: 123)
+
+        let task = manager.retrieveImage(with: url) {
+            result in
+            XCTAssertNotNil(result.error)
+            XCTAssertTrue(result.error!.isTaskCancelled)
+            exp.fulfill()
+        }
+
+        XCTAssertNotNil(task)
+        task?.cancel()
+        _ = stub.go()
+        waitForExpectations(timeout: 3, handler: nil)
+    }
+    
+    actor CallingChecker {
+        var called = false
+        func mark() {
+            called = true
+        }
+    }
+    
+    func testRetrieveImageCancelAsync() async throws {
+        let url = testURLs[0]
+        let stub = delayedStub(url, data: testImageData, length: 123)
+
+        let checker = CallingChecker()
+        let task = Task {
+            do {
+                _ = try await manager.retrieveImage(with: url)
+                XCTFail()
+            } catch {
+                await checker.mark()
+                XCTAssertTrue((error as! KingfisherError).isTaskCancelled)
+            }
+        }
+        try await Task.sleep(nanoseconds: NSEC_PER_SEC / 10)
+        task.cancel()
+         _ = stub.go()
+        try await Task.sleep(nanoseconds: NSEC_PER_SEC / 10)
+        let catchCalled = await checker.called
+        XCTAssertTrue(catchCalled)
+    }
+    
     func testSuccessCompletionHandlerRunningOnMainQueueDefaultly() {
         let progressExpectation = expectation(description: "progressBlock running on main queue")
         let completionExpectation = expectation(description: "completionHandler running on main queue")