添加两种服务器-客户端，一种是多进程，另一种是异步的，它们做的几乎一样，异步的更健壮，请在此处阅读原因：
Threads vs. Async.
我的例子：
使用多进程：
import multiprocessing
import socket
import time
HOST = "0.0.0.0"
PORT = 9000
def handle(connection, address):
try:
while True:
data = connection.recv(1024)
connection.sendall(data + ' server time {}'.format(time.time()))
except:
pass
finally:
connection.close()
class Server(object):
def __init__(self, hostname, port):
self.hostname = hostname
self.port = port
def start(self):
self.socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
self.socket.bind((self.hostname, self.port))
self.socket.listen(1)
while True:
conn, address = self.socket.accept()
process = multiprocessing.Process(
target=handle, args=(conn, address))
process.daemon = True
process.start()
if __name__ == "__main__":
server = Server(HOST, PORT)
try:
print 'start'
server.start()
except:
print 'something wrong happened, a keyboard break ?'
finally:
for process in multiprocessing.active_children():
process.terminate()
process.join()
print 'Goodbye'
还有它的客户：
import sys
import threading
import time
import socket
SOCKET_AMOUNT = 100
HOST = "localhost"
PORT = 9000
def myclient(ip, port, message):
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.connect((ip, port))
sock.sendall(message)
result = sock.recv(1024)
print result + ' final clnt time {}'.format(time.time())
sock.close()
if __name__ == "__main__":
thread_list = []
for i in range(SOCKET_AMOUNT):
msg = "Thread #{}, clnt time {}".format(i, time.time())
client_thread = threading.Thread(
target=myclient, args=(HOST, PORT, msg))
thread_list.append(client_thread)
client_thread.start()
waiting = time.time()
[x.join() for x in thread_list]
done = time.time()
print 'DONE {}. Waiting for {} seconds'.format(done, done-waiting)
下一个服务器更强大！数据不会丢失！！！
服务器：
import asyncore
import socket
import time
import logging
import json
class Server(asyncore.dispatcher):
def __init__(self, host, port):
self.logger = logging.getLogger('SERVER')
asyncore.dispatcher.__init__(self)
self.create_socket(socket.AF_INET, socket.SOCK_STREAM)
self.set_reuse_addr()
self.bind(('', port))
self.listen(confjson.get('SERVER_QUEUE_SIZE', None))
self.logger.debug('binding to {}'.format(self.socket.getsockname()))
def handle_accept(self):
socket, address = self.accept()
self.logger.debug('new connection accepted')
EchoHandler(socket)
class EchoHandler(asyncore.dispatcher_with_send):
def handle_read(self):
msg = self.recv(confjson.get('RATE', None))
self.out_buffer = msg
self.out_buffer += ' server recieve: {}'.format(time.time())
if not self.out_buffer:
self.close()
if __name__ == "__main__":
logging.basicConfig(level=logging.DEBUG,
format='%(name)s: %(message)s',
)
with open('config.json', 'r') as jfile:
confjson = json.load(jfile)
try:
logging.debug('Server start')
server = Server(confjson.get('HOST', None),
confjson.get('PORT', None))
asyncore.loop()
except:
logging.error('Something happened,\n'
'if it was not a keyboard break...\n'
'check if address taken, '
'or another instance is running. Exit')
finally:
logging.debug('Goodbye')
还有异步客户端：
import asyncore
import socket
import time
import logging
import json
class Client(asyncore.dispatcher_with_send):
def __init__(self, host, port, message, pk):
self.logger = logging.getLogger('CLIENT')
asyncore.dispatcher.__init__(self)
self.create_socket(socket.AF_INET, socket.SOCK_STREAM)
self.host = host
self.port = port
self.connect((host, port))
self.out_buffer = message
self.clientID = pk
self.logger.debug('Connected #{}'.format(self.clientID))
def handle_close(self):
self.close()
def handle_read(self):
rec_msg = self.recv(confjson.get('RATE', None))
self.logger.debug('#{}, {} back at client {}'.format(self.clientID,
rec_msg,
time.time()
)
)
self.close()
if __name__ == "__main__":
logging.basicConfig(level=logging.DEBUG,
format='%(name)s: %(message)s',
)
with open('config.json', 'r') as jfile:
confjson = json.load(jfile)
clients = []
for idx in range(confjson.get('SOCKET_AMOUNT', None)):
msg = "Start: {}".format(time.time())
clients.append(Client(confjson.get('HOST', None),
confjson.get('PORT', None),
msg,
idx)
)
start = time.time()
logging.debug(
'Starting async loop for all connections, unix time {}'.format(start))
asyncore.loop()
logging.debug('{}'.format(time.time() - start))
还有一个小的配置文件：
{
"HOST": "127.0.0.1",
"PORT": 5007,
"RATE": 8096,
"SERVER_QUEUE_SIZE": 16,
"SOCKET_AMOUNT": 100
}