@messense 感谢您的耐心回复！通过你的回复，我对 asyncio 的理解进了一步，前两天看了差不多一天的文档，都还是有点懵。现在还有两个问题，我还是不是很懂，希望您能再帮忙解一下惑。

`
async def select(pool, sql, args=(), size=None):
async with pool.acquire() as conn:
async with conn.cursor() as cur:
await cur.execute(sql.replace('?', '%s'), args)
if size:
r = await cur.fetchmany(size)
else:
r = await cur.fetchall()
return r
`

我在做关于这段代码的单元测试的时候，虽然执行正确，但是有一个警告，` ResourceWarning: Unclosed connection <aiomysql.connection.Connection object at 0x1030d1710>
ResourceWarning)`

大概意思就是数据库连接没有关闭吧。但是这里用了上下文管理器啊，它不应该会关闭连接吗？我如果在 return 语句之前，加上 conn.close()， 就不会报这个警告了。

`
def test_select(self):
sql = 'select * from minifw where id = (%s)'
rs = self.loop.run_until_complete(base_db.select(self.pool, sql, args=(1,), size=1))
self.assertEqual(len(rs), 1)
`
这段代码是测试 select。


另外还有一个问题，我想问问
self.loop = asyncio.new_event_loop()
asyncio.set_event_loop(self.loop)

和
self.loop = asyncio.get_event_loop()
的区别，因为在我执行 unnittest.main()的时候，用前者就可以执行，用后者就会报错，`Event loop is closed`，如果是测试单个数据库操作，后者就不会报错。

期望您的回复，感恩

@messense 大神，是否还有别的方法，可以把 pool 从 main() 中抽出来，这样写我又遇到问题了，比如我要写单元测试用例，应该怎么写啊，我写的运行使用不行

<pre>
import asyncio
import unittest
import aiomysql
from minifw.db import base_db


class TestDB(unittest.TestCase):
def setUp(self):
self.loop = asyncio.new_event_loop()
asyncio.set_event_loop(self.loop)
self.pool = aiomysql.create_pool(host='127.0.0.1', port=3306,
user='root', password='123456',
db='test', loop=self.loop)

def test_select(self):
sql = 'select * from minifw where id = (%s)'
rs = self.loop.run_until_complete(base_db.select(self.pool, sql, args=(1,), size=1))
self.assertEqual(len(rs), 1)

def test_insert(self):
sql = 'insert into `minifw` (`name`) values (?)'
rs = self.loop.run_until_complete(base_db.insert(self.pool, sql, args=('test_val',)))
self.assertEqual(rs, 1)

def tearDown(self):
self.loop.close()
del self.loop

if __name__ == '__main__':
unittest.main()
</pre>

这里的问题就是这个 pool 我不知道咋处理，如果我要运行 unnittest.main()，就会报错 `AttributeError: '_PoolContextManager' object has no attribute 'acquire'`

@messense 感谢，这样确实是可以的。我咋就没想到。

@qs 能否改一下给我看看，我用全局变量试过，应该是我能力不够吧，改过后调试不通。感谢

@sobigfish serverless 没怎么了解，如果可以轻松切换弹性公网 ip，那么我觉得是非常棒的

@gogobody 感谢支持

@xiubin 你说的前三个功能我都实现了啊，你看看项目介绍就明显可以得知啊。我说的给定微博，有两个来源，一个是搜索关键词获取，一个是通过给定用户来抓。不过有一点需要明确，不可能获取某个用户的所有关注和粉丝，只能获取前五页。确实你说的这个并不是项目的初衷，但也可以依托它较为轻松的实现

@xiubin 我是实现的针对给定微博，抓取它的所有评论。如果是用户的评论和赞，我没去实现。并且，好像在微博系统中，我们也看不到某个用户赞了什么和评论了什么啊

@haofly 哈哈哈，识货

@CosmoX 谢谢支持

@haofly 快献出你宝贵的 star

@haofly 靠，哪里都有你

@goodluck 咋感觉你这个没有登录啊？难道说不用登录也能抓？用的啥方法？

@goodluck 感谢支持。回踩一下

@soulmine 补充一下，单是做爬虫，机器性能要求相对来说要低一些

感觉 v 站的兄弟太给力了。多谢大家捧场！

@jtsai 你这个问题我觉得应该视具体情况而定。如果是抓文本内容的话，网速应该要不了太多，10M 独享肯定是够了，可能几兆都够了。如果抓视频或者图片等消耗 IO 的资源，那肯定是要网速越快越好咯。其实，很多时候，抓的东西都会有限制，这个时候，你会发现账号或者 ip 才是瓶颈了（对于个人开发者或者小公司而言）。

爬虫很多时候主要就是占用 IO,cpu 和内存其实依赖比较小。一台机器的话，如果不考虑代理，很容易被封 IP。考虑代理的话，代理的可用性也需要考量。额，好像偏题了...你自己多尝试吧，爬虫比较依赖外部环境，反正需要视不同情况而定

@zthxxx 谢谢捧场

@julyclyde 此话怎讲啊？我前段时间运行一个 celery 写的分布式爬虫，结果我的 redis 服务器挂了，所有抓取节点就相当于挂掉了。请问有啥方法能提高它的可用性吗？