异步io aiohttp爬虫

Author: budaLi

python中多线程，多进程，进程池/aiohttp_demo.py

@@ -0,0 +1,46 @@
+# @Time    : 2019/10/8 18:21
+# @Author  : Libuda
+# @FileName: aiohttp_demo.py
+# @Software: PyCharm
+# asyncio 没有提供http协议的接口 aiohttp
+import asyncio
+import socket
+from urllib.parse import urlparse
+
+
+async def get_url(url):
+    # 通过socket请求html
+    url = urlparse(url)
+    host = url.netloc
+    path = url.path
+    if path == "":
+        path = "/"
+
+    # 建立socket连接
+    reader, writer = await asyncio.open_connection(host, 80)
+    writer.write("GET {} HTTP/1.1\r\nHost:{}\r\nConnection:close\r\n\r\n".format(path, host).encode("utf8"))
+    all_lines = []
+    async for raw_line in reader:
+        data = raw_line.decode("utf8")
+        all_lines.append(data)
+    html = "\n".join(all_lines)
+    return html
+
+
+async def main():
+    tasks = []
+    for url in range(20):
+        url = "http://shop.projectsedu.com/goods/{}/".format(url)
+        tasks.append(asyncio.ensure_future(get_url(url)))
+    for task in asyncio.as_completed(tasks):
+        result = await task
+        print(result)
+
+
+if __name__ == "__main__":
+    import time
+
+    start_time = time.time()
+    loop = asyncio.get_event_loop()
+    loop.run_until_complete(main())
+    print('last time:{}'.format(time.time() - start_time))

python中多线程，多进程，进程池/aiohttp_spider.py

@@ -12,7 +12,7 @@
 
 class Spider:
     def __init__(self):
-        self.start_url = ''
+        self.start_url = 'https://yq.aliyun.com/articles/'
         self.pool = ""
         self.stop = False
         self.waitting_url = Queue()
@@ -24,10 +24,10 @@ def extract_urls(self, html):
         :return:
         """
         urls = []
-        pq = PyQuery()
+        pq = PyQuery(html)
         for link in pq.items("a"):
             url = link.attr("href")
-            if url and url.startwith("http") and url not in self.seen_urls:
+            if url and url.startwith("https") and url not in self.seen_urls:
                 urls.append(url)
                 self.waitting_url.put(url)
         return urls
@@ -42,35 +42,28 @@ async def article_handler(self, url, session, pool):
         html = await self.fetch(url, session)
         self.seen_urls.add(url)
         self.extract_urls(html)
-        pq = PyQuery()
-        title = pq("title").text()
+        pq = PyQuery(html)
+        title = pq("#blog-title").text()
         async with pool.acquire() as conn:
             async with conn.cursor() as cur:
                 insert_sql = "insert into spider(title) values('{}')".format(title)
                 await cur.execute(insert_sql)
 
-    async def consumer(self, pool):
-        async with aiohttp.ClientSession() as session:
-            while not self.stop:
-                url = self.waitting_url.get()
-                if re.match("http://.*?jobble.com", url):
-                    if url not in self.seen_urls:
-                        asyncio.ensure_future(self.article_handler(url, session))
-                else:
-                    if url not in self.seen_urls:
-                        asyncio.ensure_future(self.init_urls(url, session))
+    async def consumer(self, pool, session):
+        while not self.stop:
+            url = self.waitting_url.get()
+            if re.match("https://yq.aliyun.com/articles/\d*", url):
+                if url not in self.seen_urls:
+                    asyncio.ensure_future(self.article_handler(url, session, pool))
+            else:
+                if url not in self.seen_urls:
+                    asyncio.ensure_future(self.init_urls(url, session))
 
     async def init_urls(self, url, session):
         html = await self.fetch(url, session)
         self.seen_urls.add(url)
         self.extract_urls(html=html)
 
-    async def connection(self, loop):
-        pool = await aiomysql.create_pool(host="127.0.0.1", port=3306, user='root', password='root', db='aiospider',
-                                          loop=loop, charset='utf-8', autocommit=True)
-        asyncio.ensure_future(self.init_urls(self.start_url))
-        asyncio.ensure_future(self.consumer(self.pool))
-
     async def fetch(self, url, session):
         try:
             async with session.get(url) as response:
@@ -80,7 +73,15 @@ async def fetch(self, url, session):
         except Exception as e:
             print(e)
 
+    async def main(self, loop):
+        await aiomysql.create_pool(host="127.0.0.1", port=3306, user='root', password='root', db='aiospider',
+                                   loop=loop, charset='utf8', autocommit=True)
+        async with aiohttp.ClientSession() as session:
+            asyncio.ensure_future(self.init_urls(self.start_url, session))
+            asyncio.ensure_future(self.consumer(self.pool, session))
 
 if __name__ == "__main__":
     spider = Spider()
-    spider.fetch()
+    loop = asyncio.get_event_loop()
+    asyncio.ensure_future(spider.main(loop))
+    loop.run_forever()