python3+PyQt5 创建多线程网络应用-TCP客户端和TCP服务器实例
作者:basisworker 发布时间:2021-01-14 10:20:29
标签:python3,PyQt5,多线程,TCP
本文在上文的基础上重新实现支持多线程的服务器。
以下为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服务端的程序代码:
#!/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_()
来源:https://blog.csdn.net/xiaoyangyang20/article/details/71375442
0
投稿
猜你喜欢
- 本文实例讲述了php解析字符串里所有URL地址的方法。分享给大家供大家参考。具体如下:<?php// $html = the html
- Python——re模块 简介定义:re模块称为正则表达式;作用:创建一个"规则表达式",用于验证和查找符合规
- 本文实例讲述了python实现集中式的病毒扫描功能。分享给大家供大家参考,具体如下:一 点睛本次实践实现了一个集中式的病毒扫描管理
- 随着WEB标准在国内的不断普及,结构表现行为分离、模块化、语义化、优雅退化等概念也成为考核一名前端人员对WEB标准理解的重要条目,其中,由于
- 前言上一篇介绍了服务端流式RPC,客户端发送请求到服务器,拿到一个流去读取返回的消息序列。 客户端读取返回的流的数据。本篇将介绍客户端流式R
- 该域名查询系统写的很简单,只是实现了功能使用XmlHttp来获取远程查询结果,实际上就是小偷程序!相关推荐:域名注册情况查询/
- 需求背景:进行分值计算。如下图,如果只是一两个还好说,写写判断,但是如果有几十个,几百个,会不会惨不忍睹。而且,下面的还是三种情况。例如:解
- 说在前头最近在做毕设,题目是道路拥堵预测系统,学长建议我使用SVM算法进行预测,但是在此之前需要把Excel中的数据进行二次处理,原始数据不
- 经常在工作中会使用到XPath的相关知识,但每次总会在一些关键的地方不记得或不太清楚,所以免不了每次总要查一些零碎的知识,感觉即很烦又浪费时
- 这篇博客将介绍如何使用OpenCV和深度学习应用全面嵌套的边缘检测。并将对图像和视频流应用全面嵌套边缘检测,然后将结果与OpenCV的标准C
- 本文实例为大家分享了python生成圆形图片的具体代码,供大家参考,具体内容如下# -*- coding: utf-8 -*- "
- 有时我们需要较为实时的查看服务器上的网卡流量,这里我写了两个小脚本,一个用shell(先写的,一次只能查看一个网卡),另一个用Python(
- GetRepeatTimes(TheChar,TheString) 得到一个字符串在另一个字符串当中出现几次的函数(新)如:response
- 由于存在函数内部不能访问全局作用的,所以就需要一种可以引入上一级作用域的语法结构,可以通过use使用函数声明时所在作用域的变量的值。php的
- 数据共享是数据库最基本的特征之一。但是数据共享虽然为员工带来了便利,但也产生了一些负面作用。例如因用户并发存取而导致的对数据一致性的破坏、由
- easy_thumbnails:A powerful, yet easy to implement thumbnailing applica
- 阅读上一章:chapter 5 表单Chapter 6 <strong>,<em>与其他短语元素在引言和前面的章节中
- 本文实例讲述了Django框架静态文件处理、中间件、上传文件操作。分享给大家供大家参考,具体如下:Django静态文件处理、中间件、上传文件
- 我看blog里,还有很多地方都引用过我写的这个类,转了不少,但自己一直也没发表过,这次正式发表一下。在蓝色理想中有人不懂怎么用,我在baid
- 关于 游标 if,for 的例子 create or replace procedure peace_if is cursor var_c