python3+PyQt5 創(chuàng)建多線程網(wǎng)絡(luò)應(yīng)用-TCP客戶端和TCP服務(wù)器實例
更新時間：2019年06月17日 11:22:55 作者：basisworker
今天小編就為大家分享一篇python3+PyQt5 創(chuàng)建多線程網(wǎng)絡(luò)應(yīng)用-TCP客戶端和TCP服務(wù)器實例，具有很好的參考價值，希望對大家有所幫助。一起跟隨小編過來看看吧
本文在上文的基礎(chǔ)上重新實現(xiàn)支持多線程的服務(wù)器。
以下為TCP客戶端的程序代碼：
#!/usr/bin/env python3
import sys
from PyQt5.QtCore import (QByteArray, QDataStream, QDate, QIODevice,
QRegExp, Qt)
from PyQt5.QtWidgets import (QApplication, QDateEdit, QFrame, QGridLayout,
QHBoxLayout, QLabel, QLineEdit, QPushButton,
QWidget)
from PyQt5.QtGui import QRegExpValidator
from PyQt5.QtNetwork import (QTcpSocket,)
MAC = True
try:
from PyQt5.QtGui import qt_mac_set_native_menubar
except ImportError:
MAC = False
PORT = 9407
SIZEOF_UINT16 = 2
class BuildingServicesClient(QWidget):
def __init__(self, parent=None):
super(BuildingServicesClient, self).__init__(parent)
self.socket = QTcpSocket()
self.nextBlockSize = 0
self.request = None
roomLabel = QLabel("&Room")
self.roomEdit = QLineEdit()
roomLabel.setBuddy(self.roomEdit)
regex = QRegExp(r"[0-9](?:0[1-9]|[12][0-9]|3[0-4])")
self.roomEdit.setValidator(QRegExpValidator(regex, self))
self.roomEdit.setAlignment(Qt.AlignRight|Qt.AlignVCenter)
dateLabel = QLabel("&Date")
self.dateEdit = QDateEdit()
dateLabel.setBuddy(self.dateEdit)
self.dateEdit.setAlignment(Qt.AlignRight|Qt.AlignVCenter)
self.dateEdit.setDate(QDate.currentDate().addDays(1))
self.dateEdit.setDisplayFormat("yyyy-MM-dd")
responseLabel = QLabel("Response")
self.responseLabel = QLabel()
self.responseLabel.setFrameStyle(QFrame.StyledPanel|QFrame.Sunken)
self.bookButton = QPushButton("&Book")
self.bookButton.setEnabled(False)
self.unBookButton = QPushButton("&Unbook")
self.unBookButton.setEnabled(False)
quitButton = QPushButton("&Quit")
if not MAC:
self.bookButton.setFocusPolicy(Qt.NoFocus)
self.unBookButton.setFocusPolicy(Qt.NoFocus)
buttonLayout = QHBoxLayout()
buttonLayout.addWidget(self.bookButton)
buttonLayout.addWidget(self.unBookButton)
buttonLayout.addStretch()
buttonLayout.addWidget(quitButton)
layout = QGridLayout()
layout.addWidget(roomLabel, 0, 0)
layout.addWidget(self.roomEdit, 0, 1)
layout.addWidget(dateLabel, 0, 2)
layout.addWidget(self.dateEdit, 0, 3)
layout.addWidget(responseLabel, 1, 0)
layout.addWidget(self.responseLabel, 1, 1, 1, 3)
layout.addLayout(buttonLayout, 2, 1, 1, 4)
self.setLayout(layout)
self.socket.connected.connect(self.sendRequest)
self.socket.readyRead.connect(self.readResponse)
self.socket.disconnected.connect(self.serverHasStopped)
#self.connect(self.socket,
# SIGNAL("error(QAbstractSocket::SocketError)"),
# self.serverHasError)
self.socket.error.connect(self.serverHasError)
self.roomEdit.textEdited.connect(self.updateUi)
self.dateEdit.dateChanged.connect(self.updateUi)
self.bookButton.clicked.connect(self.book)
self.unBookButton.clicked.connect(self.unBook)
quitButton.clicked.connect(self.close)
self.setWindowTitle("Building Services")
def updateUi(self):
enabled = False
if (self.roomEdit.text() and
self.dateEdit.date() > QDate.currentDate()):
enabled = True
if self.request is not None:
enabled = False
self.bookButton.setEnabled(enabled)
self.unBookButton.setEnabled(enabled)
def closeEvent(self, event):
self.socket.close()
event.accept()
def book(self):
self.issueRequest("BOOK", self.roomEdit.text(),
self.dateEdit.date())
def unBook(self):
self.issueRequest("UNBOOK", self.roomEdit.text(),
self.dateEdit.date())
def issueRequest(self, action, room, date):
self.request = QByteArray()
stream = QDataStream(self.request, QIODevice.WriteOnly)
stream.setVersion(QDataStream.Qt_5_7)
stream.writeUInt16(0)
stream.writeQString(action)
stream.writeQString(room)
stream << date
stream.device().seek(0)
stream.writeUInt16(self.request.size() - SIZEOF_UINT16)#overwrite seek(0)
self.updateUi()
if self.socket.isOpen():
self.socket.close()
self.responseLabel.setText("Connecting to server...")
self.socket.connectToHost("localhost", PORT)
def sendRequest(self):
self.responseLabel.setText("Sending request...")
self.nextBlockSize = 0
self.socket.write(self.request)
self.request = None
def readResponse(self):
stream = QDataStream(self.socket)
stream.setVersion(QDataStream.Qt_5_7)
while True:
if self.nextBlockSize == 0:
if self.socket.bytesAvailable() < SIZEOF_UINT16:
break
self.nextBlockSize = stream.readUInt16()
if self.socket.bytesAvailable() < self.nextBlockSize:
break
action = ""
room = ""
date = QDate()
#stream >> action >> room
action=stream.readQString()
room=stream.readQString()
if action != "ERROR":
stream >> date
if action == "ERROR":
msg = "Error: {0}".format(room)
elif action == "BOOK":
msg = "Booked room {0} for {1}".format(room,date.toString(Qt.ISODate))
elif action == "UNBOOK":
msg = "Unbooked room {0} for {1}".format(room,date.toString(Qt.ISODate))
self.responseLabel.setText(msg)
self.updateUi()
self.nextBlockSize = 0
def serverHasStopped(self):
self.responseLabel.setText(
"Error: Connection closed by server")
self.socket.close()
def serverHasError(self, error):
self.responseLabel.setText("Error: {0}".format(self.socket.errorString()))
self.socket.close()
app = QApplication(sys.argv)
form = BuildingServicesClient()
form.show()
app.exec_()
以下為TCP服務(wù)端的程序代碼：
#!/usr/bin/env python3
import bisect
import collections
import sys
from PyQt5.QtCore import (QByteArray, QDataStream, QDate, QReadWriteLock, QThread,QIODevice, Qt)
from PyQt5.QtWidgets import (QApplication, QMessageBox, QPushButton)
from PyQt5.QtNetwork import (QAbstractSocket,QHostAddress, QTcpServer, QTcpSocket)
PORT = 9407
SIZEOF_UINT16 = 2
MAX_BOOKINGS_PER_DAY = 5
# Key = date, value = list of room IDs
Bookings = collections.defaultdict(list)
def printBookings():
for key in sorted(Bookings):
print(key, Bookings[key])
print()
class Thread(QThread):
lock = QReadWriteLock()
def __init__(self, socketId, parent):
super(Thread, self).__init__(parent)
self.socketId = socketId
def run(self):
socket = QTcpSocket()
if not socket.setSocketDescriptor(self.socketId):
#self.emit(SIGNAL("error(int)"), socket.error())
self.error.connect(socket.error)
return
while socket.state() == QAbstractSocket.ConnectedState:
nextBlockSize = 0
stream = QDataStream(socket)
stream.setVersion(QDataStream.Qt_5_7)
if (socket.waitForReadyRead() and
socket.bytesAvailable() >= SIZEOF_UINT16):
nextBlockSize = stream.readUInt16()
else:
self.sendError(socket, "Cannot read client request")
return
if socket.bytesAvailable() < nextBlockSize:
if (not socket.waitForReadyRead(60000) or
socket.bytesAvailable() < nextBlockSize):
self.sendError(socket, "Cannot read client data")
return
action = ""
room = ""
date = QDate()
action=stream.readQString()
if action in ("BOOK", "UNBOOK"):
room=stream.readQString()
stream >> date
try:
Thread.lock.lockForRead()
bookings = Bookings.get(date.toPyDate())
finally:
Thread.lock.unlock()
uroom = str(room)
if action == "BOOK":
newlist = False
try:
Thread.lock.lockForRead()
if bookings is None:
newlist = True
finally:
Thread.lock.unlock()
if newlist:
try:
Thread.lock.lockForWrite()
bookings = Bookings[date.toPyDate()]
finally:
Thread.lock.unlock()
error = None
insert = False
try:
Thread.lock.lockForRead()
if len(bookings) < MAX_BOOKINGS_PER_DAY:
if uroom in bookings:
error = "Cannot accept duplicate booking"
else:
insert = True
else:
error = "{0} is fully booked".format(date.toString(Qt.ISODate))
finally:
Thread.lock.unlock()
if insert:
try:
Thread.lock.lockForWrite()
bisect.insort(bookings, uroom)
finally:
Thread.lock.unlock()
self.sendReply(socket, action, room, date)
else:
self.sendError(socket, error)
elif action == "UNBOOK":
error = None
remove = False
try:
Thread.lock.lockForRead()
if bookings is None or uroom not in bookings:
error = "Cannot unbook nonexistent booking"
else:
remove = True
finally:
Thread.lock.unlock()
if remove:
try:
Thread.lock.lockForWrite()
bookings.remove(uroom)
finally:
Thread.lock.unlock()
self.sendReply(socket, action, room, date)
else:
self.sendError(socket, error)
else:
self.sendError(socket, "Unrecognized request")
socket.waitForDisconnected()
try:
Thread.lock.lockForRead()
printBookings()
finally:
Thread.lock.unlock()
def sendError(self, socket, msg):
reply = QByteArray()
stream = QDataStream(reply, QIODevice.WriteOnly)
stream.setVersion(QDataStream.Qt_5_7)
stream.writeUInt16(0)
stream.writeQString("ERROR")
stream.writeQString(msg)
stream.device().seek(0)
stream.writeUInt16(reply.size() - SIZEOF_UINT16)
socket.write(reply)
def sendReply(self, socket, action, room, date):
reply = QByteArray()
stream = QDataStream(reply, QIODevice.WriteOnly)
stream.setVersion(QDataStream.Qt_5_7)
stream.writeUInt16(0)
stream.writeQString(action)
stream.writeQString(room)
stream<<date
stream.device().seek(0)
stream.writeUInt16(reply.size() - SIZEOF_UINT16)
socket.write(reply)
class TcpServer(QTcpServer):
def __init__(self, parent=None):
super(TcpServer, self).__init__(parent)
def incomingConnection(self, socketId):
thread = Thread(socketId, self)
#self.connect(thread, SIGNAL("finished()"),
# thread, SLOT("deleteLater()"))
thread.finished.connect(thread.deleteLater)
thread.start()
class BuildingServicesDlg(QPushButton):
def __init__(self, parent=None):
super(BuildingServicesDlg, self).__init__(
"&Close Server", parent)
self.setWindowFlags(Qt.WindowStaysOnTopHint)
self.loadBookings()
self.tcpServer = TcpServer(self)
if not self.tcpServer.listen(QHostAddress("0.0.0.0"), PORT):
QMessageBox.critical(self, "Building Services Server","Failed to start server: {0}".format(self.tcpServer.errorString()))
self.close()
return
self.clicked.connect(self.close)
font = self.font()
font.setPointSize(24)
self.setFont(font)
self.setWindowTitle("Building Services Server")
def loadBookings(self):
# Generate fake data
import random
today = QDate.currentDate()
for i in range(10):
date = today.addDays(random.randint(7, 60))
for j in range(random.randint(1, MAX_BOOKINGS_PER_DAY)):
# Rooms are 001..534 excl. 100, 200, ..., 500
floor = random.randint(0, 5)
room = random.randint(1, 34)
bookings = Bookings[date.toPyDate()]
if len(bookings) >= MAX_BOOKINGS_PER_DAY:
continue
bisect.insort(bookings, "{0:1d}{1:02d}".format(
floor, room))
printBookings()
app = QApplication(sys.argv)
form = BuildingServicesDlg()
form.show()
form.move(0, 0)
app.exec_()
以上這篇python3+PyQt5 創(chuàng)建多線程網(wǎng)絡(luò)應(yīng)用-TCP客戶端和TCP服務(wù)器實例就是小編分享給大家的全部內(nèi)容了，希望能給大家一個參考，也希望大家多多支持腳本之家。
相關(guān)文章
在本篇文章里小編給大家整理的是一篇關(guān)于python的鏈表基礎(chǔ)知識點內(nèi)容，有興趣的朋友們可以參考學(xué)習(xí)下。2020-09-09
python3下使用cv2.imwrite存儲帶有中文路徑圖片的方法
今天小編就為大家分享一篇python3下使用cv2.imwrite存儲帶有中文路徑圖片的方法，具有很好的參考價值，希望對大家有所幫助。一起跟隨小編過來看看吧2018-05-05
Python+tkinter自定義實現(xiàn)文件選擇按鈕
這篇文章主要為大家詳細(xì)介紹了如何利用Python和tkinter自定義實現(xiàn)簡單的文件選擇按鈕和顏色選擇按鈕,有需要的小伙伴可以跟隨小編一起學(xué)習(xí)一下2023-10-10
Python面向?qū)ο蟪绦蛟O(shè)計OOP深入分析【構(gòu)造函數(shù),組合類,工具類等】
這篇文章主要介紹了Python面向?qū)ο蟪绦蛟O(shè)計OOP,較為詳細(xì)的深入分析了Python面向?qū)ο蟮臉?gòu)造函數(shù),組合類,工具類等相關(guān)概念、使用方法及操作注意事項,需要的朋友可以參考下2019-01-01
moviepy是一個非常強大的視頻處理庫,可以輕松進行視頻剪切、合并、添加特效等操作,本文我們就來使用moviepy實現(xiàn)刪除視頻的某一段并保留其他時間段吧2024-11-11
turtle庫是python的基礎(chǔ)繪圖庫，經(jīng)常被用來給孩子們介紹編程知識的方法庫，是標(biāo)準(zhǔn)庫之一，利用turtle可以制作很多復(fù)雜的繪圖。本文將為大家介紹通過turtle庫繪制的一些有趣的圖畫，感興趣的小伙伴可以學(xué)習(xí)一下2021-12-12
這篇文章主要介紹了Python實現(xiàn)的多線程http壓力測試代碼,結(jié)合實例形式分析了Python多線程操作的相關(guān)實現(xiàn)技巧,需要的朋友可以參考下2017-02-02
Python selenium實現(xiàn)斷言3種方法解析
這篇文章主要介紹了Python selenium實現(xiàn)斷言3種方法解析,文中通過示例代碼介紹的非常詳細(xì)，對大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價值,需要的朋友可以參考下2020-09-09
這篇文章主要介紹了如何在Python?中獲取單成員集合中的唯一元素,本文給大家介紹的非常詳細(xì)，對大家的學(xué)習(xí)或工作具有一定的參考借鑒價值，需要的朋友可以參考下2023-03-03
這篇文章主要為大家詳細(xì)介紹了Python實現(xiàn)LRU緩存置換算法，文中示例代碼介紹的非常詳細(xì)，具有一定的參考價值，感興趣的小伙伴們可以參考一下2022-05-05