Implement async multi get

Author: cnblogs-dudu

Enyim.Caching/IMemcachedClient.cs

@@ -16,8 +16,9 @@ public interface IMemcachedClient : IDisposable
         object Get(string key);
 		T Get<T>(string key);
 		IDictionary<string, T> Get<T>(IEnumerable<string> keys);
+        Task<IDictionary<string, T>> GetAsync<T>(IEnumerable<string> keys);
 
-		bool TryGet(string key, out object value);
+        bool TryGet(string key, out object value);
 		bool TryGetWithCas(string key, out CasResult<object> value);
 
 		CasResult<object> GetWithCas(string key);

Enyim.Caching/MemcachedClient.cs

@@ -920,6 +920,11 @@ public IDictionary<string, T> Get<T>(IEnumerable<string> keys)
             return PerformMultiGet<T>(keys, (mget, kvp) => this.transcoder.Deserialize<T>(kvp.Value));
         }
 
+        public async Task<IDictionary<string, T>> GetAsync<T>(IEnumerable<string> keys)
+        {
+            return await PerformMultiGetAsync<T>(keys, (mget, kvp) => this.transcoder.Deserialize<T>(kvp.Value));
+        }
+
         public IDictionary<string, CasResult<object>> GetWithCas(IEnumerable<string> keys)
         {
             return PerformMultiGet<CasResult<object>>(keys, (mget, kvp) => new CasResult<object>
@@ -965,10 +970,7 @@ protected virtual IDictionary<string, T> PerformMultiGet<T>(IEnumerable<string>
                                 string original;
                                 if (hashed.TryGetValue(kvp.Key, out original))
                                 {
-                                    Console.WriteLine(kvp.Key);
-                                    Console.WriteLine(kvp.Value);
                                     var result = collector(mget, kvp);
-
                                     // the lock will serialize the merge,
                                     // but at least the commands were not waiting on each other
                                     lock (retval) retval[original] = result;
@@ -992,6 +994,48 @@ protected virtual IDictionary<string, T> PerformMultiGet<T>(IEnumerable<string>
             return retval;
         }
 
+        protected virtual async Task<IDictionary<string, T>> PerformMultiGetAsync<T>(IEnumerable<string> keys, Func<IMultiGetOperation, KeyValuePair<string, CacheItem>, T> collector)
+        {
+            // transform the keys and index them by hashed => original
+            // the mget results will be mapped using this index
+            var hashed = new Dictionary<string, string>();
+            foreach (var key in keys)
+            {
+                hashed[this.keyTransformer.Transform(key)] = key;
+            }
+
+            var byServer = GroupByServer(hashed.Keys);
+
+            var retval = new Dictionary<string, T>(hashed.Count);
+            var tasks = new List<Task>();
+
+            //execute each list of keys on their respective node
+            foreach (var slice in byServer)
+            {
+                var node = slice.Key;
+                var nodeKeys = slice.Value;
+                var mget = this.pool.OperationFactory.MultiGet(nodeKeys);
+                var task = Task.Run(async () => 
+                {
+                    if ((await node.ExecuteAsync(mget)).Success)
+                    {
+                        foreach (var kvp in mget.Result)
+                        {
+                            if (hashed.TryGetValue(kvp.Key, out var original))
+                            {
+                                lock (retval) retval[original] = collector(mget, kvp);
+                            }
+                        }
+                    }
+                });
+                tasks.Add(task);               
+            }
+
+            await Task.WhenAll(tasks);
+
+            return retval;
+        }
+
         protected Dictionary<IMemcachedNode, IList<string>> GroupByServer(IEnumerable<string> keys)
         {
             var retval = new Dictionary<IMemcachedNode, IList<string>>();

Enyim.Caching/NullMemcachedClient.cs

@@ -88,6 +88,11 @@ public IDictionary<string, T> Get<T>(IEnumerable<string> keys)
             return new Dictionary<string, T>();
         }
 
+        public Task<IDictionary<string, T>> GetAsync<T>(IEnumerable<string> keys)
+        {
+            return Task.FromResult<IDictionary<string, T>>(new Dictionary<string, T>());
+        }
+
         public object Get(string key)
         {
             return null;

MemcachedTest/MemcachedClientTest.cs

@@ -261,7 +261,7 @@ public async Task MultiGetTest()
                     Assert.True(await client.StoreAsync(StoreMode.Set, k, i, DateTime.Now.AddSeconds(30)), "Store of " + k + " failed");
                 }
 
-                IDictionary<string, int> retvals = client.Get<int>(keys);
+                IDictionary<string, int> retvals = await client.GetAsync<int>(keys);
 
                 Assert.NotEmpty(retvals);
                 Assert.Equal(keys.Count, retvals.Count);
@@ -282,7 +282,7 @@ public async Task MultiGetTest()
                 await client.StoreAsync(StoreMode.Set, key1, obj1, DateTime.Now.AddSeconds(10));
                 await client.StoreAsync(StoreMode.Set, key2, obj2, DateTime.Now.AddSeconds(10));
 
-                var multiResult = client.Get<HashSet<int>>(new string[] { key1, key2 });
+                var multiResult = await client.GetAsync<HashSet<int>>(new string[] { key1, key2 });
                 Assert.Equal(2, multiResult.Count);
                 Assert.Equal(obj1.First(), multiResult[key1].First());
                 Assert.Equal(obj2.First(), multiResult[key2].First());