Repository navigation
Python3 AsyncIO implementation #55
Description
Activity
This package already provides asynchronicity (
Producer.send_asyncand themessage_listenerargument to the client).While those don't provide
asyncorawaitsyntactic constructs, they are asynchronous (nonblocking) and can be integrated with Python's asyncio (or trio etc.) futures to provide awaitable tokens.How would async/await syntax save on IO waiting time if this client is already asynchronous? Something would still have to wait on I/O as is already being done.
I think a larger point to possibly be made here is the difficulty of using pulsar with other libraries without these features. For example, consider a user trying to write a simple FastAPI server that simply exposes a REST endpoint for a client that gives that latest value in a topic.
If they were using Kafka, they could use
broadcaster(like suggested here), or they could useaiokafka(an example of this would be here and here). These third party libraries that allow background processing generally require coroutines to be passed in and don't support synchronous functions because that's obviously a nightmare with python and GIL.I'm not even sure on how to nicely implement the FastAPI example with Pulsar client (apart from implementing blocking threads), or if the increased complexity means that for many Python developers they might simply see the provided kafka solutions and be swayed.
Reacted by Joeri Hermans, Travis Sturzl, Ralph Meijer, Emanuele Domingo and Nikolas Achatz@zbentley Integrating this client with asyncio is frustrating, because it's not entirely clear which thread things are running in and much of asyncio is not threadsafe. In order to get these callbacks to work with asyncio you have to resort to low level threadsafe facilities of asyncio, making it far less trivial to safely integrate this client with asyncio than you portray.
For example if you want to use
send_async. Intuition might lead you to just pass a future into that callback, and set the result of that future when you're done. You cannot do that, because Future is not threadsafe. You instead need to callfuture.set_resultusingloop.call_soon_threadsafe(future.set_result, res)to ensure thatset_futureis called from the main thread. This is actually a simpler example, whereas wrapping message_listener callbacks to support async callbacks is even more frustrating. Basically you're always forced to use low-level threadsafe facilities to schedule into the main threads event loop.I think the argument that this is strictly syntactical is simply not true. These concurrency models are fundamentally different. Pulsar is using a threaded blocking network model, where threads handle blocking operation as to not block the main thread, these threads are put to sleep by the OS when they make a blocking syscall. Asyncio is driven by an event loop and non-blocking operations running on a single thread. I understand if an asyncio wrapper is outside the scope of this project, but these concurrency models diverge on more than just syntax, and integrating with asyncio is not trivial or straightforward especially for most python developers who may not understand the low level details of how asyncio works.
Reacted by Ralph Meijer and Emanuele DomingoAssign this issue to myself. I wrote a demo locally and I'm going to start this work.
I see the asyncio module in the reference docs (https://pulsar.apache.org/api/python/3.5.x/pulsar.asyncio.html). However, I get
ModuleNotFoundError: No module named 'pulsar.asyncio'when I try to import. pip installing "pulsar-client[all]==3.5.0" doesn't help.Can someone suggest the proper way to access these classes using the Python Pulsar Client?. Info @BewareMyPower . Does it require a custom build from src?
It might be something wrong with the release process, let me check it next week.
Reacted by daveqsI can find this module when I installed it:
>>> pulsar.__file__ '/opt/homebrew/lib/python3.12/site-packages/pulsar/__init__.py' >>> import pulsar.asyncio >>> pulsar.__version__ '3.5.0'Could you check if you installed an old version that is not removed? @daveqs
Is there any plan to support asyncio for consumer?
No plan for now
No plan for now
sounds a bad news.
Reacted by Nikolas AchatzHi guys, i am currently implementing the consumer using asyncio in my fork and would like to open a pull request as soon as its finished. I used the c++ clients Async methods in c++ and futures in python (in the same way as the producer) to archive nonblocking methods. As soon as the code is ready help and feedback would be greatly appreciated.
Reacted by Yunze Xu, Carlos A. Wong and Nikolas AchatzReacted by Cherilyn Buren, shubhmay-potdar-aera and Nikolas AchatzHey all, looks like we're closing in on a year later (almost three from the initial issue). Curious if there's any updates here, this seems to be a fairly large oversight for async based applications. Is the pull request for this still underway and does it need any assistance? This feels to motivate the transition to other messaging technologies for async applications
I think no one is working on this issue currently.
- There is a open PR for the consumer side implementation, but the comments are not addressed: Implementation of the asyncio consumer #228
- There is a follow-up issue to add more configurations for producer: Implementation of the asyncio consumer #228
Reacted by JSDuI added all missed partsI added the consumer support in #277
Copy from: apache/pulsar#6308
Is your feature request related to a problem? Please describe.
Implement python3 await/ async and using the event loop in order to save IO waiting time
Describe the solution you'd like
Using async, await and execute via the event loop
Describe alternatives you've considered
Using threads although the GIL preventing pure multi threading