![]()
# -*- coding: utf-8 -*-
import datetime
import os
import socket # 导入 socket 模块
import threading
import time
from Log import Log
log = Log()
class Socket:
# 初始化
def __init__(self, host="0.0.0.0", port=60000, handle=lambda x: x, exited=lambda x: x):
self.host = host
self.port = port
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) # 创建 socket 对象
sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
sock.bind((host, port))
sock.listen(5) # 最大等待数
self.sock = sock
self.client_pool = {} # 连接池
self.thread_pool = {} # 线程池
self.handle = handle # 回调 消息处理
self.exited = exited # 回调 断开处理
thread = threading.Thread(target=self.accept_client, args=())
thread.setDaemon(True)
thread.start()
def accept_client(self):
# 接收新连接
while True:
client, addr = self.sock.accept() # 阻塞，等待客户端连接
try:
# 加入连接池
client_id = str(id(client))
log.write("建立新链接: %s -> %s" % (client_id, addr))
print "\033[1;32m 建立新链接: %s -> %s \033[0m" % (client_id,addr)
self.client_pool[client_id] = client
thread = threading.Thread(target=self.message_handle, args=(client,))
thread.setDaemon(True)
thread.start()
self.thread_pool[client_id] = thread
except BaseException as e:
print "accept_client - ERROR!"
def message_handle(self, client):
# 消息处理
# 当连接被断开或意外断开时 跳出循环 线程终止
try:
while True:
bytes = client.recv(1024)
if len(bytes) > 0:
self.handle(bytes)
else:
print "connect closed:",
print self.close_client(client)
except BaseException as e:
self.close_client(client)
def send_message(self, cid, msg):
cid = str(cid)
try:
sock_id = self.hardware[cid]
client = self.pool[sock_id]
client.sendall(msg.decode("utf-8"))
return True
except BaseException as e:
return False
def close_client(self, client):
client_id = str(id(client))
# # 删除连接
if client_id in list(self.client_pool.keys()):
del self.client_pool[client_id]
del self.thread_pool[client_id]
try:
client.shutdown(socket.SHUT_RDWR)
client.close()
except BaseException as e:
pass
self.exited(bytes)
return client_id
def heartbeat(self):
# 轮询链接测试连通性 不通则断开
for client_id in list(self.client_pool.keys()):
client = self.client_pool[client_id]
timestamp = str(int(time.time()))
try:
client.sendall(timestamp.decode("utf-8"))
except BaseException as e:
self.close_client(client)
if __name__ == '__main__':
tcp = Socket()
while True:
print tcp.client_pool
print tcp.thread_pool
print threading.active_count()
tcp.heartbeat()
print
time.sleep(1)
