商业源码 python 怎么搭建简单的web服务器?
利用Python自带的包可以建立简单的web服务器。在DOS里cd到准备做服务器根目录的路径下，输入命令：\x0d\python -m Web服务器模块 [端口号，默认8000]\x0d\例如：\x0d\python -m SimpleHTTPServer 8080\x0d\然后就可以在浏览器中输入\x0d\h ttp://loca lhost:端口号/路径\x0d\来访问服务器资源。 \x0d\例如：\x0d\h ttp://local host:808 0/indexh tm（当然indexhtm文件得自己创建）\x0d\其他机器也可以通过服务器的IP地址来访问。\x0d\\x0d\这里的“Web服务器模块”有如下三种：\x0d\\x0d\BaseHTTPServer: 提供基本的Web服务和处理器类，分别是HTTPServer和BaseHTTPRequestHandler。\x0d\SimpleHTTPServer: 包含执行GET和HEAD请求的SimpleHTTPRequestHandler类。\x0d\CGIHTTPServer: 包含处理POST请求和执行CGIHTTPRequestHandler类。
用twisted，用工厂管理连接，每个连接建立transport使用简单方便！参看下面代码：
#Copyright(c)ThePyAMFProject
#SeeLICENSEtxtfordetails
\"\"\"
ExamplesocketserverusingTwisted
@see:U{Documentationforthisexample
raiseSystemExit
fromtwistedinternetprotocolimportProtocol,Factory
fromtwistedinternetimportreactor
fromdatetimeimportdatetime
importpyamf
classTimerProtocol(Protocol):
interval=10#客户端链接到server后，server往客户端发送时间的间隔
encoding=pyamfAMF3
timeout=20#客户端链接到server后多少时间不操作就断开链接的timeout
def__init__(self):
selfstarted=False
#设置编码器
selfencoder=pyamfget_encoder(selfencoding)、
#设置server端将数据编码成amf后存放的缓存地址
selfstream=selfencoderstream
defconnectionLost(self,reason):
ProtocolconnectionLost(self,reason)
print\"locstconnection:
socket服务器再细分可分为多种了，tcp,udp,websocket，都是调用socket模块，但是具体实现起来有一点细微的差别
先给出一个tcp和udp通过socket协议实现的聊天室的例子
python聊天室(python27版本)：
都是分别运行serverpy和clientpy，就可以进行通讯了。
TCP版本：
socket-tcp-serverpy（服务端）:
#-- encoding:utf-8 --
#socketgetaddrinfo(host, port, family=0, socktype=0, proto=0, flags=0)
#根据给定的参数host/port，相应的转换成一个包含用于创建socket对象的五元组，
#参数host为域名，以字符串形式给出代表一个IPV4/IPV6地址或者None
#参数port如果字符串形式就代表一个服务名，比如“http”"ftp""email"等，或者为数字，或者为None
#参数family为地主族，可以为AF_INET ，AF_INET6 ，AF_UNIX
#参数socktype可以为SOCK_STREAM(TCP)或者SOCK_DGRAM(UDP)
#参数proto通常为0可以直接忽略
#参数flags为AI_的组合，比如AI_NUMERICHOST，它会影响函数的返回值
#附注：给参数host,port传递None时建立在C基础，通过传递NULL。
#该函数返回一个五元组(family, socktype, proto, canonname, sockaddr)，同时第五个参数sockaddr也是一个二元组(address, port)
#更多的方法及链接请访问
# Echo server program
from socket import
import sys
import threading
from time import ctime
from time import localtime
import traceback
import time
import subprocess
reload(sys)
syssetdefaultencoding("utf8")
HOST='127001'
PORT=8555 #设置侦听端口
BUFSIZ=1024
class TcpServer():
def __init__(self):
selfADDR=(HOST, PORT)
try:
selfsock=socket(AF_INET, SOCK_STREAM)
print '%d is open' % PORT
selfsockbind(selfADDR)
selfsocklisten(5)
#设置退出条件
selfSTOP_CHAT=False
# 所有监听的客户端
selfclients = {}
selfthrs = {}
selfstops = []
except Exception,e:
print "%d is down" % PORT
return False
def IsOpen(ip, port):
s = socket(AF_INET, SOCK_STREAM)
try:
sconnect((ip, int(port)))
# sshutdown(2)
# 利用shutdown()函数使socket双向数据传输变为单向数据传输。shutdown()需要一个单独的参数，
# 该参数表示s了如何关闭socket。具体为：0表示禁止将来读；1表示禁止将来写；2表示禁止将来读和写。
print '%d is open' % port
return True
except:
print '%d is down' % port
return False
def listen_client(self):
while not selfSTOP_CHAT:
print(u'等待接入，侦听端口:%d' % (PORT))
selftcpClientSock, selfaddr=selfsockaccept()
print(u'接受连接，客户端地址：',selfaddr)
address = selfaddr
#将建立的client socket链接放到列表selfclients中
selfclients[address] = selftcpClientSock
#分别将每个建立的链接放入进程中，接收且分发消息
selfthrs[address] = threadingThread(target=selfreadmsg, args=[address])
selfthrs[address]start()
timesleep(05)
def readmsg(self,address):
#如果地址不存在，则返回False
if address not in selfclients:
return False
#得到发送消息的client socket
client = selfclients[address]
while True:
try:
#获取到消息内容data
data=clientrecv(BUFSIZ)
except:
print(e)
selfclose_client(address)
break
if not data:
break
#python3使用bytes，所以要进行编码
#s='%s发送给我的信息是:[%s] %s' %(addr[0],ctime(), datadecode('utf8'))
#对日期进行一下格式化
ISOTIMEFORMAT='%Y-%m-%d %X'
stime=timestrftime(ISOTIMEFORMAT, localtime())
s=u'%s发送给我的信息是:%s' %(str(address),datadecode('utf8'))
#将获得的消息分发给链接中的client socket
for k in selfclients:
selfclients[k]send(sencode('utf8'))
selfclients[k]sendall('sendall:'+sencode('utf8'))
print str(k)
print([stime], ':', datadecode('utf8'))
#如果输入quit(忽略大小写),则程序退出
STOP_CHAT=(datadecode('utf8')upper()=="QUIT")
if STOP_CHAT:
print "quit"
selfclose_client(address)
print "already quit"
break
def close_client(self,address):
try:
client = selfclientspop(address)
selfstopsappend(address)
clientclose()
for k in selfclients:
selfclients[k]send(str(address) + u"已经离开了")
except:
pass
print(str(address)+u'已经退出')
if __name__ == '__main__':
tserver = TcpServer()
tserverlisten_client()
——————————华丽的分割线——————————
socket-tcp-clientpy （客户端）：
#-- encoding:utf-8 --
from socket import
import sys
import threading
import time
reload(sys)
syssetdefaultencoding("utf8")
#测试，连接本机
HOST='127001'
#设置侦听端口
PORT=8555
BUFSIZ=1024
class TcpClient:
ADDR=(HOST, PORT)
def __init__(self):
selfHOST = HOST
selfPORT = PORT
selfBUFSIZ = BUFSIZ
#创建socket连接
selfclient = socket(AF_INET, SOCK_STREAM)
selfclientconnect(selfADDR)
#起一个线程，监听接收的信息
selftrecv = threadingThread(target=selfrecvmsg)
selftrecvstart()
def sendmsg(self):
#循环发送聊天消息，如果socket连接存在则一直循环，发送quit时关闭链接
while selfclientconnect_ex(selfADDR):
data=raw_input('>:')
if not data:
break
selfclientsend(dataencode('utf8'))
print(u'发送信息到%s：%s' %(selfHOST,data))
if dataupper()=="QUIT":
selfclientclose()
print u"已关闭"
break
def recvmsg(self):
#接收消息，如果链接一直存在，则持续监听接收消息
try:
while selfclientconnect_ex(selfADDR):
data=selfclientrecv(selfBUFSIZ)
print(u'从%s收到信息：%s' %(selfHOST,datadecode('utf8')))
except Exception,e:
print str(e)
if __name__ == '__main__':
client=TcpClient()
clientsendmsg()
UDP版本：
socket-udp-serverpy
# -- coding:utf8 --
import sys
import time
import traceback
import threading
reload(sys)
syssetdefaultencoding('utf-8')
import socket
import traceback
HOST = "127001"
PORT = 9555
CHECK_PERIOD = 20
CHECK_TIMEOUT = 15
class UdpServer(object):
def __init__(self):
selfclients = []
selfbeats = {}
selfADDR = (HOST,PORT)
try:
selfsock = socketsocket(socketAF_INET, socketSOCK_DGRAM)
selfsockbind(selfADDR) # 绑定同一个域名下的所有机器
selfbeattrs = threadingThread(target=selfcheckheartbeat)
selfbeattrsstart()
except Exception,e:
tracebackprint_exc()
return False
def listen_client(self):
while True:
timesleep(05)
print "hohohohohoo"
try:
recvData,address = selfsockrecvfrom(2048)
if not recvData:
selfclose_client(address)
break
if address in selfclients:
senddata = u"%s发送给我的信息是:%s" %(str(address),recvDatadecode('utf8'))
if recvDataupper() == "QUIT":
selfclose_client(address)
if recvData == "HEARTBEAT":
selfheartbeat(address)
continue
else:
selfclientsappend(address)
senddata = u"%s发送给我的信息是:%s" %(str(address),u'进入了聊天室')
for c in selfclients:
try:
selfsocksendto(senddata,c)
except Exception,e:
print str(e)
selfclose_client(c)
except Exception,e:
# tracebackprint_exc()
print str(e)
pass
def heartbeat(self,address):
selfbeats[address] = timetime()
def checkheartbeat(self):
while True:
print "checkheartbeat"
print selfbeats
try:
for c in selfclients:
print timetime()
print selfbeats[c]
if selfbeats[c] + CHECK_TIMEOUT <timetime():
print u"%s心跳超时，连接已经断开" %str(c)
selfclose_client(c)
else:
print u"checkp%s,没有断开" %str(c)
except Exception,e:
tracebackprint_exc()
print str(e)
pass
timesleep(CHECK_PERIOD)
def close_client(self,address):
try:
if address in selfclients:
selfclientsremove(address)
if selfbeatshas_key(address):
del selfbeats[address]
print selfclients
for c in selfclients:
selfsocksendto(u'%s已经离开了' % str(address),c)
print(str(address)+u'已经退出')
except Exception,e:
print str(e)
raise
if __name__ == "__main__":
udpServer = UdpServer()
udpServerlisten_client()
——————————华丽的分割线——————————
socket-udp-clientpy:
# -- coding:utf8 --
import sys
import threading
import time
reload(sys)
syssetdefaultencoding('utf-8')
import socket
HOST = "127001"
PORT = 9555
#BEAT_PORT = 43278
BEAT_PERIOD = 5
class UdpClient(object):
def __init__(self):
selfclientsock = socketsocket(socketAF_INET,socketSOCK_DGRAM)
selfHOST = HOST
selfADDR = (HOST,PORT)
selfclientsocksendto(u'请求建立链接',selfADDR)
selfrecvtrs = threadingThread(target=selfrecvmsg)
selfrecvtrsstart()
selfhearttrs = threadingThread(target=selfheartbeat)
selfhearttrsstart()
def sendmsg(self):
while True:
data = raw_input(">:")
if not data:
break
selfclientsocksendto(dataencode('utf-8'),selfADDR)
if dataupper() == 'QUIT':
selfclientsockclose()
break
def heartbeat(self):
while True:
selfclientsocksendto('HEARTBEAT',selfADDR)
timesleep(BEAT_PERIOD)
def recvmsg(self):
while True:
recvData,addr = selfclientsockrecvfrom(1024)
if not recvData:
break
print(u'从%s收到信息：%s' %(selfHOST,recvDatadecode('utf8')))
if __name__ == "__main__":
udpClient = UdpClient()
udpClientsendmsg()
# -- coding: utf-8 --
# -- version: beta-00 --
####################################################################################################
import socket
####################################################################################################
class Main():
def __init__(self):
selfhost = '127001'
selfport = 9999
#--------------------------------------------------------------------------------------------------#
def Start(self):
clientSock = socketsocket(socketAF_INET, socketSOCK_STREAM)
clientSockconnect((selfhost, selfport))
while True:
clientSocksend(b'hello')
dataRecv = clientSockrecv(1024)
if not dataRecv:
break
clientSockclose()
####################################################################################################
def test():
m = Main()
mStart()
if __name__ == '__main__':
test()
Python 是脚本语言，也就是中间件语言，其内核仍然是纯 c 的性能表达的，而主要性能消耗在脚本的实时编译上。
而对比c，同样的功能，python可以表达比c更加精炼，当然牺牲了一些性能。
如果确实对某个模组性能不满意，还可以使用 c 编写 Python 模块为其加速。
面对一些并不需要过多性能而可以获得更快捷的开发速度，Python的优势完全可以盖过“劣势”。
可以说，Python基本完美。
例如知乎就是使用 Python 制作的。
Google 的许多页面也是使用 Python 进行渲染的。
0条评论