134、gRPC实现推荐系统客户端和服务端

编写客户端

在toutiao-backend/common/rpc目录下新建client.py

import grpc
import reco_pb2
import reco_pb2_grpc
import time


def feed_articles(stub):
    # 构建rpc调用的调用参数
    user_request = reco_pb2.UserRequest()
    user_request.user_id = '1'
    user_request.channel_id = 1
    user_request.article_num = 10
    user_request.time_stamp = round(time.time()*1000)

    # 通过stub进行方法调用，并接收调用返回值
    ret = stub.user_recommend(user_request)
    print('ret={}'.format(ret))

def run():
    """
    rpc客户端调用的方法
    """
    # 使用with语句连接rpc服务器
    with grpc.insecure_channel('127.0.0.1:8888') as channel:
        # 创建调用rpc远端服务的辅助对象stub
        stub = reco_pb2_grpc.UserRecommendStub(channel)
        # 通过stub进行rpc调用
        feed_articles(stub)

if __name__ == '__main__':
    run()
           

补全服务端

为了方便看到效果，我们编写补全服务端代码。

注意：此处实际推荐的代码在后续推荐系统课程中会涉及到

在toutiao-backend/common/rpc目录下新建server.py文件

import reco_pb2
import reco_pb2_grpc
import grpc
from concurrent.futures import ThreadPoolExecutor
import time


# rpc接口定义中服务对应成Python的类
class UserRecommendService(reco_pb2_grpc.UserRecommendServicer):

    # 在接口定义的同名方法中补全，被调用时应该执行的逻辑
    def user_recommend(self, request, context):
        # request是调用的请求数据对象
        user_id = request.user_id
        channel_id = request.channel_id
        article_num = request.article_num
        time_stamp = request.time_stamp

        response = reco_pb2.ArticleResponse()
        response.exposure = 'exposure param'
        response.time_stamp = round(time.time()*1000)
        recommends = []
        for i in range(article_num):
            article = reco_pb2.Article()
            article.track.click = 'click param {}'.format(i+1)
            article.track.collect = 'collect param {}'.format(i+1)
            article.track.share = 'share param {}'.format(i+1)
            article.track.read = 'read param {}'.format(i+1)
            article.article_id = i+1
            recommends.append(article)
        response.recommends.extend(recommends)

        # 最终要返回一个调用结果
        return response


def serve():
    """
    rpc服务端启动方法
    """
    # 创建一个rpc服务器
    server = grpc.server(ThreadPoolExecutor(max_workers=10))

    # 向服务器中添加被调用的服务方法
    reco_pb2_grpc.add_UserRecommendServicer_to_server(UserRecommendService(), server)

    # 微服务器绑定ip地址和端口
    server.add_insecure_port('127.0.0.1:8888')

    # 启动rpc服务
    server.start()

    # start()不会阻塞，此处需要加上循环睡眠 防止程序退出
    while True:
        time.sleep(10)


if __name__ == '__main__':
    serve()