**Apache Airflow version**:
2.0.1

**Kubernetes version (if you are using kubernetes)** (use `kubectl version`):
Not applicable

**Environment**:

- **Cloud provider or hardware configuration**: None
- **OS** (e.g. from /etc/os-release): Ubuntu
- **Kernel** (e.g. `uname -a`): 
- **Install tools**:
- **Others**:

**What happened**:

1. Ran airflow db init on mysql in parallel in 2 command lines. Only one command did the migrations, the other one waited. But connections were inserted twice. I would like them not to be added twice.
2. Ran airflow db init on postgres in parallel in 2 command lines. Both command lines started doing migrations on the same db in parallel. I would like one command to run, the other to wait.

**What you expected to happen**:

1. For MySQL. Connections and other config objects to be inserted only once.
2. For Postgres. Only one migration can be performed in the same time for the same db.

**How to reproduce it**:
Scenario 1:
Setup Airflow so that it uses MySQL.
Run `airflow init db` in 2 command lines, side by side.

Scenario 2:
Setup Airflow so that it uses Postgres.
Run `airflow init db` in 2 command lines, side by side.

**Anything else we need to know**:
This problem occurs every time.
airflow/utils/session.py
"""Contextmanager that will create and teardown a global db lock."""
    dialect = session.connection().dialect
    try:
        if dialect.name == 'postgresql':
            session.connection().execute(f'select PG_ADVISORY_LOCK({pg_lock_id});')
        if dialect.name == 'mysql' and dialect.server_version_info >= (
            5,
            6,
        ):
            session.connection().execute(f"select GET_LOCK('{lock_name}',{mysql_lock_timeout});")
        if dialect.name == 'mssql':
            session.connection().execute(
                f"sp_getapplock @Resource = '{lock_name}', @LockMode = 'Exclusive', @LockOwner = 'Session';"
            )
        yield None
    finally:
        if dialect.name == 'postgresql':
            session.connection().execute(f'select PG_ADVISORY_UNLOCK({pg_lock_id});')
        if dialect.name == 'mysql' and dialect.server_version_info >= (
            5,
            6,
        ):
            session.connection().execute(f"select RELEASE_LOCK('{lock_name}');")
        if dialect.name == 'mssql':
            session.connection().execute(f"sp_releaseapplock @Resource = '{lock_name}';")