庆祝中华人民共和国成立70周年

高举旗帜 团结一致 锐意进取

简易socket服务器

# -*- 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)

点赞

发表评论

电子邮件地址不会被公开。 必填项已用*标注