python多进程检查端口并写日志

本文涉及的产品
日志服务 SLS,月写入数据量 50GB 1个月
简介: 版权声明:本文可能为博主原创文章,若标明出处可随便转载。 https://blog.
版权声明:本文可能为博主原创文章,若标明出处可随便转载。 https://blog.csdn.net/Jailman/article/details/77573916
#!C:\Python27\python.exe
#coding: utf-8

from multiprocessing import Pool
import socket
socket.setdefaulttimeout(10)


#################################################################
import ctypes

STD_INPUT_HANDLE = -10
STD_OUTPUT_HANDLE = -11
STD_ERROR_HANDLE = -12

FOREGROUND_BLACK = 0x0
FOREGROUND_BLUE = 0x01  # text color contains blue.
FOREGROUND_GREEN = 0x02  # text color contains green.
FOREGROUND_RED = 0x04  # text color contains red.
FOREGROUND_INTENSITY = 0x08  # text color is intensified.

BACKGROUND_BLUE = 0x10  # background color contains blue.
BACKGROUND_GREEN = 0x20  # background color contains green.
BACKGROUND_RED = 0x40  # background color contains red.
BACKGROUND_INTENSITY = 0x80  # background color is intensified.


class Color:
    ''''' See http://msdn.microsoft.com/library/default.asp?url=/library/en-us/winprog/winprog/windows_api_reference.asp
    for information on Windows APIs.'''
    std_out_handle = ctypes.windll.kernel32.GetStdHandle(STD_OUTPUT_HANDLE)

    def set_cmd_color(self, color, handle=std_out_handle):
        """(color) -> bit
        Example: set_cmd_color(FOREGROUND_RED | FOREGROUND_GREEN | FOREGROUND_BLUE | FOREGROUND_INTENSITY)
        """
        bool = ctypes.windll.kernel32.SetConsoleTextAttribute(handle, color)
        return bool

    def reset_color(self):
        self.set_cmd_color(FOREGROUND_RED | FOREGROUND_GREEN | FOREGROUND_BLUE)

    def print_red_text(self, print_text):
        self.set_cmd_color(FOREGROUND_RED | FOREGROUND_INTENSITY)
        print print_text
        self.reset_color()

    def print_green_text(self, print_text):
        self.set_cmd_color(FOREGROUND_GREEN | FOREGROUND_INTENSITY)
        print print_text
        self.reset_color()

    def print_blue_text(self, print_text):
        self.set_cmd_color(FOREGROUND_BLUE | FOREGROUND_INTENSITY)
        print print_text
        self.reset_color()

    def print_red_text_with_blue_bg(self, print_text):
        self.set_cmd_color(FOREGROUND_RED | FOREGROUND_INTENSITY | BACKGROUND_BLUE | BACKGROUND_INTENSITY)
        print print_text
        self.reset_color()

clr = Color()
# clr.print_red_text('red')
# clr.print_green_text('green')
# clr.print_blue_text('blue')
# clr.print_red_text_with_blue_bg('background')
#################################################################

def check_server(address, port):
    s = socket.socket()
    # print "Attempting to connect to %s on port %s" % (address, port)
    try:
        port = int(port)
        s.connect((address, port))
        # print "Connected to %s on port %s" % (address, port)
        return True
    except socket.error, e:
        clr.print_red_text("Connection to %s on port %s failed: %s" % (address, port, e))
        return False


def single_func(i):
    with open('proxy.txt') as f:
        lines = f.readlines()
        if lines[i] != '':
            ip = lines[i].split(':')[0].strip()
            port = lines[i].split(':')[1].strip()
            if check_server(ip, port):
                msg = "Connected to %s on port %s" % (ip, port)
                clr.print_green_text(msg)
                return msg
            else:
                return "Connection to %s on port %s failed" % (ip, port)
        else:
            msg = "Empty line!"
            return msg
    # f.close()


def logger(msg):
    with open('log.txt', 'a+') as f:
        f.write(str(msg) + "\n")
        f.flush()


if __name__ == '__main__':
    try:
        pool = Pool(processes=4)
        with open('proxy.txt') as f:
            lines = f.readlines()
            for i in range(1, lines.__len__() + 1):
                result = pool.apply_async(single_func, (i,), callback=logger).get(99999)
            pool.close()
            pool.join()
            # f.close()
    except KeyboardInterrupt as e:
        clr.print_red_text("Interrupt Reason: %s" % str(e))




相关实践学习
日志服务之使用Nginx模式采集日志
本文介绍如何通过日志服务控制台创建Nginx模式的Logtail配置快速采集Nginx日志并进行多维度分析。
目录
相关文章
|
2月前
|
存储 Prometheus 监控
Docker容器内进行应用调试与故障排除的方法与技巧,包括使用日志、进入容器检查、利用监控工具及检查配置等,旨在帮助用户有效应对应用部署中的挑战,确保应用稳定运行
本文深入探讨了在Docker容器内进行应用调试与故障排除的方法与技巧,包括使用日志、进入容器检查、利用监控工具及检查配置等,旨在帮助用户有效应对应用部署中的挑战,确保应用稳定运行。
82 5
|
3月前
|
监控 数据挖掘 数据安全/隐私保护
Python脚本:自动化下载视频的日志记录
Python脚本:自动化下载视频的日志记录
|
3月前
|
并行计算 数据处理 调度
Python中的并发编程:探索多线程与多进程的奥秘####
本文深入探讨了Python中并发编程的两种主要方式——多线程与多进程,通过对比分析它们的工作原理、适用场景及性能差异,揭示了在不同应用需求下如何合理选择并发模型。文章首先简述了并发编程的基本概念,随后详细阐述了Python中多线程与多进程的实现机制,包括GIL(全局解释器锁)对多线程的影响以及多进程的独立内存空间特性。最后,通过实例演示了如何在Python项目中有效利用多线程和多进程提升程序性能。 ####
|
3月前
|
调度 iOS开发 MacOS
python多进程一文够了!!!
本文介绍了高效编程中的多任务原理及其在Python中的实现。主要内容包括多任务的概念、单核和多核CPU的多任务实现、并发与并行的区别、多任务的实现方式(多进程、多线程、协程等)。详细讲解了进程的概念、使用方法、全局变量在多个子进程中的共享问题、启动大量子进程的方法、进程间通信(队列、字典、列表共享)、生产者消费者模型的实现,以及一个实际案例——抓取斗图网站的图片。通过这些内容,读者可以深入理解多任务编程的原理和实践技巧。
195 1
|
4月前
|
Python
python读写操作excel日志
主要是读写操作,创建表格
83 2
|
4月前
|
Python
Python中的多线程与多进程
本文将探讨Python中多线程和多进程的基本概念、使用场景以及实现方式。通过对比分析,我们将了解何时使用多线程或多进程更为合适,并提供一些实用的代码示例来帮助读者更好地理解这两种并发编程技术。
|
4月前
|
Python Windows
python知识点100篇系列(24)- 简单强大的日志记录器loguru
【10月更文挑战第11天】Loguru 是一个功能强大的日志记录库,支持日志滚动、压缩、定时删除、高亮和告警等功能。安装简单,使用方便,可通过 `pip install loguru` 快速安装。支持将日志输出到终端或文件,并提供丰富的配置选项,如按时间或大小滚动日志、压缩日志文件等。还支持与邮件通知模块结合,实现邮件告警功能。
python知识点100篇系列(24)- 简单强大的日志记录器loguru
|
3月前
|
监控 JavaScript 前端开发
python中的线程和进程(一文带你了解)
欢迎来到瑞雨溪的博客,这里是一位热爱JavaScript和Vue的大一学生分享技术心得的地方。如果你从我的文章中有所收获,欢迎关注我,我将持续更新更多优质内容,你的支持是我前进的动力!🎉🎉🎉
46 0
|
4月前
|
运维 安全 网络协议
Python 网络编程:端口检测与IP解析
本文介绍了使用Python进行网络编程的两个重要技能:检查端口状态和根据IP地址解析主机名。通过`socket`库实现端口扫描和主机名解析的功能,并提供了详细的示例代码。文章最后还展示了如何整合这两部分代码,实现一个简单的命令行端口扫描器,适用于网络故障排查和安全审计。
79 0
|
7月前
|
运维 关系型数据库 MySQL
掌握taskset:优化你的Linux进程,提升系统性能
在多核处理器成为现代计算标准的今天,运维人员和性能调优人员面临着如何有效利用这些处理能力的挑战。优化进程运行的位置不仅可以提高性能,还能更好地管理和分配系统资源。 其中,taskset命令是一个强大的工具,它允许管理员将进程绑定到特定的CPU核心,减少上下文切换的开销,从而提升整体效率。
掌握taskset:优化你的Linux进程,提升系统性能