利用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.proto2）在虛擬環(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ù)瀏覽下面的相關文章希望大家以后多多支持腳本之家！
相關文章
Python3.5編程實現(xiàn)修改IIS WEB.CONFIG的方法示例
這篇文章主要介紹了Python3.5編程實現(xiàn)修改IIS WEB.CONFIG的方法,涉及Python針對xml格式文件的讀寫以及節(jié)點操作相關技巧,需要的朋友可以參考下2017-08-08
Python操作SQLite/MySQL/LMDB數(shù)據(jù)庫的方法
這篇文章主要介紹了Python操作SQLite/MySQL/LMDB數(shù)據(jù)庫的方法，本文給大家介紹的非常詳細，具有一定的參考借鑒價值,需要的朋友可以參考下2019-11-11
python結(jié)合多線程爬取英雄聯(lián)盟皮膚(原理分析)
多線程是為了同步完成多項任務，不是為了提高運行效率，而是為了提高資源使用效率來提高系統(tǒng)的效率。這篇文章主要介紹了python爬取英雄聯(lián)盟皮膚結(jié)合多線程的方法,需要的朋友可以參考下2021-05-05
下面小編就為大家分享一篇python list刪除元素時要注意的坑點分享，具有很好的參考價值，希望對大家有所幫助。一起跟隨小編過來看看吧2018-04-04
每天你都可能會執(zhí)行許多重復的任務，例如閱讀 pdf、播放音樂、查看天氣、打開書簽、清理文件夾等等，使用自動化腳本，就無需手動一次又一次地完成這些任務，非常方便?？旄S小編一起試一試吧2022-01-01
Python 使用 PyQt5 開發(fā)的關機小工具分享
這篇文章主要介紹了Python 使用 PyQt5 開發(fā)的關機小工具分享，具有很好的參考價值，希望對大家有所幫助。一起跟隨小編過來看看吧2020-07-07
這篇文章主要介紹了pytorch之Resize()函數(shù)具體使用詳解，文中通過示例代碼介紹的非常詳細，對大家的學習或者工作具有一定的參考學習價值，需要的朋友們下面隨著小編來一起學習學習吧2020-02-02
python中numpy數(shù)組與list相互轉(zhuǎn)換實例方法
在本篇文章里小編給大家整理的是一篇關于python中numpy數(shù)組與list相互轉(zhuǎn)換實例方法，對此有興趣的朋友們可以學習下。2021-01-01
Pytorch使用MNIST數(shù)據(jù)集實現(xiàn)基礎GAN和DCGAN詳解
今天小編就為大家分享一篇Pytorch使用MNIST數(shù)據(jù)集實現(xiàn)基礎GAN和DCGAN詳解，具有很好的參考價值，希望對大家有所幫助。一起跟隨小編過來看看吧2020-01-01
這篇文章主要介紹了python中的__slots__使用示例,__slots__用來限制class能添加的屬性,需要的朋友可以參考下2015-02-02