Python使用Websocket推送數(shù)據(jù)實(shí)例
更新時(shí)間：2023年05月11日 15:30:54 作者：冰川_
WebSocket是一種在單個(gè)TCP連接上進(jìn)行全雙工通信的協(xié)議,上述HTML和Python文件可能用于建立WebSocket服務(wù)器,實(shí)現(xiàn)客戶端與服務(wù)器之間的實(shí)時(shí)通信
websocket.html
<!DOCTYPE html>
<html>
<head>
<title>WebSocket</title>
<style>
html, body {
font: normal 1em arial, helvetica;
}
#log {
width: 400px;
height: 200px;
border: 1px solid #000000;
overflow: auto;
}
#msg {
width: 330px;
}
</style>
<script>
var socket;
function init() {
var host = "ws://172.16.10.233:5555/";
try {
socket = new WebSocket(host);
socket.onopen = function (msg) {
log("Begin Connection!");
};
socket.onmessage = function (msg) {
log(msg.data);
};
socket.onclose = function (msg) {
log("Lose Connection!");
};
}
catch (ex) {
log(ex);
}
$("msg").focus();
}
function send() {
var txt, msg;
txt = $("msg");
msg = txt.value;
if (!msg) {
alert("Message can not be empty");
return;
}
txt.value = "";
txt.focus();
try {
socket.send(msg);
} catch (ex) {
log(ex);
}
}
window.onbeforeunload = function () {
try {
socket.send('quit');
socket.close();
socket = null;
}
catch (ex) {
log(ex);
}
};
function $(id) {
return document.getElementById(id);
}
function log(msg) {
$("log").innerHTML += "<br>" + msg;
}
function onkey(event) {
if (event.keyCode == 13) {
send();
}
}
</script>
</head>
<body onload="init()">
<h3>WebSocket</h3>
<br>
<div id="log"></div>
<input id="msg" type="textbox" onkeypress="onkey(event)"/>
<button onclick="send()">發(fā)送</button>
</body>
</html>
websocket.py
#coding=utf8
import struct, socket, sys
import hashlib
import threading, random
import time
from base64 import b64encode, b64decode
from multiprocessing import Process,Queue
connectionlist = {}
conn_yl = Queue()
conn_wd = Queue()
conn_yc = Queue()
conn_yt = Queue()
conn = Queue()
conn.put(connectionlist)
def decode(data):
if not len(data):
return False
# 用數(shù)據(jù)包的第二個(gè)字節(jié)，與127作與位運(yùn)算，拿到前七位。
length = data[1] & 127
# 這七位在數(shù)據(jù)頭部分成為payload，如果payload等于126，就要再擴(kuò)展2個(gè)字節(jié)。
# 如果等于127，就要再擴(kuò)展8個(gè)字節(jié)。
# 如果小于等于125，那它就占這一個(gè)字節(jié)。
if length == 126:
extend_payload_len = data[2:4]
mask = data[4:8]
decoded = data[8:]
elif length == 127:
extend_payload_len = data[2:10]
mask = data[10:14]
decoded = data[14:]
else:
extend_payload_len = None
mask = data[2:6]
decoded = data[6:]
byte_list = bytearray()
print(mask)
print(decoded)
# 當(dāng)payload確定之后，再往后數(shù)4個(gè)字節(jié)，這4個(gè)字節(jié)成為masking key，再之后的內(nèi)容就是接收到的數(shù)據(jù)部分。
# 數(shù)據(jù)部分的每一字節(jié)都要和masking key作異或位運(yùn)算，得出來的結(jié)果就是真實(shí)的數(shù)據(jù)內(nèi)容。
for i in range(len(decoded)):
chunk = decoded[i] ^ mask[i % 4]
byte_list.append(chunk)
new_str = str(byte_list, encoding="utf-8")
print(new_str)
return new_str
def encode(data):
data=str.encode(data)
head = b'\x81'
if len(data) < 126:
head += struct.pack('B', len(data))
elif len(data) <= 0xFFFF:
head += struct.pack('!BH', 126, len(data))
else:
head += struct.pack('!BQ', 127, len(data))
return head+data
def sendMessage(message, qu):
# global connectionlist
if(qu.qsize()>0):
connectionlist = qu.get()
qu.put(connectionlist)
print("sendMessage conn: ", connectionlist)
for connection in connectionlist.values():
connection.send(encode(message))
def deleteconnection(item):
global connectionlist
# if(conn.qsize()>0):
# connectionlist = conn.get()
# conn.put(connectionlist)
del connectionlist['connection'+item]
class WebSocket(threading.Thread):
GUID = "258EAFA5-E914-47DA-95CA-C5AB0DC85B11"
qu = None
def __init__(self,conn,index,name,remote, path="/", qu=None):
threading.Thread.__init__(self)
self.conn = conn
self.index = index
self.name = name
self.remote = remote
self.path = path
self.buffer = ""
self.qu = qu
def run(self):
print('Socket%s Start!' % self.index)
headers = {}
self.handshaken = False
while True:
if self.handshaken == False:
print ('Socket%s Start Handshaken with %s!' % (self.index,self.remote))
self.buffer += bytes.decode(self.conn.recv(1024))
if self.buffer.find('\r\n\r\n') != -1:
header, data = self.buffer.split('\r\n\r\n', 1)
for line in header.split("\r\n")[1:]:
key, value = line.split(": ", 1)
headers[key] = value
headers["Location"] = ("ws://%s%s" %(headers["Host"], self.path))
key = headers['Sec-WebSocket-Key']
token = b64encode(hashlib.sha1(str.encode(str(key + self.GUID))).digest())
handshake="HTTP/1.1 101 Switching Protocols\r\n"\
"Upgrade: websocket\r\n"\
"Connection: Upgrade\r\n"\
"Sec-WebSocket-Accept: "+bytes.decode(token)+"\r\n"\
"WebSocket-Origin: "+str(headers["Origin"])+"\r\n"\
"WebSocket-Location: "+str(headers["Location"])+"\r\n\r\n"
self.conn.send(str.encode(str(handshake)))
self.handshaken = True
print('Socket%s Handshaken with %s success!' %(self.index, self.remote))
sendMessage('Welcome, ' + self.name + ' !', self.qu)
else:
msg = decode(self.conn.recv(1024))
if msg == 'quit':
print ('Socket%s Logout!' % (self.index))
nowTime = time.strftime('%H:%M:%S',time.localtime(time.time()))
sendMessage('%s %s say: %s' % (nowTime, self.remote, self.name+' Logout'), self.qu)
deleteconnection(str(self.index))
self.conn.close()
break
else:
print('Socket%s Got msg:%s from %s!' % (self.index, msg, self.remote))
nowTime = time.strftime('%H:%M:%S',time.localtime(time.time()))
sendMessage('%s %s say: %s' % (nowTime, self.remote, msg), self.qu)
self.buffer = ""
class WebSocketServer(object):
def __init__(self):
self.socket = None
def begin(self, conn):
print( 'WebSocketServer Start!')
self.socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
self.socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) ##socket被關(guān)閉或服務(wù)器進(jìn)程終止后馬上釋放該服務(wù)器的端口
self.socket.bind(("172.16.10.233", 4003))
self.socket.listen(50)
global connectionlist
# if(conn.qsize()>0):
# connectionlist = conn.get()
# conn.put(connectionlist)
i = 0
while True:
connection, address = self.socket.accept()
username=address[0]
newSocket = WebSocket(connection,i,username,address, '/', conn)
newSocket.start()
connectionlist['connection'+str(i)]=connection
i = i + 1
##更新conn
if(conn.qsize()>0):
conn.get()
conn.put(connectionlist)
##虛擬壓力
# def presSim(qu):
if __name__ == "__main__":
server = WebSocketServer()
# server.begin(conn)
p=Process(target=server.begin, args=(conn,))
p.start()
import time
time.sleep(2)
for i in range(100):
sendMessage("hello "+ str(i), conn)
print(i)
if(conn.qsize()>0):
connectionlist = conn.get()
conn.put(connectionlist)
print(connectionlist)
time.sleep(1)
p.join()
總結(jié)
以上為個(gè)人經(jīng)驗(yàn)，希望能給大家一個(gè)參考，也希望大家多多支持腳本之家。
相關(guān)文章
py3nvml實(shí)現(xiàn)GPU相關(guān)信息讀取的案例分析
這篇文章主要介紹了py3nvml實(shí)現(xiàn)GPU相關(guān)信息讀取,此時(shí)就可以考慮使用py3nvml這樣的工具，針對(duì)于GPU任務(wù)執(zhí)行的過程進(jìn)行細(xì)化的分析，有助于提升GPU的利用率和程序執(zhí)行的性能，需要的朋友可以參考下2022-01-01
Python實(shí)現(xiàn)PDF按頁分割的技術(shù)指南
PDF文件處理是日常工作中的常見需求,特別是當(dāng)我們需要將大型PDF文檔拆分為多個(gè)部分時(shí),下面我們就來看看如何使用Python創(chuàng)建一個(gè)靈活的PDF分割工具吧2025-07-07
這篇文章主要介紹了Python常用外部指令執(zhí)行代碼實(shí)例,文中通過示例代碼介紹的非常詳細(xì)，對(duì)大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價(jià)值,需要的朋友可以參考下2020-11-11
pandas批量拆分與合并Excel文件的實(shí)現(xiàn)示例
本文介紹了Pandas中基于整數(shù)位置的iloc和基于標(biāo)簽的loc方法進(jìn)行數(shù)據(jù)索引和切片的操作,并將大Excel文件拆分合并,具有一定的參考價(jià)值,感興趣的可以了解一下2025-11-11
這篇文章主要介紹了Python AES加密實(shí)例解析，具有一定借鑒價(jià)值,需要的朋友可以參考下2018-01-01
這篇文章主要介紹了OpenCV基本操作之圖像梯度，本文給大家介紹的非常詳細(xì)，對(duì)大家的學(xué)習(xí)或工作具有一定的參考借鑒價(jià)值，需要的朋友可以參考下2021-09-09
今天小編就為大家分享一篇Python實(shí)現(xiàn)圖像的垂直投影示例，具有很好的參考價(jià)值，希望對(duì)大家有所幫助。一起跟隨小編過來看看吧2020-01-01
這篇文章主要介紹了Python語言中的按位運(yùn)算符，有需要的朋友可以參考一下2013-11-11
這篇文章主要介紹了Python?列表(list)的常用方法,這節(jié)主要講列表，列表用于存儲(chǔ)任意數(shù)目、任意類型的數(shù)據(jù)集合，本文通過示例代碼給大家詳細(xì)講解，需要的朋友可以參考下2023-05-05
python實(shí)現(xiàn)超時(shí)退出的三種方式總結(jié)
這篇文章主要介紹了python實(shí)現(xiàn)超時(shí)退出的三種方式總結(jié)，具有很好的參考價(jià)值，希望對(duì)大家有所幫助。如有錯(cuò)誤或未考慮完全的地方，望不吝賜教2022-11-11