Use pipeline for _set

This reduces the number of connections used, as every command takes a connection from the pool and returns it afterwards.

Author: pepijndevos

celery/backends/redis.py

@@ -148,12 +148,13 @@ def set(self, key, value, **retry_policy):
         return self.ensure(self._set, (key, value), **retry_policy)
 
     def _set(self, key, value):
-        client = self.client
+        pipe = self.client.pipeline()
         if self.expires:
-            client.setex(key, value, self.expires)
+            pipe.setex(key, value, self.expires)
         else:
-            client.set(key, value)
-        client.publish(key, value)
+            pipe.set(key, value)
+        pipe.publish(key, value)
+        pipe.execute()
 
     def delete(self, key):
         self.client.delete(key)