您好,登錄后才能下訂單哦!
小編給大家分享一下python3+PyQt5如何創(chuàng)建多線程網(wǎng)絡(luò)應(yīng)用,相信大部分人都還不怎么了解,因此分享這篇文章給大家參考一下,希望大家閱讀完這篇文章后大有收獲,下面讓我們一起去了解一下吧!
以下為T(mén)CP客戶(hù)端的程序代碼:
#!/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_()
以下為T(mén)CP服務(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)用”這篇文章的所有內(nèi)容,感謝各位的閱讀!相信大家都有了一定的了解,希望分享的內(nèi)容對(duì)大家有所幫助,如果還想學(xué)習(xí)更多知識(shí),歡迎關(guān)注億速云行業(yè)資訊頻道!
免責(zé)聲明:本站發(fā)布的內(nèi)容(圖片、視頻和文字)以原創(chuàng)、轉(zhuǎn)載和分享為主,文章觀點(diǎn)不代表本網(wǎng)站立場(chǎng),如果涉及侵權(quán)請(qǐng)聯(lián)系站長(zhǎng)郵箱:is@yisu.com進(jìn)行舉報(bào),并提供相關(guān)證據(jù),一經(jīng)查實(shí),將立刻刪除涉嫌侵權(quán)內(nèi)容。