Skip to main content

RPC protocol based on kafka

Project description

Kafka RPC


Introduction

Kafka RPC, a RPC protocol that based on kafka, is meant to provide a swift, stable, reliable remote calling service.

The reason we love about kafka is its fault tolerance, scalability and wicked large throughput.

So if you want a RPC service with kafka features, kRPC is the kind of tool you're looking for.


Installation

    pip install kafka-rpc

FAQ

  1. What is RPC?

    RPC is a request–response protocol. An RPC is initiated by the client, which sends a request message to a known remote server to execute a specified procedure with supplied parameters.

    The one that posts the job request is called Client, while the other one that gets the job and responses is called Server.

  2. When should I use a RPC?

    When you have to call a function that doesn't belong to local process or computer, but you don't want to build up another complex network framework just to implement restful api or soap, etc. RPC is much faster than restful api and easier to use. This is only a very few lines of code to be adjusted to implement a RPC service.

  3. Why kafka-rpc, not the other RPC protocols like zerorpc, grpc, mprpc?

    Here is the comparison.

    RPC MiddleWare Serialization Speed(QPS) Features
    kafka-rpc kafka msgpack 200+ (sync) 4700+(async) dynamic load rebalance, large throughput, data persistence, faster serialization
    zerorpc zeromq msgpack 450+ dynamic load rebalance(failed when all server are busy), faster serialization
    grpc unknown protobuf not tested dynamic load rebalance, large throughput, support only function rpc
    mprpc no msgpack 19000+ lightspeed

    Benchmark enviroment:

    Ubuntu 19.10 x64 (5.3.0-21-generic)

    Intel i5-8400


    The only reason that I developed kafka-rpc is that zerorpc failed me!

    After months of searching and testing, zerorpc was the best rpc service I'd ever used, but it's bugged! Normally, developers won't notice, because most of the time, we use RPC to post the job from one client directly to one server.

    But when you develop a distributing system, you will have to post K jobs of different types from N clients to M servers.

    The problem is that you don't really know which server is currently available, or right one for the job.

    And that's where load balancing comes in.

    Zeromq supports that, and zerorpc supports that too. In reverse proxy mode, zerorpc will post the jobs, but not sending to servers. Available servers will come looking for job actively. So the job will always be coped by the most available, therefore the most performant servers. However, sadly, zerorpc doesn't really queue up the job, so when you have no server available at all, the jobs will not wait in line but instead, be abandoned.

    That's why I need kafka. Unlike most of the MQ, kafka provides data persistence, so no more job abandon. When all servers is unavailable, the job will queue up and wait in line.

    If the client crash, jobs will still be on disk (kafka features), unharmed, with replicas(can you imagine that).

    Also, kafka-rpc supports dynamic scalability, servers can be always added to the cluster or be removed, so jobs will be fairly distribute to all the servers, and will be reroute to another healthy server if assigned server is down.

    Despite the minor disadvantages, they are all good tool developed by great programmers, you're always welcome to contribute to their original repositories and my forked zerorpc and mprpc, which support numpy array.

Next Step

  • [X] optimize the QPS by allow asynchronous calls considering its large throughput advantage
  • [X] use gevent instead of built-in threading to speed up and now it's 40% faster.
  • [ ] rewrite it in cython

Usage

Assuming you already have a object and everything works, like this

local_call.py

# Part1: define a class
class Sum:
    def add(self, x, y):
        return x + y

# Part2: instantiate a class to an object
s = Sum()

# Part3: call a method of the object
result = s.add(1, 2)  # result = 3

Then you can use RPC to run Part1 and Part2 on process1, and call Part3, the method of process 1 from process2

kafka_rpc_server_demo.py

from kafka_rpc import KRPCServer

# Part1: define a class
class Sum:
    def add(self, x, y):
        return x + y

# Part2: instantiate a class to an object
s = Sum()

# assuming you kafka broker is on 0.0.0.0:9092
krs = KRPCServer('0.0.0.0:9092', handle=s, topic_name='sum')
krs.server_forever()

kafka_rpc_client_demo.py

from kafka_rpc import KRPCClient

# assuming you kafka broker is on 0.0.0.0:9092
krc = KRPCClient('0.0.0.0:9092', topic_name='sum')

# call method from client to server
result = krc.add(1, 2)

print(result)

krc.close()

# you can find the returned result in result['ret']
# result = {
#     'ret': 3,
#     'tact_time': 0.007979869842529297,  # total process time
#     'tact_time_server': 0.006567955017089844,  # process time on server side
#     'server_id': '192.168.1.x'  # client ip
# }

Advanced Usage

  1. enable server side concurrency. Respectively, multiple requests must be sent concurrently.

    kafka_rpc_server_concurrency_demo.py

     import time
     from kafka_rpc import KRPCServer
     
     
     # Part1: define a class
     class Sum:
         def add(self, x, y):
     
             # simulate blocking actions like I/O
             time.sleep(0.1)
     
             return x + y
     
     
     # Part2: instantiate a class to an object
     s = Sum()
     
     # assuming you kafka broker is on 0.0.0.0:9092
     krs = KRPCServer('0.0.0.0:9092', handle=s, topic_name='sum', concurrent=128)
     krs.server_forever()
    

    kafka_rpc_client_async_demo.py

     from concurrent.futures import ThreadPoolExecutor, as_completed
     from kafka_rpc import KRPCClient
     
     pool = ThreadPoolExecutor(128)
     
     # assuming you kafka broker is on 0.0.0.0:9092
     krc = KRPCClient('0.0.0.0:9092', topic_name='sum')
     
     # call method concurrently from client to server
     # use pool.map if you like
     futures = []
     for i in range(128):
         futures.append(pool.submit(krc.add, 1, 2, timeout=20))
     
     for future in as_completed(futures):
         result = future.result()
         print(result)
     
     krc.close()
    
  2. enable redis to speed up caching and temporarily store input/output data, by adding use_redis=True to KRPCClient, or specify redis port, db and password. But redis doesn't support async operations, it will crash, it's only faster in sync mode.

     krc = KRPCClient('0.0.0.0:9092', topic_name='sum', use_redis=True, redis_port=6379, redis_db=0, redis_password='kafka_rpc.no.1')
    
  3. enhance the communication security, by adding verify=True or encrypt='whatever_password+you/want' or both to both of the client and the server.But enabling verification and encryption will have a little impact on performance.

     # basic verification and encryption
     krs = KRPCServer('0.0.0.0:9092', handle=s, topic_name='sum', verify=True, encrypt='whatever_password+you/want')
     krc = KRPCClient('0.0.0.0:9092', 9092, topic_name='sum', verify=True, encrypt='whatever_password+you/want')
     
     # advanced verification and encryption with custom hash function, input: bytes, output: bytes
     krs = KRPCServer('0.0.0.0:9092', handle=s, topic_name='sum', verify=True, encrypt='whatever_password+you want', verification=lambda x: sha3_224(x).hexdigest().encode())
     krc = KRPCClient('0.0.0.0:9092', topic_name='sum', verify=True, encrypt='whatever_password+you/want', verification=lambda x: sha3_224(x).hexdigest().encode())
    

Warning

If use_redis=False, KRPCClient cannot be instantiated more than once

If use_redis=False, KRPCClient cannot be instantiated more than once

If use_redis=False, KRPCClient cannot be instantiated more than once

Do you remember now?

Project details


Download files

Download the file for your platform. If you're not sure which to choose, learn more about installing packages.

Files for kafka-rpc, version 1.0.8
Filename, size File type Python version Upload date Hashes
Filename, size kafka-rpc-1.0.8.tar.gz (15.8 kB) File type Source Python version None Upload date Hashes View hashes

Supported by

Elastic Elastic Search Pingdom Pingdom Monitoring Google Google BigQuery Sentry Sentry Error logging AWS AWS Cloud computing DataDog DataDog Monitoring Fastly Fastly CDN DigiCert DigiCert EV certificate StatusPage StatusPage Status page