RabbitMQ: Python naudojimas prisijungimui prie RabbitMQ
Ši dokumentacija yra Darbo pradžia vadovo dalis. Visą vadovą rasite čia: Kaip naudotis Stackhero su RabbitMQ.
Pažvelkime, kaip galite prijungti savo Python aplikaciją prie RabbitMQ naudodami Aio Pika biblioteką. Daugeliu atvejų, saugiam prisijungimui pakanka nurodyti AMQPS URL:
connection = await aio_pika.connect_robust(
"amqps://admin:<PASSWORD>@<XXXXXX>.stackhero-network.com:<AMQP_PORT_TLS>",
)
Toliau pateikiamas pilnas pavyzdys, kaip užmegzti saugų ryšį su RabbitMQ, sukurti kanalą ir deklaruoti bazinę eilę (queue). Tai puikus būdas patikrinti jūsų konfigūraciją:
import asyncio
import logging
import aio_pika
async def main() -> None:
# Jei norite matyti debug žurnalus, galite atkomentuoti šią eilutę
# logging.basicConfig(level=logging.DEBUG)
connection = await aio_pika.connect_robust(
"amqps://admin:<PASSWORD>@<XXXXXX>.stackhero-network.com:<AMQP_PORT_TLS>"
)
async with connection:
print("The connection worked!")
channel = await connection.channel()
await channel.set_qos(prefetch_count=10)
queue = await channel.declare_queue("test_queue", auto_delete=True)
if __name__ == "__main__":
asyncio.run(main())
Klaidos unable to get local issuer certificate sprendimas
Jei jungiantis iš Python gaunate tokią klaidą:
aiormq.exceptions.AMQPConnectionError: [Errno 5] [SSL: CERTIFICATE_VERIFY_FAILED] certificate verify failed: unable to get local issuer certificate (_ssl.c:1006)
dažniausiai tai reiškia, kad jūsų sistemoje trūksta Let's Encrypt CA sertifikato. Norėdami tai išspręsti, įdiekite savo operacinei sistemai skirtą CA sertifikatų paketą:
-
Ubuntu/Debian sistemose vykdykite:
sudo apt install ca-certificates -
Alpine Linux sistemose vykdykite:
apk add ca-certificates
Jei negalite naudoti šių komandų, vis dar galite įdiegti CA sertifikatą rankiniu būdu:
-
Atsisiųskite Let's Encrypt CA sertifikatą iš https://letsencrypt.org/certs/isrgrootx1.pem.
-
Tuomet Python kode jungiantis prie RabbitMQ nurodykite CA sertifikato failą:
import ssl ssl_context = ssl.create_default_context() ssl_context.load_verify_locations(cafile='isrgrootx1.pem') connection = await aio_pika.connect_robust( "amqps://admin:<PASSWORD>@<XXXXXX>.stackhero-network.com:<AMQP_PORT_TLS>", ssl_context=ssl_context )
Toliau pateikiamas pilnas pavyzdys, kaip naudoti Let's Encrypt CA sertifikatą saugiam prisijungimui:
import asyncio
import logging
import ssl
import aio_pika
async def main() -> None:
# Norėdami įjungti debug žurnalus, galite atkomentuoti šią eilutę
# logging.basicConfig(level=logging.DEBUG)
ssl_context = ssl.create_default_context()
# Įkelkite atsisiųstą Let's Encrypt CA sertifikatą
# Pavyzdžiui: wget https://letsencrypt.org/certs/isrgrootx1.pem
ssl_context.load_verify_locations(cafile='isrgrootx1.pem')
connection = await aio_pika.connect_robust(
"amqps://admin:<PASSWORD>@<XXXXXX>.stackhero-network.com:<AMQP_PORT_TLS>",
ssl_context=ssl_context
)
async with connection:
print("The connection worked!")
channel = await connection.channel()
await channel.set_qos(prefetch_count=10)
queue = await channel.declare_queue("test_queue", auto_delete=True)
if __name__ == "__main__":
asyncio.run(main())