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
with dask_testing_cluster(
            worker_kwargs={'security': tls_security(), "protocol": "tls"},
            scheduler_kwargs={'security': tls_security(), "protocol": "tls"},
        ) as (cluster, _):
            executor = DaskExecutor(cluster_address=cluster['address'])
            self.assert_tasks_on_executor(executor)
            executor.end()
            
            
            executor.client.close()
    @mock.patch('airflow.executors.dask_executor.DaskExecutor.sync')
    @mock.patch('airflow.executors.base_executor.BaseExecutor.trigger_tasks')
    @mock.patch('airflow.executors.base_executor.Stats.gauge')
    def test_gauge_executor_metrics(self, mock_stats_gauge, mock_trigger_tasks, mock_sync):
        executor = DaskExecutor()
        executor.heartbeat()
        calls = [
            mock.call('executor.open_slots', mock.ANY),
            mock.call('executor.queued_tasks', mock.ANY),
            mock.call('executor.running_tasks', mock.ANY),
        ]
        mock_stats_gauge.assert_has_calls(calls)