基于多线程与多进程的简易聊天室

刺骨的言语ヽ痛彻心扉 2022-05-16 06:19 197阅读 0赞

### 功能实现 ###

**服务端**

*  当有新的客户端连接时，显示连接信息
 *  当有客户端发送消息时，显示消息发送者和内容

**客户端**

*  客户端连接时，显示本客户端的连接信息
 *  客户端连接时，已连接的客户端收到新客户端的连接信息
 *  当某一客户端发送消息时，所有客户端收到发送者和内容

### 运行预览 ###

![70][]

![70 1][]

![70 2][]

![70 3][]

### 思路 ###

**服务端**

1.设cpu核数为n，首先开n个进程

2.每个进程开2n个线程

3.使用进程安全队列和字典，前者存放发送的消息，后者保存连接的客户端

4.写send\_data方法，用于向每个客户端发送数据

5.写worker\_process方法，进程调用这个方法用于开线程并接收客户端连接，并调用send\_data方法

6.开线程调用worker\_thread方法用于客户端接收数据

**客户端**

1.recv\_data方法，接收并显示服务端发送的数据

2.while True可以不断发送信息

### 完整代码 ###

**服务端**

from multiprocessing import Pool, cpu_count,Manager
    from multiprocessing.pool import ThreadPool
    import socket
    
    
    #从队列中拿出数据，发给所有连接上的客户端
    def send_data(dict_proxy, queue_proxy):
        while True:
            data = queue_proxy.get()        #从队列中拿出消息，发送给所有连接上来的客户端
            print(data.decode())
            for conn in dict_proxy.values():
                conn.send(data)
    
    
    
    def worker_thread(connection, addr, dict_proxy, queue_proxy):
        while True:
            try:
                recv_data = connection.recv(1024)
                if recv_data:
                    data = "来自{addr} 的消息:{data}".format(addr=addr, data=recv_data.decode())
                    queue_proxy.put(data.encode())     #把消息添加到到队列中
                else:
                    raise Exception
            except:
                dict_proxy.pop(addr)                #从字典中删掉退出的客户端
                data = '用户{}退出'.format(addr)
                queue_proxy.put(data.encode())      #把退出消息添加到队列中
                connection.close()
                break
    
    
    
    def worker_process(server, dict_proxy, queue_proxy):
        thread_pool = ThreadPool( cpu_count()*2 )   #分配2倍CPU个数的线程
        thread_pool.apply_async(send_data, args=(dict_proxy, queue_proxy))
        #发送数据
        while True:
            conncetion, remote_address = server.accept()        #接收客户端连接，生成对等套接字
    
            dict_proxy[remote_address] = conncetion   #把套接字加入字典中
    
            data = '用户{}登录'.format(remote_address)
            queue_proxy.put(data.encode())            #将用户登录消息添加到队列中
            thread_pool.apply_async(worker_thread, args=(conncetion, remote_address, dict_proxy, queue_proxy))
            #接收数据
    
    if __name__ == '__main__':
    
        server = socket.socket()
        server.bind(('127.0.0.1', 8088))
        server.listen(1000) #最多供1000客户端连接
    
        mgr = Manager()
        dict_proxy = mgr.dict()         #进程安全字典，用来保存连接上来的客户端，
        queue_proxy = mgr.Queue()       #进程安全队列，把客户端发过来的消息通过队列传递
    
        n = cpu_count()                 #cpu核数
        process_pool = Pool(n)
        for i in range(n):              #充分利用CPU，为每一个CPU分配一个进程
            process_pool.apply_async(worker_process, args=(server, dict_proxy, queue_proxy))
                 #把server服务端放到进程里面
    
        process_pool.close() #停止提交数据
        process_pool.join() #阻塞，等待子进程运行完毕才继续运行主进程

**客户端**

import socket
    import threading
    
    
    def recv_data():  #接收并显示服务端发送的信息
        while True:
            data = client.recv(1024)
            print(data.decode())
    
    
    
    client = socket.socket()
    client.connect(('127.0.0.1', 8088))
    
    thread = threading.Thread(target=recv_data, daemon=True)
    thread.start()
    
    while True:
        a = input('')
        client.send(a.encode())

[70]: /images/20220516/e63240ce1cb84f49ba28b07145b3ab65.png
[70 1]: /images/20220516/5cb82a2d331c4dd6baacdd33cbbf1fe0.png
[70 2]: /images/20220516/34dd450c439545978a1d315d4702f9a0.png
[70 3]: /images/20220516/165f84c85a8141f98c3779fbe4fee405.png