찾다

집 >백엔드 개발 >파이썬 튜토리얼 >Python Asyncio를 사용하여 웹사이트 상태 확인을 구현하는 방법

Python Asyncio를 사용하여 웹사이트 상태 확인을 구현하는 방법

PHPz
PHPz앞으로
2023-04-21 14:10:18833검색

asyncio를 사용하면 스트림을 열고 HTTP 요청 및 응답을 쓰고 읽어 웹사이트의 HTTP 상태를 쿼리할 수 있습니다.

그런 다음 asyncio를 사용하여 여러 웹사이트의 상태를 동시에 쿼리하고 결과를 동적으로 보고할 수도 있습니다.

1. Asyncio를 사용하여 HTTP 상태를 확인하는 방법

asyncio 모듈은 소켓 연결 열기 및 스트림을 통해 데이터 읽기 및 쓰기를 지원합니다. 이 기능을 사용하여 웹페이지의 상태를 확인할 수 있습니다.

여기에는 네 가지 단계가 포함될 수 있습니다.

  • 연결 열기

  • 요청 쓰기

  • 응답 읽기

  • 연결 닫기

2. HTTP 연결 열기

asyncio.open_connection() 함수를 사용하여 asyncio에서 연결을 열 수 있습니다. 많은 매개변수 중에서 이 함수는 문자열 호스트 이름과 정수 포트 번호를 사용합니다.

이것은 소켓을 사용하여 읽고 쓰기 위해 StreamReader 및 StreamWriter를 반환하는 반드시 기다려야 하는 코루틴입니다.

이것은 포트 80에서 HTTP 연결을 여는 데 사용될 수 있습니다. 

...
# open a socket connection
reader, writer = await asyncio.open_connection('www.google.com', 80)

ssl=True 매개변수를 사용하여 SSL 연결을 열 수도 있습니다. 이는 포트 443에서 HTTPS 연결을 여는 데 사용될 수 있습니다. 

...
# open a socket connection
reader, writer = await asyncio.open_connection('www.google.com', 443)

3. HTTP 요청 쓰기

열고 나면 StreamWriter에 쿼리를 작성하여 HTTP 요청을 할 수 있습니다. 예를 들어 HTTP 버전 1.1 요청은 일반 텍스트로 되어 있습니다. 다음과 같은 파일 경로 "/"를 요청할 수 있습니다: 

GET / HTTP/1.1
Host: www.google.com

중요한 점은 각 줄의 끝에는 캐리지 리턴과 개행 문자(rn)가 있어야 하고 끝에는 빈 줄이 있어야 한다는 것입니다.

Python 문자열로 보면 다음과 같습니다. 

'GET / HTTP/1.1\r\n'
'Host: www.google.com\r\n'
'\r\n'

이 문자열은 StreamWriter에 쓰기 전에 바이트로 인코딩되어야 합니다. 이는 문자열 자체에 대해 encode() 메소드를 사용하여 달성할 수 있습니다. 기본 "utf-8" 인코딩이면 충분할 수 있습니다. 

...
# encode string as bytes
byte_data = string.encode()

그런 다음 StreamWriter의 write() 메서드를 통해 바이트를 소켓에 쓸 수 있습니다. 

...
# write query to socket
writer.write(byte_data)

요청을 작성한 후 데이터 바이트가 전송되고 소켓이 준비될 때까지 기다리는 것이 가장 좋습니다. 이는 Drain() 메서드를 통해 달성할 수 있습니다. 기다려야 하는 코루틴입니다. 

...
# wait for the socket to be ready.
await writer.drain()

4. HTTP 응답 읽기

HTTP 요청을 한 후 응답을 읽을 수 있습니다. 이는 소켓의 StreamReader를 통해 달성할 수 있습니다. 응답은 큰 바이트 블록을 읽는 read() 메서드 또는 바이트 행을 읽는 readline() 메서드를 사용하여 읽을 수 있습니다.

우리는 HTML 데이터를 한 번에 한 줄씩 보내는 텍스트 기반 HTTP 프로토콜을 사용하기 때문에 readline() 메서드를 선호할 수 있습니다. readline() 메서드는 코루틴이므로 기다려야 합니다. 

...
# read one line of response
line_bytes = await reader.readline()

HTTP 1.1 응답은 빈 줄로 구분된 헤더와 빈 줄로 끝나는 본문의 두 부분으로 구성됩니다. 헤더에는 요청 성공 여부와 어떤 형식의 파일을 보낼 것인지에 대한 정보가 포함되고, 본문에는 HTML 웹페이지와 같은 파일 내용이 포함됩니다.

HTTP 헤더의 첫 번째 줄에는 서버에서 요청한 페이지의 HTTP 상태가 포함됩니다. 각 줄은 바이트에서 문자열로 디코딩되어야 합니다.

이것은 바이트 데이터에 대해 decode() 메소드를 사용하여 달성할 수 있습니다. 다시 말하지만, 기본 인코딩은 "utf_8"입니다. 

...
# decode bytes into a string
line_data = line_bytes.decode()

5. HTTP 연결 닫기

StreamWriter를 닫으면 소켓 연결을 닫을 수 있습니다. 이는 close() 메소드를 호출하여 달성할 수 있습니다. 

...
# close the connection
writer.close()

이것은 차단되지 않으며 소켓을 즉시 닫지 않을 수도 있습니다. 이제 asyncio를 사용하여 HTTP 요청을 만들고 응답을 읽는 방법을 알았으므로 웹 페이지 상태를 확인하는 몇 가지 예를 살펴보겠습니다.

6. HTTP 상태를 순차적으로 확인하는 예제

asyncio를 사용하여 여러 웹사이트의 HTTP 상태를 확인하는 예제를 개발할 수 있습니다.

이 예에서는 먼저 주어진 URL의 상태를 확인하는 코루틴을 개발하겠습니다. 그런 다음 상위 10개 웹사이트 각각에 대해 이 코루틴을 한 번씩 호출합니다.

먼저, URL 문자열을 받아들이고 HTTP 상태를 반환하는 코루틴을 정의할 수 있습니다. 

# get the HTTP/S status of a webpage
async def get_status(url):
	# ...

URL을 구성 요소 부분으로 구문 분석해야 합니다. HTTP 요청을 할 때 호스트 이름과 파일 경로가 필요합니다. 또한 SSL이 필요한지 확인하려면 URL 체계(HTTP 또는 HTTPS)를 알아야 합니다.

이는 URL 문자열을 받아들이고 모든 URL 요소의 명명된 튜플을 반환하는 urllib.parse.urlsplit() 함수를 사용하여 달성할 수 있습니다. 

...
# split the url into components
url_parsed = urlsplit(url)

그런 다음 URL 구성표를 기반으로 HTTP 연결을 열고 URL 호스트 이름을 사용할 수 있습니다. 

...
# open the connection
if url_parsed.scheme == 'https':
    reader, writer = await asyncio.open_connection(url_parsed.hostname, 443, ssl=True)
else:
    reader, writer = await asyncio.open_connection(url_parsed.hostname, 80)

다음으로 호스트 이름과 파일 경로를 사용하여 HTTP GET 요청을 생성하고 StreamWriter를 사용하여 인코딩된 바이트를 소켓에 쓸 수 있습니다. 

...
# send GET request
query = f'GET {url_parsed.path} HTTP/1.1\r\nHost: {url_parsed.hostname}\r\n\r\n'
# write query to socket
writer.write(query.encode())
# wait for the bytes to be written to the socket
await writer.drain()

다음으로 HTTP 응답을 읽을 수 있습니다. HTTP 상태가 포함된 응답의 첫 번째 줄만 필요합니다. 

...
# read the single line response
response = await reader.readline()

그러면 연결이 종료될 수 있습니다. 

...
# close the connection
writer.close()

마지막으로 서버에서 읽은 바이트, 원격 후행 공백을 디코딩하고 HTTP 상태를 반환할 수 있습니다. 

...
# decode and strip white space
status = response.decode().strip()
# return the response
return status

이를 모두 종합하면 완전한 get_status() 코루틴이 아래에 나열됩니다. 호스트에 액세스할 수 없거나 응답이 느린 등의 오류 처리 기능이 없습니다. 이러한 추가 사항은 독자에게 멋진 확장 기능을 제공할 것입니다. 

# get the HTTP/S status of a webpage
async def get_status(url):
    # split the url into components
    url_parsed = urlsplit(url)
    # open the connection
    if url_parsed.scheme == 'https':
        reader, writer = await asyncio.open_connection(url_parsed.hostname, 443, ssl=True)
    else:
        reader, writer = await asyncio.open_connection(url_parsed.hostname, 80)
    # send GET request
    query = f'GET {url_parsed.path} HTTP/1.1\r\nHost: {url_parsed.hostname}\r\n\r\n'
    # write query to socket
    writer.write(query.encode())
    # wait for the bytes to be written to the socket
    await writer.drain()
    # read the single line response
    response = await reader.readline()
    # close the connection
    writer.close()
    # decode and strip white space
    status = response.decode().strip()
    # return the response
    return status

다음으로 확인하려는 여러 웹페이지 또는 웹사이트에 대해 get_status() 코루틴을 호출할 수 있습니다. 이 경우 전 세계 상위 10개 웹페이지 목록을 정의하겠습니다. 

...
# list of top 10 websites to check
sites = ['https://www.google.com/',
    'https://www.youtube.com/',
    'https://www.facebook.com/',
    'https://twitter.com/',
    'https://www.instagram.com/',
    'https://www.baidu.com/',
    'https://www.wikipedia.org/',
    'https://yandex.ru/',
    'https://yahoo.com/',
    'https://www.whatsapp.com/'
    ]

然后我们可以使用我们的 get_status() 协程依次查询每个。在这种情况下，我们将在一个循环中按顺序这样做，并依次报告每个状态。

...
# check the status of all websites
for url in sites:
    # get the status for the url
    status = await get_status(url)
    # report the url and its status
    print(f'{url:30}:\t{status}')

在使用 asyncio 时，我们可以做得比顺序更好，但这提供了一个很好的起点，我们可以在以后进行改进。将它们结合在一起，main() 协程查询前 10 个网站的状态。

# main coroutine
async def main():
    # list of top 10 websites to check
    sites = ['https://www.google.com/',
        'https://www.youtube.com/',
        'https://www.facebook.com/',
        'https://twitter.com/',
        'https://www.instagram.com/',
        'https://www.baidu.com/',
        'https://www.wikipedia.org/',
        'https://yandex.ru/',
        'https://yahoo.com/',
        'https://www.whatsapp.com/'
        ]
    # check the status of all websites
    for url in sites:
        # get the status for the url
        status = await get_status(url)
        # report the url and its status
        print(f'{url:30}:\t{status}')

最后，我们可以创建 main() 协程并将其用作 asyncio 程序的入口点。

...
# run the asyncio program
asyncio.run(main())

将它们结合在一起，下面列出了完整的示例。

# SuperFastPython.com
# check the status of many webpages
import asyncio
from urllib.parse import urlsplit
 
# get the HTTP/S status of a webpage
async def get_status(url):
    # split the url into components
    url_parsed = urlsplit(url)
    # open the connection
    if url_parsed.scheme == 'https':
        reader, writer = await asyncio.open_connection(url_parsed.hostname, 443, ssl=True)
    else:
        reader, writer = await asyncio.open_connection(url_parsed.hostname, 80)
    # send GET request
    query = f'GET {url_parsed.path} HTTP/1.1\r\nHost: {url_parsed.hostname}\r\n\r\n'
    # write query to socket
    writer.write(query.encode())
    # wait for the bytes to be written to the socket
    await writer.drain()
    # read the single line response
    response = await reader.readline()
    # close the connection
    writer.close()
    # decode and strip white space
    status = response.decode().strip()
    # return the response
    return status
 
# main coroutine
async def main():
    # list of top 10 websites to check
    sites = ['https://www.google.com/',
        'https://www.youtube.com/',
        'https://www.facebook.com/',
        'https://twitter.com/',
        'https://www.instagram.com/',
        'https://www.baidu.com/',
        'https://www.wikipedia.org/',
        'https://yandex.ru/',
        'https://yahoo.com/',
        'https://www.whatsapp.com/'
        ]
    # check the status of all websites
    for url in sites:
        # get the status for the url
        status = await get_status(url)
        # report the url and its status
        print(f'{url:30}:\t{status}')
 
# run the asyncio program
asyncio.run(main())

运行示例首先创建 main() 协程并将其用作程序的入口点。main() 协程运行，定义前 10 个网站的列表。然后顺序遍历网站列表。 main()协程挂起调用get_status()协程查询一个网站的状态。

get_status() 协程运行、解析 URL 并打开连接。它构造一个 HTTP GET 查询并将其写入主机。读取、解码并返回响应。main() 协程恢复并报告 URL 的 HTTP 状态。

对列表中的每个 URL 重复此操作。该程序大约需要 5.6 秒才能完成，或者平均每个 URL 大约需要半秒。这突出了我们如何使用 asyncio 来查询网页的 HTTP 状态。

尽管如此，它并没有充分利用 asyncio 来并发执行任务。

https://www.google.com/       :    HTTP/1.1 200 OK
https://www.youtube.com/      :    HTTP/1.1 200 OK
https://www.facebook.com/     :    HTTP/1.1 302 Found
https://twitter.com/          :    HTTP/1.1 200 OK
https://www.instagram.com/    :    HTTP/1.1 200 OK
https://www.baidu.com/        :    HTTP/1.1 200 OK
https://www.wikipedia.org/    :    HTTP/1.1 200 OK
https://yandex.ru/            :    HTTP/1.1 302 Moved temporarily
https://yahoo.com/            :    HTTP/1.1 301 Moved Permanently
https://www.whatsapp.com/     :    HTTP/1.1 302 Found

7. 并发查看网站状态示例

asyncio 的一个好处是我们可以同时执行许多协程。我们可以使用 asyncio.gather() 函数在 asyncio 中并发查询网站的状态。

此函数采用一个或多个协程，暂停执行提供的协程，并将每个协程的结果作为可迭代对象返回。然后我们可以遍历 URL 列表和可迭代的协程返回值并报告结果。

这可能是比上述方法更简单的方法。首先，我们可以创建一个协程列表。

...
# create all coroutine requests
coros = [get_status(url) for url in sites]

接下来，我们可以执行协程并使用 asyncio.gather() 获取可迭代的结果。

请注意，我们不能直接提供协程列表，而是必须将列表解压缩为单独的表达式，这些表达式作为位置参数提供给函数。

...
# execute all coroutines and wait
results = await asyncio.gather(*coros)

这将同时执行所有协程并检索它们的结果。然后我们可以遍历 URL 列表和返回状态并依次报告每个。

...
# process all results
for url, status in zip(sites, results):
    # report status
    print(f'{url:30}:\t{status}')

将它们结合在一起，下面列出了完整的示例。

# SuperFastPython.com
# check the status of many webpages
import asyncio
from urllib.parse import urlsplit
 
# get the HTTP/S status of a webpage
async def get_status(url):
    # split the url into components
    url_parsed = urlsplit(url)
    # open the connection
    if url_parsed.scheme == 'https':
        reader, writer = await asyncio.open_connection(url_parsed.hostname, 443, ssl=True)
    else:
        reader, writer = await asyncio.open_connection(url_parsed.hostname, 80)
    # send GET request
    query = f'GET {url_parsed.path} HTTP/1.1\r\nHost: {url_parsed.hostname}\r\n\r\n'
    # write query to socket
    writer.write(query.encode())
    # wait for the bytes to be written to the socket
    await writer.drain()
    # read the single line response
    response = await reader.readline()
    # close the connection
    writer.close()
    # decode and strip white space
    status = response.decode().strip()
    # return the response
    return status
 
# main coroutine
async def main():
    # list of top 10 websites to check
    sites = ['https://www.google.com/',
        'https://www.youtube.com/',
        'https://www.facebook.com/',
        'https://twitter.com/',
        'https://www.instagram.com/',
        'https://www.baidu.com/',
        'https://www.wikipedia.org/',
        'https://yandex.ru/',
        'https://yahoo.com/',
        'https://www.whatsapp.com/'
        ]
    # create all coroutine requests
    coros = [get_status(url) for url in sites]
    # execute all coroutines and wait
    results = await asyncio.gather(*coros)
    # process all results
    for url, status in zip(sites, results):
        # report status
        print(f'{url:30}:\t{status}')
 
# run the asyncio program
asyncio.run(main())

运行该示例会像以前一样执行 main() 协程。在这种情况下，协程列表是在列表理解中创建的。

然后调用 asyncio.gather() 函数，传递协程并挂起 main() 协程，直到它们全部完成。协程执行，同时查询每个网站并返回它们的状态。

main() 协程恢复并接收可迭代的状态值。然后使用 zip() 内置函数遍历此可迭代对象和 URL 列表，并报告状态。

这突出了一种更简单的方法来同时执行协程并在所有任务完成后报告结果。它也比上面的顺序版本更快，在我的系统上完成大约 1.4 秒。

https://www.google.com/       :    HTTP/1.1 200 OK
https://www.youtube.com/      :    HTTP/1.1 200 OK
https://www.facebook.com/     :    HTTP/1.1 302 Found
https://twitter.com/          :    HTTP/1.1 200 OK
https://www.instagram.com/    :    HTTP/1.1 200 OK
https://www.baidu.com/        :    HTTP/1.1 200 OK
https://www.wikipedia.org/    :    HTTP/1.1 200 OK
https://yandex.ru/            :    HTTP/1.1 302 Moved temporarily
https://yahoo.com/            :    HTTP/1.1 301 Moved Permanently
https://www.whatsapp.com/     :    HTTP/1.1 302 Found

위 내용은 Python Asyncio를 사용하여 웹사이트 상태 확인을 구현하는 방법의 상세 내용입니다. 자세한 내용은 PHP 중국어 웹사이트의 기타 관련 기사를 참조하세요!

Python html 字符串 循环 并发 对象 http https ssl
성명：
이 기사는 yisu.com에서 복제됩니다. 침해가 있는 경우 admin@php.cn으로 문의하시기 바랍니다. 삭제
이전 기사：Python을 사용하여 지정된 이미지를 일괄적으로 삭제하거나 이동하는 방법은 무엇입니까?다음 기사：Python을 사용하여 지정된 이미지를 일괄적으로 삭제하거나 이동하는 방법은 무엇입니까?

관련 기사

더보기