distributed Dask queue : producer consumer

Hello, 😄

I'm a data scientist/data engineer working on a specific workflow where I need to process a huge amount of documents so i tried using `distributed.Queue` to right a producer/consumer pattern using Dask distributed.

I might miss something about how to correctly implement the producer/consumer using distributed but here are some important features that could be added to `distributed.Queue`. I would also like to help writing this class, so thanks for your guidance :

- [ ] The queue is not garbage collected, calling `del q` doesnt actually free up the distributed memory
- [ ] We might need to implement a `q.join()` in distributed manner to block until all the queue items have been processed
- [ ] Batching in queue doesnt have a '_last object mode'_ where we return lesser objects than the `batch_size` if the queue is empty
- [ ] Spawning a "pool of consumers" in a distributed system can be tricky : basically the machines producing are also the ones waiting for the tasks to be added ? Is there a way to load balance producers and consumers in the dask scheduler ?

Thanks a lot and I hope these features make sens. I hope, I can be of some help in advancing this class !