feat: Add queues for triggerers.

[avkirilishin]

Closes: #33818

Pull Request for discussing the approach. If this approach is acceptable, I will add queue support for more triggers, add tests, update the documentation, and assess the database load under a large number of triggers.

You can check how triggerer's queues work with the sample DAG:

from datetime import time
import pendulum

from airflow import DAG
from airflow.operators.empty import EmptyOperator
from airflow.sensors.time_sensor import TimeSensorAsync


with DAG(
    'trigger-dag',
    schedule='@daily',
    start_date=pendulum.DateTime(2023, 12, 19),
) as dag:
    sensor = TimeSensorAsync(task_id="sync_sensor", target_time=time(10, 59))
    sensor_default = TimeSensorAsync(
        task_id="sync_sensor_default", target_time=time(23, 59), trigger_queue="default")
    sensor_q1 = TimeSensorAsync(task_id="sync_sensor_q1", target_time=time(10, 59), trigger_queue="queue1")
    dummy_op = EmptyOperator(task_id="dummy_op")
    sensor >> dummy_op
    sensor_default >> dummy_op
    sensor_q1 >> dummy_op

And you can run the second triggerer with the following command:

airflow triggerer -q queue1

[github-actions]

This pull request has been automatically marked as stale because it has not had recent activity. It will be closed in 5 days if no further activity occurs. Thank you for your contributions.

[github-actions]

This pull request has been automatically marked as stale because it has not had recent activity. It will be closed in 5 days if no further activity occurs. Thank you for your contributions.