import subprocess

import re

import socket

import time

# 执行 ARP 扫描，获取局域网内所有活跃设备的 IP 地址

def arp_scan ():

    arp_result = subprocess.check_output (['arp', '-a']).decode ('cp1252', errors='ignore')

    ip_list = re.findall (r'(\d+\.\d+\.\d+\.\d+)', arp_result)

    return ip_list

# 尝试连接到指定 IP 的 50000 端口

def try_connect (ip):

    with socket.socket (socket.AF_INET, socket.SOCK_STREAM) as s:

        s.settimeout (1)

        try:

            s.connect ((ip, 50000))

            return True

        except socket.error:

            return False

# 执行 ARP 扫描，获取 IP 列表

ips = arp_scan ()

# 标记是否找到服务端

server_found = False

# 遍历 IP 列表，尝试连接到 50000 端口

def scan_list (lst):

    for ip in ips:

        if try_connect (ip):

            print (f"服务端存在，IP 地址为：{ip}")

            server_found = True

            break

scan_list (lst=ips)

class threading.Thread (group=None,   ## 一般设置为 None ，这是为以后的一些特性预留的

                       target=None,  ## 当线程启动的时候要执行的函数

                       name=None,    ## 线程的名字，默认会分配一个唯一名字 Thread-N

                       args=(),      ## 传递给 target 的参数元组（整形和浮点型数据输入到 args 里）。

                       kwargs={})    ## 传递给 target 的关键字参数字典（字符串型数据输入到这里）。

import threading

def function (i):

    print ("function called by thread % i\n" % i)

    return

#threads = []

for i in range (5):

    t = threading.Thread (target=function, args=(i,)) ## 用 function 函数初始化一个 Thread 对象 t，并将参数 i 传入；

    #threads.append (t) 

    t.start () ## 线程被创建后不会马上执行，需要手动调用 .start () 方法执行线程

    t.join () ## 阻塞调用 t 线程的主线程，t 线程执行结束，主线程才会继续执行

import threading

# 定义一个线程函数，接受浮点型和字符串型参数

def calculate (data_float, data_string):

    result = data_float * 2

    print (f"Thread result for {data_float}: {result}")

    print (f"Additional string data: {data_string}")

# 创建多个线程并启动

threads = []

data_float = [1.5, 2.5, 3.5]  # 浮点型数据

data_string = ["Hello", "World", "Python"]  # 字符串型数据

for i in range (len (data_float)):

    thread = threading.Thread (target=calculate, args=(data_float [i], data_string [i]))

    threads.append (thread)

    thread.start ()

# 等待所有线程执行完成

for thread in threads:

    thread.join ()

print ("All threads have finished execution.")

import threading

class MyThread (threading.Thread):

    def run (self):

        print (' 执行线程 ', self.args)

t = MyThread (args=(1,))

t.start ()

from concurrent.futures import ThreadPoolExecutor

from unittest import result

# 创建一个包含 5 个线程的线程池

pool = ThreadPoolExecutor (5)

# 定义一个函数，该函数将被线程池中的线程执行

def task_function (x):

    return x * x

# 使用线程池执行任务

results = [pool.submit (task_function, i) for i in range (10)]

# 获取所有任务的结果

for future in results:

    print (future.result ())

import math

import concurrent.futures

def calculate_factorial (number):

    """计算一个数的阶乘"""

    return math.factorial (number)

def main ():

    numbers = [1, 2, 3, 4, 5]  # 任务列表，计算这些数的阶乘

    # 创建一个包含 3 个线程的线程池

    with concurrent.futures.ThreadPoolExecutor (max_workers=3) as executor:

        # 使用 submit 函数来提交任务

        futures = {executor.submit (calculate_factorial, number): number for number in numbers}

        # 获取并打印结果

        for future in concurrent.futures.as_completed (futures):

            number = futures [future]

            try:

                result = future.result ()

            except Exception as exc:

                print (f"{number} generated an exception: {exc}")

            else:

                print (f"The factorial of {number} is {result}")

if __name__ == "__main__":

    main ()

import math

import concurrent.futures

def calculate_factorial (number):

    """计算一个数的阶乘"""

    return math.factorial (number)

def main ():

    numbers = [1, 2, 3, 4, 5]  # 任务列表，计算这些数的阶乘

    # 创建一个包含 3 个线程的线程池

    with concurrent.futures.ThreadPoolExecutor (max_workers=3) as executor:

        # 使用 map 函数来提交任务并保持执行顺序

        # 使用 executor.map () 方法将任务提交给线程池。与 executor.submit () 不同，map () 方法会等待所有任务完成，并返回一个结果列表，保持了输入列表的顺序

        results = executor.map (calculate_factorial, numbers)

        # 打印结果

        for number, result in zip (numbers, results):

            print (f"The factorial of {number} is {result}")

if __name__ == "__main__":

    main ()

import threading

shared_resource_with_lock = 0

shared_resource_with_no_lock = 0

COUNT = 100000

shared_resource_lock = threading.Lock () ## 锁

## 有锁的情况

def increment_with_lock ():

    # shared_resource_with_lock 即最外面的 shared_resource_with_lock

    # 这样写就不需要再通过函数的参数引入 shared_resource_with_lock 了

    global shared_resource_with_lock

    for _ in range (COUNT):

        shared_resource_lock.acquire () ## 获取 锁

        shared_resource_with_lock += 1

        shared_resource_lock.release () ## 释放 锁

def decrement_with_lock ():

    global shared_resource_with_lock

    for _ in range (COUNT):

        shared_resource_lock.acquire ()

        shared_resource_with_lock -= 1

        shared_resource_lock.release ()

## 没有锁的情况

def increment_without_lock ():

    global shared_resource_with_no_lock

    for _ in range (COUNT):

        shared_resource_with_no_lock += 1

def decrement_without_lock ():

    global shared_resource_with_no_lock

    for _ in range (COUNT):

        shared_resource_with_no_lock -= 1

if __name__ == "__main__":

    t1 = threading.Thread (target=increment_with_lock)

    t2 = threading.Thread (target=decrement_with_lock)

    t3 = threading.Thread (target=increment_without_lock)

    t4 = threading.Thread (target=decrement_without_lock)

    ## 开启线程

    t1.start ()

    t2.start ()

    t3.start ()

    t4.start ()

    ## .join ()

    t1.join ()

    t2.join ()

    t3.join ()

    t4.join ()

    print ("the value of shared variable with lock management is % s" % shared_resource_with_lock)

    print ("the value of shared variable with race condition is % s" % shared_resource_with_no_lock)

import threading

import time

class Box (object):

    lock = threading.RLock ()

    def __init__(self):

        self.total_items = 0

    def execute (self, n):

        Box.lock.acquire ()

        self.total_items += n

        Box.lock.release ()

    def add (self):

        Box.lock.acquire ()

        self.execute (1)

        Box.lock.release ()

    def remove (self):

        Box.lock.acquire ()

        self.execute (-1)

        Box.lock.release ()

def adder (box, items):

    while items > 0:

        print ("adding 1 item in the box")

        box.add ()

        time.sleep (1)

        items -= 1

def remover (box, items):

    while items > 0:

        print ("removing 1 item in the box")

        box.remove ()

        time.sleep (1)

        items -= 1

if __name__ == "__main__":

    items = 5

    print ("putting % s items in the box"% items)

    box = Box ()

    t1 = threading.Thread (target=adder, args=(box, items))

    t2 = threading.Thread (target=remover, args=(box, items))

    t1.start ()

    t2.start ()

    t1.join ()

    t2.join ()

    print ("% s items still remain in the box" % box.total_items)

import threading

import time

import random

# 创建一个信号量 semaphore，初始值为 0。

# 信号量是一种同步机制，用于控制对共享资源的访问。

semaphore = threading.Semaphore (0)

print ("init semaphore % s" % semaphore._value)  # 打印初始信号量的值。

# 消费者线程将执行的函数。

def consumer ():

    print ("consumer is waiting.")  # 打印信息，表明消费者正在等待。

    semaphore.acquire ()  # 消费者尝试获取信号量，如果信号量的值小于 1，则等待。

    print ("consumer notify: consumed item number % s" % item)  # 打印消费者消费的项目编号。

    print ("consumer semaphore % s" % semaphore._value)  # 在消费后打印信号量的当前值。

# 生产者线程将执行的函数。

def producer ():

    global item  # 声明 item 为全局变量，以便在函数内部修改。

    time.sleep (10)  # 生产者线程暂停 10 秒，模拟生产过程耗时。

    item = random.randint (0, 1000)  # 生产者生成一个随机的项目编号。

    print ("producer notify : produced item number % s" % item)  # 打印生产者生产的产品编号。

    semaphore.release ()  # 生产者释放信号量，增加信号量的值，允许其他等待的线程继续执行。

    print ("producer semaphore % s" % semaphore._value)  # 在生产后打印信号量的当前值。

# 主程序入口。

if __name__ == "__main__":

    for _ in range (0, 5):  # 循环 5 次，模拟生产和消费过程。

        t1 = threading.Thread (target=producer)  # 创建生产者线程。

        t2 = threading.Thread (target=consumer)  # 创建消费者线程。

        t1.start ()  # 启动生产者线程。

        t2.start ()  # 启动消费者线程。

        t1.join ()  # 等待生产者线程完成。

        t2.join ()  # 等待消费者线程完成。

    print ("program terminated")  # 打印程序结束的信息。

class Process (

    group: None = None,

    target: ((...) -> object) | None = None,

    name: str | None = None,

    args: Iterable [Any] = (),

    kwargs: Mapping [str, Any] = {},

    *,

    daemon: bool | None = None

)

start ()：启动子进程实例（创建子进程）

is_alive ()：判断子进程是否还在活着

join ([timeout])：是否等待子进程执行结束，或等待多少秒

terminate ()：不管任务是否完成，立即终止子进程用

import time

import multiprocessing

def fun_process (t):

    while True:

        print ('-process-')

        time.sleep (t)

if __name__ == '__main__':

    p = multiprocessing.Process (target=fun_process,args=(1,))

    p.start ()

    # 主进程单独执行的代码

    while True:

        print ("-main-")

        time.sleep (1)

import queue

# 创建一个 FIFO 队列

fifo_queue = queue.Queue ()

# 创建一个 LIFO 队列

lifo_queue = queue.LifoQueue ()

# 创建一个优先队列

priority_queue = queue.PriorityQueue ()

import queue

# 创建一个 FIFO 队列

fifo_queue = queue.Queue (maxsize=3)  # 最大长度为 3

# 添加元素

fifo_queue.put (1)

fifo_queue.put (2)

fifo_queue.put (3)

print (fifo_queue.qsize ())  # 输出：3

# 取出元素

print (fifo_queue.get ())  # 输出：1

print (fifo_queue.get ())  # 输出：2

print (fifo_queue.empty ())  # 输出：False

# 添加一个新元素

fifo_queue.put (4)

print (fifo_queue.get ())  # 输出：3

print (fifo_queue.get ())  # 输出：4

print (fifo_queue.empty ())  # 输出：True

import queue

# 创建一个 LIFO 队列

lifo_queue = queue.LifoQueue ()

# 添加元素

lifo_queue.put (1)

lifo_queue.put (2)

lifo_queue.put (3)

# 取出元素

print (lifo_queue.get ())  # 输出：3

print (lifo_queue.get ())  # 输出：2

import queue

# 创建一个优先队列

priority_queue = queue.PriorityQueue ()

# 添加元素，元素是元组 (优先级，值)

priority_queue.put ((2, 'task2'))

priority_queue.put ((1, 'task1'))

priority_queue.put ((3, 'task3'))

# 取出元素

print (priority_queue.get ())  # 输出：(1, 'task1')

print (priority_queue.get ())  # 输出： (2, 'task2')