最新国产好看的视频,伊人天堂AV在线,国产Aaaaaa视频,蜜臀视频在线观看一区,人妻av色图,密臀久久久精品影片,青青视频免费观看毛片,久草在线观看视,国产三级精品色情在线

利用Python語言的grpc實現(xiàn)消息傳送詳解

 更新時間:2023年03月21日 09:49:22   作者:we34dfg  
gRPC是一個高性能、通用的開源RPC框架,其由Google主要面向移動應用開發(fā)并基于HTTP/2協(xié)議標準而設計。本文主要介紹了如何利用Python語言的grpc實現(xiàn)消息傳送,感興趣的可以了解一下

1. grpc開源包的安裝

# conda
$ conda create -n grpc_env python=3.9
 
# install grpc
$ pip install grpc -i https://pypi.doubanio.com/simple
$ pip install grpc-tools -i https://pypi.doubanio.com/simple
 
# 有時proto生成py文件不對就是得換換grpc兩個包的版本

2. grpc的使用之傳送消息

整體結(jié)構(gòu),client.py server.py 和proto目錄下的example.proto

1)在example.proto定義傳送體

// 聲明
syntax = "proto3";
package proto;
 
// service創(chuàng)建
service HelloService{
  rpc Hello(Request) returns (Response) {}  // 單單傳送消息
}
 
// 請求參數(shù)消息體 1、2是指參數(shù)順序
message Request {
  string data = 1;
}
 
// 返回參數(shù)消息體
message Response {
  int32 ret = 1;    //返回碼
  string data = 2;
}
 
//python -m grpc_tools.protoc -I ./ --python_out=./ --grpc_python_out=./ ./example.proto

2) 在虛擬環(huán)境里使用命令生成py文件

$ conda activate grpc_env
$ f:
$ cd F:\examples
$ python -m grpc_tools.protoc -I ./ --python_out=./ --grpc_python_out=./ ./example.proto

在proto目錄下會生成兩個py文件,如下圖所示:

3) 編輯client.py 和 server.py

# server.py
import time
import grpc
from concurrent import futures
from proto import example_pb2_grpc, example_pb2
 
 
class ServiceBack(example_pb2_grpc.HelloServiceServicer):
    """接口的具體功能實現(xiàn)"""
 
    def Hello(self, request, context):
        """hello"""
        data = request.data
        print(data)
        ret_data = "Response:" + data
        return example_pb2.Response(ret=0, data=ret_data)
 
 
def server(ip: str, port: int) -> None:
    server = grpc.server(futures.ThreadPoolExecutor(max_workers=10))  # ??為10的線程池
    ai_servicer = ServiceBack()
    example_pb2_grpc.add_HelloServiceServicer_to_server(ai_servicer, server)
    server.add_insecure_port(f"{ip}:{port}")  
    server.start()
    try:
        print(f"server is started! ip:{ip} port:{str(port)}")
        while True:
            time.sleep(60 * 60)
    except Exception as es:
        print(es)
        server.stop(0)
 
 
if __name__ == '__main__':
    server("127.0.0.1", 8000)
# client.py
import grpc
from proto import example_pb2_grpc, example_pb2
 
 
def client(ip: str, port: int) -> None:
    target = str(ip) + ":" + str(port)
    channel = grpc.insecure_channel(target)  # 連接rpc服務器
    cli = example_pb2_grpc.HelloServiceStub(channel)  # 創(chuàng)建Stub
 
    data = "hello 123"
    request = example_pb2.Request(data=data)
    res = cli.Hello(request)
    print(f"ret:{res.ret}, data:{res.data}")
 
 
if __name__ == '__main__':
    client("127.0.0.1", 8000)

3. grpc的使用之數(shù)據(jù)傳輸大小配置

默認情況下,gRPC 將傳入消息限制為 4 MB。 傳出消息沒有限制。

1)example.proto定義不變

2)編輯client.py 和 server.py

# server.py
import time
import grpc
from concurrent import futures
from proto import example_pb2_grpc, example_pb2
 
 
class ServiceBack(example_pb2_grpc.HelloServiceServicer):
    """接口的具體功能實現(xiàn)"""
 
    def Hello(self, request, context):
        """hello"""
        data = request.data
        print(data)
        ret_data = "Response:" + data
        return example_pb2.Response(ret=0, data=ret_data)
 
 
def server(ip: str, port: int) -> None:
    # 數(shù)據(jù)傳輸大小配置
    max_message_length = 1024 * 1024 * 1024  # 1G
    options = [('grpc.max_send_message_length', max_message_length),
               ('grpc.max_receive_message_length', max_message_length),
               ('grpc.enable_retries', 1),
               ]
    server = grpc.server(futures.ThreadPoolExecutor(max_workers=10), options=options)  # ??為10的線程池
    ai_servicer = ServiceBack()
    example_pb2_grpc.add_HelloServiceServicer_to_server(ai_servicer, server)
    server.add_insecure_port(f"{ip}:{port}")  
    server.start()
    try:
        print(f"server is started! ip:{ip} port:{str(port)}")
        while True:
            time.sleep(60 * 60)
    except Exception as es:
        print(es)
        server.stop(0)
 
 
if __name__ == '__main__':
    server("127.0.0.1", 8000)
# client.py
import grpc
from proto import example_pb2_grpc, example_pb2
 
 
def client(ip: str, port: int) -> None:
    # 數(shù)據(jù)傳輸大小配置
    max_message_length = 1024 * 1024 * 1024  # 1G
    options = [('grpc.max_send_message_length', max_message_length),
               ('grpc.max_receive_message_length', max_message_length),
               ('grpc.enable_retries', 1),
               ]
    target = str(ip) + ":" + str(port)
    channel = grpc.insecure_channel(target, options=options)  # 連接rpc服務器
    cli = example_pb2_grpc.HelloServiceStub(channel)  # 創(chuàng)建Stub
 
    data = "hello 123" * 1024 * 1024
    request = example_pb2.Request(data=data)
    res = cli.Hello(request)
    print(f"ret:{res.ret}, data:{res.data}")
 
 
if __name__ == '__main__':
    client("127.0.0.1", 8000)

4. grpc的使用之超時配置

1)example.proto定義不變

2)編輯client.py 和 server.py

# server.py
import time
import grpc
from concurrent import futures
from proto import example_pb2_grpc, example_pb2
 
 
class ServiceBack(example_pb2_grpc.HelloServiceServicer):
    """接口的具體功能實現(xiàn)"""
 
    def Hello(self, request, context):
        """hello"""
        data = request.data
        print(data)
        time.sleep(2)
        ret_data = "Response:" + data
        return example_pb2.Response(ret=0, data=ret_data)
 
 
def server(ip: str, port: int) -> None:
    # 數(shù)據(jù)傳輸大小配置
    max_message_length = 1024 * 1024 * 1024  # 1G
    options = [('grpc.max_send_message_length', max_message_length),
               ('grpc.max_receive_message_length', max_message_length),
               ('grpc.enable_retries', 1),
               ]
    server = grpc.server(futures.ThreadPoolExecutor(max_workers=10), options=options)  # ??為10的線程池
    ai_servicer = ServiceBack()
    example_pb2_grpc.add_HelloServiceServicer_to_server(ai_servicer, server)
    server.add_insecure_port(f"{ip}:{port}")  
    server.start()
    try:
        print(f"server is started! ip:{ip} port:{str(port)}")
        while True:
            time.sleep(60 * 60)
    except Exception as es:
        print(es)
        server.stop(0)
 
 
if __name__ == '__main__':
    server("127.0.0.1", 8000)
# client.py
import sys
import grpc
from proto import example_pb2_grpc, example_pb2
 
 
def client(ip: str, port: int) -> None:
    # 數(shù)據(jù)傳輸大小配置
    max_message_length = 1024 * 1024 * 1024  # 1G
    options = [('grpc.max_send_message_length', max_message_length),
               ('grpc.max_receive_message_length', max_message_length),
               ('grpc.enable_retries', 1),
               ]
    target = str(ip) + ":" + str(port)
    channel = grpc.insecure_channel(target, options=options)  # 連接rpc服務器
    cli = example_pb2_grpc.HelloServiceStub(channel)  # 創(chuàng)建Stub
    try:
        data = "hello 123"
        request = example_pb2.Request(data=data)
        res = cli.Hello(request, timeout=1)  # timeout 單位:秒
        print(f"ret:{res.ret}, data:{res.data}")
    except grpc.RpcError as rpc_error:
        print("grpc.RpcError", rpc_error.details())
    except Exception as es:
        print(es)
    finally:
        sys.exit(-1)
 
 
if __name__ == '__main__':
    client("127.0.0.1", 8000)

運行結(jié)果:

grpc.RpcError Deadline Exceeded

5. grpc之大文件之流stream傳輸

1)在example.proto重新定義傳送體

// 聲明
syntax = "proto3";
package proto;
 
// service創(chuàng)建
service HelloService{
  rpc Hello(Request) returns (Response) {}  // 單單傳送消息
  rpc ClientTOServer(stream UpFileRequest) returns (Response) {}  // 流式上傳文件
  rpc ServerTOClient(Request) returns (stream UpFileRequest) {}  // 流式下載文件
}
 
// 請求參數(shù)消息體 1、2是指參數(shù)順序
message Request {
  string data = 1;
}
 
// 返回參數(shù)消息體
message Response {
  int32 ret = 1;    //返回碼
  string data = 2;
}
 
message UpFileRequest {
  string filename = 1;
  int64 sendsize = 2;
  int64 totalsize = 3;
  bytes data = 4;
}
 
 
//python -m grpc_tools.protoc -I ./ --python_out=./ --grpc_python_out=./ ./example.proto

2)在虛擬環(huán)境里使用命令生成py文件,參考2. 2)

3)編輯client.py 和 server.py

# server.py
import os
import time
import grpc
from concurrent import futures
from proto import example_pb2_grpc, example_pb2
 
 
class ServiceBack(example_pb2_grpc.HelloServiceServicer):
    """接口的具體功能實現(xiàn)"""
 
    def Hello(self, request, context):
        """hello"""
        data = request.data
        print(data)
        time.sleep(2)
        ret_data = "Response:" + data
        return example_pb2.Response(ret=0, data=ret_data)
 
    def ClientTOServer(self, request_iterator, context):
        """上傳文件"""
        data = bytearray()
        for UpFileRequest in request_iterator:
            file_name = UpFileRequest.filename
            file_size = UpFileRequest.totalsize
            file_data = UpFileRequest.data
            print(f"文件名稱:{file_name}, 文件總長度:{file_size}")
            data.extend(file_data)  # 拼接兩個bytes
            print(f"已接收長度:{len(data)}")
        if len(data) == file_size:
            with open("242_copy.mp3", "wb") as fw:
                fw.write(data)
            print(f"{file_name=} 下載完成")
            (ret, res) = (0, file_name)
        else:
            print(f"{file_name=} 下載失敗")
            (ret, res) = (-1, file_name)
        return example_pb2.Response(ret=ret, data=res)
 
    def ServerTOClient(self, request, context):
        """下載文件"""
        fp = request.data
        print(f"下載文件:{fp=}")
        # 獲取文件名和文件大小
        file_name = os.path.basename(fp)
        file_size = os.path.getsize(fp)  # 獲取文件大小
        # 發(fā)送文件內(nèi)容
        part_size = 1024 * 1024  # 每次讀取1MB數(shù)據(jù)
        count = 1
 
        with open(fp, "rb") as fr:
            while True:
                try:
                    if count == 1:
                        count += 1
                        yield example_pb2.UpFileRequest(filename=file_name, totalsize=file_size, sendsize=0, data=b"")
                    else:
                        context = fr.read(part_size)
                        if context:
                            yield example_pb2.UpFileRequest(filename=file_name, totalsize=file_size,
                                                            sendsize=len(context),
                                                            data=context)
                        else:
                            print(f"發(fā)送完畢")
                            return 0
                except Exception as es:
                    print(es)
 
 
def server(ip: str, port: int) -> None:
    # 數(shù)據(jù)傳輸大小配置
    max_message_length = 1024 * 1024 * 1024  # 1G
    options = [('grpc.max_send_message_length', max_message_length),
               ('grpc.max_receive_message_length', max_message_length),
               ('grpc.enable_retries', 1),
               ]
    server = grpc.server(futures.ThreadPoolExecutor(max_workers=10), options=options)  # ??為10的線程池
    ai_servicer = ServiceBack()
    example_pb2_grpc.add_HelloServiceServicer_to_server(ai_servicer, server)
    server.add_insecure_port(f"{ip}:{port}")  
    server.start()
    try:
        print(f"server is started! ip:{ip} port:{str(port)}")
        while True:
            time.sleep(60 * 60)
    except Exception as es:
        print(es)
        server.stop(0)
 
 
if __name__ == '__main__':
    server("127.0.0.1", 8000)
# client.py
import os
import sys
import grpc
from proto import example_pb2_grpc, example_pb2
 
 
def send_stream_data(fp: str):
    """迭代器發(fā)送大文件"""
    # 獲取文件名和文件大小
    file_name = os.path.basename(fp)
    file_size = os.path.getsize(fp)  # 獲取文件大小
    # 發(fā)送文件內(nèi)容
    part_size = 1024 * 1024  # 每次讀取1MB數(shù)據(jù)
    count = 1
 
    with open(fp, "rb") as fr:
        while True:
            try:
                if count == 1:
                    count += 1
                    yield example_pb2.UpFileRequest(filename=file_name, totalsize=file_size, sendsize=0, data=b"")
                else:
                    context = fr.read(part_size)
                    if context:
                        yield example_pb2.UpFileRequest(filename=file_name, totalsize=file_size, sendsize=len(context),
                                                        data=context)
                    else:
                        print(f"發(fā)送完畢")
                        return 0
            except Exception as es:
                print(es)
 
 
def client(ip: str, port: int) -> None:
    # 數(shù)據(jù)傳輸大小配置
    max_message_length = 1024 * 1024 * 1024  # 1G
    options = [('grpc.max_send_message_length', max_message_length),
               ('grpc.max_receive_message_length', max_message_length),
               ('grpc.enable_retries', 1),
               ]
    target = str(ip) + ":" + str(port)
    channel = grpc.insecure_channel(target, options=options)  # 連接rpc服務器
    cli = example_pb2_grpc.HelloServiceStub(channel)  # 創(chuàng)建Stub
    try:
        data = "hello 123"
        request = example_pb2.Request(data=data)
        res = cli.Hello(request, timeout=1)  # timeout 單位:秒
        print(f"ret:{res.ret}, data:{res.data}")
    except grpc.RpcError as rpc_error:
        print("grpc.RpcError", rpc_error.details())
    except Exception as es:
        print(es)
    finally:
        sys.exit(-1)
 
 
def client_to_server(ip: str, port: int, fp: str):
    """
    流式上傳數(shù)據(jù)。
    """
    # 數(shù)據(jù)傳輸大小配置
    max_message_length = 1024 * 1024 * 1024  # 1G
    options = [('grpc.max_send_message_length', max_message_length),
               ('grpc.max_receive_message_length', max_message_length),
               ('grpc.enable_retries', 1),
               ]
    target = str(ip) + ":" + str(port)
    channel = grpc.insecure_channel(target, options=options)  # 連接rpc服務器
    cli = example_pb2_grpc.HelloServiceStub(channel)  # 創(chuàng)建Stub
    try:
        request = send_stream_data(fp=fp)
        res = cli.ClientTOServer(request, timeout=600)  # timeout 單位:秒
        print(f"ret:{res.ret}, data:{res.data}")
    except grpc.RpcError as rpc_error:
        print("grpc.RpcError", rpc_error.details())
    except Exception as es:
        print(es)
    finally:
        sys.exit(-1)
 
 
def server_to_client(ip: str, port: int, fp: str):
    """
    流式上傳數(shù)據(jù)。
    """
    # 數(shù)據(jù)傳輸大小配置
    max_message_length = 1024 * 1024 * 1024  # 1G
    options = [('grpc.max_send_message_length', max_message_length),
               ('grpc.max_receive_message_length', max_message_length),
               ('grpc.enable_retries', 1),
               ]
    target = str(ip) + ":" + str(port)
    channel = grpc.insecure_channel(target, options=options)  # 連接rpc服務器
    cli = example_pb2_grpc.HelloServiceStub(channel)  # 創(chuàng)建Stub
    try:
        data = bytearray()
        request = example_pb2.Request(data=fp)
        filename = ""
        for res in cli.ServerTOClient(request, timeout=300):
            filename = res.filename
            total_size = res.totalsize
            data += res.data
        if total_size == len(data):
            with open("242_1.mp3", "wb") as fw:
                fw.write(data)
            print(f"{filename=} : {total_size=} 下載完成!")
        else:
            print(f"{filename=} 下載失?。?)
    except grpc.RpcError as rpc_error:
        print("grpc.RpcError", rpc_error.details())
    except Exception as es:
        print(es)
    finally:
        sys.exit(-1)
 
 
if __name__ == '__main__':
    # client("127.0.0.1", 8000)
    # client_to_server("127.0.0.1", 8000, "242.mp3")
    server_to_client("127.0.0.1", 8000, "242.mp3")

6. grpc之大文件之流async異步傳輸

# server.py
import os
import time
import grpc
from concurrent import futures
from proto import example_pb2_grpc, example_pb2
import asyncio
 
 
class ServiceBack(example_pb2_grpc.HelloServiceServicer):
    """接口的具體功能實現(xiàn)"""
 
    def Hello(self, request, context):
        """hello"""
        data = request.data
        print(data)
        time.sleep(2)
        ret_data = "Response:" + data
        return example_pb2.Response(ret=0, data=ret_data)
 
    def ClientTOServer(self, request_iterator, context):
        """上傳文件"""
        data = bytearray()
        for UpFileRequest in request_iterator:
            file_name = UpFileRequest.filename
            file_size = UpFileRequest.totalsize
            file_data = UpFileRequest.data
            print(f"文件名稱:{file_name}, 文件總長度:{file_size}")
            data.extend(file_data)  # 拼接兩個bytes
            print(f"已接收長度:{len(data)}")
        if len(data) == file_size:
            with open("242_copy.mp3", "wb") as fw:
                fw.write(data)
            print(f"{file_name=} 下載完成")
            (ret, res) = (0, file_name)
        else:
            print(f"{file_name=} 下載失敗")
            (ret, res) = (-1, file_name)
        return example_pb2.Response(ret=ret, data=res)
 
    def ServerTOClient(self, request, context):
        """下載文件"""
        fp = request.data
        print(f"下載文件:{fp=}")
        # 獲取文件名和文件大小
        file_name = os.path.basename(fp)
        file_size = os.path.getsize(fp)  # 獲取文件大小
        # 發(fā)送文件內(nèi)容
        part_size = 1024 * 1024  # 每次讀取1MB數(shù)據(jù)
        count = 1
 
        with open(fp, "rb") as fr:
            while True:
                try:
                    if count == 1:
                        count += 1
                        yield example_pb2.UpFileRequest(filename=file_name, totalsize=file_size, sendsize=0, data=b"")
                    else:
                        context = fr.read(part_size)
                        if context:
                            yield example_pb2.UpFileRequest(filename=file_name, totalsize=file_size,
                                                            sendsize=len(context),
                                                            data=context)
                        else:
                            print(f"發(fā)送完畢")
                            return 0
                except Exception as es:
                    print(es)
 
 
async def server(ip: str, port: int) -> None:
    # 數(shù)據(jù)傳輸大小配置
    max_message_length = 1024 * 1024 * 1024  # 1G
    options = [('grpc.max_send_message_length', max_message_length),
               ('grpc.max_receive_message_length', max_message_length),
               ('grpc.enable_retries', 1),
               ]
    server = grpc.aio.server(futures.ThreadPoolExecutor(max_workers=10), options=options)  # ??為10的線程池
    ai_servicer = ServiceBack()
    example_pb2_grpc.add_HelloServiceServicer_to_server(ai_servicer, server)
    server.add_insecure_port(f"{ip}:{port}")
    await server.start()
    try:
        print(f"server is started! ip:{ip} port:{str(port)}")
        await server.wait_for_termination()
    except Exception as es:
        print(es)
        await server.stop(None)
 
 
if __name__ == '__main__':
    loop = asyncio.get_event_loop()
    loop.run_until_complete(asyncio.wait([server("127.0.0.1", 8000)]))
    loop.close()
# client.py
import os
import sys
import grpc
from proto import example_pb2_grpc, example_pb2
import asyncio
 
 
def send_stream_data(fp: str):
    """迭代器發(fā)送大文件"""
    # 獲取文件名和文件大小
    file_name = os.path.basename(fp)
    file_size = os.path.getsize(fp)  # 獲取文件大小
    # 發(fā)送文件內(nèi)容
    part_size = 1024 * 1024  # 每次讀取1MB數(shù)據(jù)
    count = 1
 
    with open(fp, "rb") as fr:
        while True:
            try:
                if count == 1:
                    count += 1
                    yield example_pb2.UpFileRequest(filename=file_name, totalsize=file_size, sendsize=0, data=b"")
                else:
                    context = fr.read(part_size)
                    if context:
                        yield example_pb2.UpFileRequest(filename=file_name, totalsize=file_size, sendsize=len(context),
                                                        data=context)
                    else:
                        print(f"發(fā)送完畢")
                        return 0
            except Exception as es:
                print(es)
 
 
async def client(ip: str, port: int) -> None:
    # 數(shù)據(jù)傳輸大小配置
    max_message_length = 1024 * 1024 * 1024  # 1G
    options = [('grpc.max_send_message_length', max_message_length),
               ('grpc.max_receive_message_length', max_message_length),
               ('grpc.enable_retries', 1),
               ]
    target = str(ip) + ":" + str(port)
    async with grpc.aio.insecure_channel(target, options=options) as channel:  # 連接rpc服務器
        cli = example_pb2_grpc.HelloServiceStub(channel)  # 創(chuàng)建Stub
        try:
            data = "hello 123"
            request = example_pb2.Request(data=data)
            res = await cli.Hello(request, timeout=3)  # timeout 單位:秒
            print(f"ret:{res.ret}, data:{res.data}")
        except grpc.RpcError as rpc_error:
            print("grpc.RpcError", rpc_error.details())
        except Exception as es:
            print(es)
        finally:
            sys.exit(-1)
 
 
async def client_to_server(ip: str, port: int, fp: str):
    """
    流式上傳數(shù)據(jù)。
    """
    # 數(shù)據(jù)傳輸大小配置
    max_message_length = 1024 * 1024 * 1024  # 1G
    options = [('grpc.max_send_message_length', max_message_length),
               ('grpc.max_receive_message_length', max_message_length),
               ('grpc.enable_retries', 1),
               ]
    target = str(ip) + ":" + str(port)
    async with grpc.aio.insecure_channel(target, options=options) as channel:  # 連接rpc服務器
        cli = example_pb2_grpc.HelloServiceStub(channel)  # 創(chuàng)建Stub
        try:
            request = send_stream_data(fp=fp)
            res = await cli.ClientTOServer(request, timeout=600)  # timeout 單位:秒
            print(f"ret:{res.ret}, data:{res.data}")
        except grpc.RpcError as rpc_error:
            print("grpc.RpcError", rpc_error.details())
        except Exception as es:
            print(es)
        finally:
            sys.exit(-1)
 
 
def server_to_client(ip: str, port: int, fp: str):
    """
    流式上傳數(shù)據(jù)。
    """
    # 數(shù)據(jù)傳輸大小配置
    max_message_length = 1024 * 1024 * 1024  # 1G
    options = [('grpc.max_send_message_length', max_message_length),
               ('grpc.max_receive_message_length', max_message_length),
               ('grpc.enable_retries', 1),
               ]
    target = str(ip) + ":" + str(port)
    channel = grpc.insecure_channel(target, options=options)  # 連接rpc服務器
    cli = example_pb2_grpc.HelloServiceStub(channel)  # 創(chuàng)建Stub
    try:
        data = bytearray()
        request = example_pb2.Request(data=fp)
        filename = ""
        for res in cli.ServerTOClient(request, timeout=300):
            filename = res.filename
            total_size = res.totalsize
            data += res.data
        if total_size == len(data):
            with open("242_1.mp3", "wb") as fw:
                fw.write(data)
            print(f"{filename=} : {total_size=} 下載完成!")
        else:
            print(f"{filename=} 下載失?。?)
    except grpc.RpcError as rpc_error:
        print("grpc.RpcError", rpc_error.details())
    except Exception as es:
        print(es)
    finally:
        sys.exit(-1)
 
 
if __name__ == '__main__':
    # asyncio.run(client("127.0.0.1", 8000))
    asyncio.run(client_to_server("127.0.0.1", 8000, "242.mp3"))
    # server_to_client("127.0.0.1", 8000, "242.mp3")

結(jié)論: 在本地測了一下不加async和加async的文件上傳傳送, async還慢點,嘿嘿嘿。

到此這篇關于利用Python語言的grpc實現(xiàn)消息傳送詳解的文章就介紹到這了,更多相關Python grpc消息傳送內(nèi)容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關文章希望大家以后多多支持腳本之家!

相關文章

  • python運行其他程序的實現(xiàn)方法

    python運行其他程序的實現(xiàn)方法

    這篇文章主要介紹了python運行其他程序的實現(xiàn)方法的相關資料,需要的朋友可以參考下
    2017-07-07
  • 基于Python實現(xiàn)DIT-FFT算法

    基于Python實現(xiàn)DIT-FFT算法

    FFT(Fast Fourier Transformation)是離散傅氏變換(DFT)的快速算法。即為快速傅氏變換。本文將用Python語言實現(xiàn)DIT-FFT算法,感興趣的可以了解一下
    2022-10-10
  • 模型訓練時GPU利用率太低的原因及解決

    模型訓練時GPU利用率太低的原因及解決

    這篇文章主要介紹了模型訓練時GPU利用率太低的原因及解決方案,具有很好的參考價值,希望對大家有所幫助。如有錯誤或未考慮完全的地方,望不吝賜教
    2023-02-02
  • Python中字符串String的基本內(nèi)置函數(shù)與過濾字符模塊函數(shù)的基本用法

    Python中字符串String的基本內(nèi)置函數(shù)與過濾字符模塊函數(shù)的基本用法

    這篇文章主要介紹了Python中字符串String的基本內(nèi)置函數(shù)與過濾字符模塊函數(shù)的基本用法 ,本文給大家介紹的非常詳細,具有一定的參考借鑒價值,需要的朋友可以參考下
    2019-05-05
  • Django中使用CORS實現(xiàn)跨域請求過程解析

    Django中使用CORS實現(xiàn)跨域請求過程解析

    這篇文章主要介紹了Django中使用CORS實現(xiàn)跨域請求過程解析,文中通過示例代碼介紹的非常詳細,對大家的學習或者工作具有一定的參考學習價值,需要的朋友可以參考下
    2019-08-08
  • Flask??request?對象介紹

    Flask??request?對象介紹

    本文介紹?Flask??request?對象,一個完整的?HTTP?請求,包括客戶端向服務端發(fā)送的Request?請求和服務器端發(fā)送?Response?響應.為了能方便訪問獲取請求及響應報文信息,Flask?框架提供了一些內(nèi)建對象,下面就來說一下?Flask?針對請求提供內(nèi)建對象reques,需要的朋友可以參考一下
    2021-11-11
  • 詳解Python 合并字典

    詳解Python 合并字典

    這篇文章主要為大家介紹了Python的合并字典,具有一定的參考價值,感興趣的小伙伴們可以參考一下,希望能夠給你帶來幫助
    2021-12-12
  • Python整數(shù)與Numpy數(shù)據(jù)溢出問題解決

    Python整數(shù)與Numpy數(shù)據(jù)溢出問題解決

    這篇文章主要介紹了Python 的整數(shù)與 Numpy 的數(shù)據(jù)溢出,文中通過示例代碼介紹的非常詳細,對大家的學習或者工作具有一定的參考學習價值,需要的朋友可以參考下
    2019-09-09
  • Django框架登錄加上驗證碼校驗實現(xiàn)驗證功能示例

    Django框架登錄加上驗證碼校驗實現(xiàn)驗證功能示例

    這篇文章主要介紹了Django框架登錄加上驗證碼校驗實現(xiàn)驗證功能,結(jié)合實例形式分析了Django框架基于Pillow模塊的圖形驗證碼生成與使用相關操作技巧,需要的朋友可以參考下
    2019-05-05
  • 推薦技術人員一款Python開源庫(造數(shù)據(jù)神器)

    推薦技術人員一款Python開源庫(造數(shù)據(jù)神器)

    今天小編給大家推薦一款Python開源庫,技術人必備的造數(shù)據(jù)神器!非常不錯,對大家的學習或工作具有一定的參考借鑒價值,需要的朋友參考下吧
    2020-07-07

最新評論

福海县| 昆明市| 隆回县| 肇源县| 玉山县| 沭阳县| 泰来县| 呼伦贝尔市| 横山县| 娱乐| 香港 | 大竹县| 健康| 临泽县| 盐山县| 富川| 同心县| 马公市| 克拉玛依市| 大洼县| 韩城市| 庆云县| 孟州市| 桐庐县| 修文县| 日照市| 天台县| 仪陇县| 肇州县| 阿拉尔市| 宁津县| 金昌市| 建德市| 南江县| 夏津县| 通海县| 辽中县| 滕州市| 中江县| 崇信县| 怀来县|