腳本如下：

# -*- coding:utf-8 -*- ''' @Author: Stefan@File: server_listener.py@Date: 2016-11-09If you execute this test script on servers, you have to modify the system parameter below: sysctl net.ipv4.tcp_tw_recycle=1 ''' import sys sys.path.append('/export/servers/app/xxxx') import time import socket import threading import multiprocessing from src.xxxxclient.xxxx_client.lib.socket import header_pack, receive HEADER_LENGTH = 16 DEFAULT_TCP_PORT = 1104 DATA = {'xxxx_ADMIN': 'thread_keepalive'}count_of_processes = range(1) # DO NOT modify the parameter below count_of_threads = range(1)# Unit: second execute_time = 60def loop(times):""":return:"""time_start_now = time.time()time_future = time_start_now + execute_timewhile True:if time_start_now > time_future:breakelse:time_start = time.time()S = socket.socket(socket.AF_INET, socket.SOCK_STREAM)# S.connect(('127.0.0.1', DEFAULT_TCP_PORT))S.connect(('x.x.x.x', DEFAULT_TCP_PORT))header, data = header_pack(DATA, HEADER_LENGTH)S.sendall(header+data)header, result = receive(S, HEADER_LENGTH, 1024)S.close()time_end = time.time()time_used = time_end - time_startprint "Thread lasted %.2f sec. Status: %s" % (time_used, result)time_end_now = time.time()time_gap = time_end_now - time_start_nowtime_start_now += time_gapdef multi_threads():""":return:"""threads_list = list()for i in count_of_threads:t = threading.Thread(target=loop, args=str(count_of_threads[i]).split())threads_list.append(t)for i in count_of_threads:threads_list[i].start()for i in count_of_threads:threads_list[i].join()class muliti_process(multiprocessing.Process):""""""def __init__(self):""":return:"""multiprocessing.Process.__init__(self)self.processes_list = list()def run(self):""":return:"""for i in count_of_processes:p = multiprocessing.Process(target=multi_threads,)self.processes_list.append(p)for i in self.processes_list:i.start()if __name__ == '__main__':p = muliti_process()global_start = time.time()p.start()p.join()global_end = time.time()global_used = global_end - global_startprint "==================\nTotal %.2f sec" % global_used

轉載于:https://www.cnblogs.com/stefan-liu/p/6138565.html

總結

以上是生活随笔為你收集整理的socket 多线程安全、粘包问题的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。