Currently airflow's DaskExecutor does  not support specifying queues for tasks, due to dask's lack of an explicit queue specification feature. However, this can be reliably mimicked using dask resources ([details here](https://distributed.dask.org/en/latest/resources.html)). So the set up would look something like this:

```
# starting dask worker that can service airflow tasks submitted with queue=queue_name_1 or queue_name_2
$ dask-worker <address> --resources "queue_name_1=inf, queue_name_2=inf"
```
~~(Unfortunately as far as I know you need to provide a finite resource limit for the workers, so you'd need to provide an arbitrarily large limit, but I think it's worth the minor inconvenience to allow a queue functionality in the dask executor.)~~

```
# airflow/executors/dask_executor.py
def execute_async(
    self,
    key: TaskInstanceKey,
    command: CommandType,
    queue: Optional[str] = None,
    executor_config: Optional[Any] = None,
) -> None:

    self.validate_command(command)

    def airflow_run():
        return subprocess.check_call(command, close_fds=True)

    if not self.client:
        raise AirflowException(NOT_STARTED_MESSAGE)

    ################ change made here #################
    resources = None
    if queue:
        resources = {queue: 1}

    future = self.client.submit(airflow_run, pure=False, resources=resources)
    self.futures[future] = key  # type: ignore
```

tests/executors/test_dask_executor.py
def assert_tasks_on_executor(self, executor):
        success_command = ['airflow', 'tasks', 'run', '--help']
        fail_command = ['airflow', 'tasks', 'run', 'false']
        
        executor.start()
        executor.execute_async(key='success', command=success_command)
        executor.execute_async(key='fail', command=fail_command)
        success_future = next(k for k, v in executor.futures.items() if v == 'success')
        fail_future = next(k for k, v in executor.futures.items() if v == 'fail')
        
        timeout = timezone.utcnow() + timedelta(seconds=30)
        while not (success_future.done() and fail_future.done()):
            if timezone.utcnow() > timeout:
                raise ValueError(
                    'The futures should have finished; there is probably '
                    'an error communicating with the Dask cluster.'
                )
        
        assert success_future.done()
        assert fail_future.done()
        
        assert success_future.exception() is None
        assert fail_future.exception() is not None
class TestDaskExecutor(TestBaseDask):